大语言模型（LLM）的智能体框架正在成为人工智能应用的核心架构。LLM Agent通过工具调用（Tool Calling）和推理-行动循环（ReAct模式）实现自主任务规划与执行，将语言模型的文本生成能力扩展为可操作的系统组件。本文围绕智能对话系统的开发需求，拆解Agent框架的关键设计模式与代码实现。
LLM Agent架构与核心组件
一个完整的智能体框架包含三个核心模块：推理引擎、工具注册表和记忆系统。推理引擎负责接收用户指令，拆解为可执行的子任务；工具注册表管理可调用的外部API和函数；记忆系统维护对话历史和中间状态。三者通过消息总线协作，形成”感知-推理-执行”的闭环。
与传统的对话系统不同，Agent框架的关键在于模型能够自主决定调用哪些工具、以什么参数调用、以及如何组合工具的输出来完成复杂任务。这种能力依赖于函数调用（Function Calling）机制，模型输出的不再是纯文本，而是结构化的工具调用指令。
ReAct推理模式实现原理
ReAct（Reasoning and Acting）模式是当前最主流的Agent推理框架。其核心思想是让模型交替进行推理（Thought）和行动（Action），每一步先思考当前状态和下一步策略，再执行具体动作，最后观察执行结果，循环往复直到任务完成。
ReAct的标准流程为：Thought → Action → Observation → Thought → Action → … → Final Answer。每个环节的输出都作为上下文注入下一轮推理，使模型能够基于中间结果动态调整策略。
import json
from openai import OpenAI
client = OpenAI()
# 工具定义
tools = [
{
"type": "function",
"function": {
"name": "search_database",
"description": "在知识库中检索相关文档",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "检索关键词"},
"top_k": {"type": "integer", "description": "返回结果数量", "default": 5}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "执行数学计算",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
}
}
}
]
# 工具执行器
def execute_tool(name, arguments):
if name == "search_database":
return json.dumps({"results": [
{"doc": "向量数据库选型指南", "score": 0.92},
{"doc": "Embedding模型对比", "score": 0.87}
]})
elif name == "calculate":
try:
result = eval(arguments["expression"])
return json.dumps({"result": result})
except Exception as e:
return json.dumps({"error": str(e)})
return json.dumps({"error": "unknown tool"})
def run_agent(user_query, max_steps=10):
messages = [
{"role": "system", "content": "你是一个智能助手，通过调用工具帮助用户完成任务。"},
{"role": "user", "content": user_query}
]
for step in range(max_steps):
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
msg = response.choices[0].message
messages.append(msg)
if not msg.tool_calls:
return msg.content
for tool_call in msg.tool_calls:
args = json.loads(tool_call.function.arguments)
result = execute_tool(tool_call.function.name, args)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
print(f"Step {step+1}: 调用 {tool_call.function.name}({args}) -> {result}")
return "达到最大推理步数限制"
answer = run_agent("检索向量数据库相关的技术文档，并计算检索结果的总分")
print(answer)多工具编排与任务分解策略
当任务复杂度超过单步工具调用时，需要对任务进行分解（Task Decomposition）。常见策略包括链式调用（Chain）、树状分解（Tree）和图状编排（DAG）。链式调用适合线性流程，每个工具的输出作为下一个工具的输入；树状分解将复杂任务拆为多个并行子任务，各自独立执行后汇总结果；图状编排则允许工具之间存在复杂的依赖关系。
在代码层面，任务分解通过系统提示（System Prompt）引导模型输出结构化的执行计划，再逐项执行。以LangChain为例，可以使用Plan-and-Execute模式：
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
plan_prompt = ChatPromptTemplate.from_messages([
("system", '''你是一个任务规划器。将用户请求分解为具体步骤。
每步包含: step_id, action, tool_name, parameters。
输出JSON数组格式。可用工具: search, calculate, summarize'''),
("user", "{input}")
])
llm = ChatOpenAI(model="gpt-4o")
planner = plan_prompt | llm
plan_result = planner.invoke({"input": "分析2024年大模型行业趋势"})
import json
steps = json.loads(plan_result.content)
for step in steps:
print(f"执行步骤 {step['step_id']}: {step['action']}")
result = execute_tool(step["tool_name"], step["parameters"])
step["result"] = result
summary_prompt = ChatPromptTemplate.from_messages([
("system", "根据以下步骤的执行结果，生成最终回答。"),
("user", "{results}")
])
final = (summary_prompt | llm).invoke({"results": json.dumps(steps, ensure_ascii=False)})
print(final.content)记忆系统与会话状态管理
Agent的记忆系统分为短期记忆和长期记忆。短期记忆存储当前对话的上下文，通常以消息列表形式存在内存中；长期记忆持久化历史交互数据，支持跨会话的知识检索。短期记忆的关键挑战是上下文窗口限制——当对话轮数过多时，早期信息会被截断。
滑动窗口和摘要压缩是两种主流解决方案。滑动窗口保留最近N轮对话原文，超出部分丢弃；摘要压缩则将早期对话用LLM生成摘要，压缩后保留关键信息。两者可以结合使用，最近几轮保留原文，更早的对话以摘要形式保留。
class AgentMemory:
def __init__(self, max_messages=20, summarize_threshold=15):
self.messages = []
self.max_messages = max_messages
self.summarize_threshold = summarize_threshold
self.summary = ""
def add(self, role, content):
self.messages.append({"role": role, "content": content})
if len(self.messages) > self.summarize_threshold:
self._compress()
def _compress(self):
to_summarize = self.messages[:-10]
recent = self.messages[-10:]
summary_text = summarize_with_llm(to_summarize)
self.summary = summary_text
self.messages = [{"role": "system", "content": f"历史摘要: {self.summary}"}] + recent
def get_context(self):
return self.messages
def summarize_with_llm(messages):
prompt = "将以下对话历史压缩为简洁摘要，保留关键信息：\n"
for m in messages:
prompt += f"{m['role']}: {m['content']}\n"
return "对话摘要结果"工具调用错误处理与重试机制
生产环境中工具调用可能因网络超时、参数错误或API限流而失败。健壮的Agent框架需要实现错误处理和重试机制。基本策略包括：指数退避重试、参数自动修正、降级方案和人工介入。
import time
from functools import wraps
def retry_with_backoff(max_retries=3, base_delay=1):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
return json.dumps({"error": f"工具执行失败: {str(e)}"})
delay = base_delay * (2 ** attempt)
time.sleep(delay)
return wrapper
return decorator
@retry_with_backoff(max_retries=3, base_delay=1)
def search_database(query, top_k=5):
pass
@retry_with_backoff(max_retries=2, base_delay=2)
def calculate(expression):
passAgent框架的性能优化方向
Agent的性能瓶颈通常在三个方面：工具调用的延迟、上下文窗口的利用率和模型推理速度。工具调用延迟可以通过异步并发执行多个独立工具调用来优化——当任务分解后发现多个子任务互不依赖时，使用asyncio并发执行可以将总延迟从串行叠加降为最大单次延迟。上下文窗口利用率的关键是控制注入的历史信息量，避免无关注词占用窗口。模型推理速度则取决于API选择和本地部署方案。
实际部署中，建议为Agent设置明确的超时阈值和最大推理步数，防止单次请求消耗过多资源。同时实现调用量监控和成本告警，避免工具调用产生的API费用失控。对于高频使用的Agent，可以引入结果缓存——相同或相似的查询直接返回缓存的工具调用结果，减少重复计算。
原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/llmagent-zhi-neng-ti-kuang-jia-kai-fa-gong-ju-diao-yong-yu/
