术语与假设
1. LLM Agent定义
LLM Agent（大型语言模型代理） 是一个能够感知环境、做出决策并执行行动的智能系统，它基于大型语言模型的推理能力，通过工具调用和外部系统交互来完成复杂任务。
graph TD
A[LLM Agent] --> B[感知模块]
A --> C[推理决策模块]
A --> D[执行模块]
A --> E[记忆模块]
B --> F[环境输入]
B --> G[用户交互]
C --> H[任务规划]
C --> I[行动选择]
D --> J[工具调用]
D --> K[外部API]
E --> L[短期记忆]
E --> M[长期记忆]2. Agent与普通LLM调用的本质区别
| 维度 | 普通LLM调用 | LLM Agent |
|---|---|---|
| 交互方式 | 单一回合对话 | 多回合、状态维护 |
| 行动能力 | 仅文本输出 | 可调用工具、修改环境 |
| 记忆机制 | 无持久记忆 | 短期+长期记忆管理 |
| 决策机制 | 被动响应 | 主动规划、反思优化 |
| 环境感知 | 仅接收输入 | 实时环境反馈 |
| 复杂度处理 | 简单任务 | 复杂多步骤任务 |
核心区别示例
普通LLM调用示例：
# 传统方式：单一函数调用
def traditional_llm_call(prompt):
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.contentLLM Agent调用示例：
class LLMAgent:
def __init__(self):
self.memory = MemoryManager()
self.tools = ToolRegistry()
self.planner = TaskPlanner()
async def execute(self, user_request):
# 1. 感知：理解用户意图
intent = await self.perceive(user_request)
# 2. 规划：制定执行计划
plan = await self.planner.create_plan(intent, self.memory.get_context())
# 3. 执行：按计划调用工具
results = []
for step in plan.steps:
result = await self.execute_step(step)
results.append(result)
# 4. 反思：评估结果并调整
if not self.evaluate(result):
plan = await self.planner.revise_plan(plan, result)
# 5. 记忆：更新状态
await self.memory.update(results)
return self.synthesize(results)3. 常见假设详解
3.1 工具调用假设（Tool Calling）
定义： Agent具备调用外部工具、API和系统接口的能力，通过工具执行具体操作而非仅生成文本。
graph LR
A[LLM Agent] --> B[工具选择]
B --> C[参数构造]
C --> D[工具执行]
D --> E[结果解析]
E --> F[状态更新]
F --> A
B --> G[数据库查询]
B --> H[API调用]
B --> I[文件系统操作]
B --> J[计算服务]技术实现框架：
class ToolCallingAgent:
def __init__(self):
self.available_tools = {
"database_query": DatabaseTool(),
"web_search": WebSearchTool(),
"calculator": CalculatorTool(),
"file_manager": FileManagerTool()
}
async def process_with_tools(self, query):
# 1. 意图识别与工具选择
tool_calls = self.extract_tool_calls(query)
# 2. 并行工具执行
tasks = []
for tool_call in tool_calls:
tool = self.available_tools[tool_call.name]
task = tool.execute(tool_call.parameters)
tasks.append(task)
# 3. 结果聚合与处理
results = await asyncio.gather(*tasks)
# 4. 响应生成
return self.generate_response(query, results)
def extract_tool_calls(self, query):
# 基于模型的能力或特定格式提取工具调用
# 可以使用Function Calling、OpenAI格式等
pass3.2 长期记忆假设（Long-term Memory）
定义： Agent能够维护和利用跨会话的持久化信息，包括用户偏好、历史交互、领域知识等。
graph TD
A[记忆输入] --> B[记忆分类]
B --> C[短期记忆
会话上下文]
B --> D[长期记忆
持久存储]
C --> E[工作记忆
当前任务]
D --> F[语义记忆
事实知识]
D --> G[程序记忆
操作习惯]
D --> H[情景记忆
历史事件]
E --> I[记忆检索]
F --> I
G --> I
H --> I
I --> J[记忆融合]
J --> K[决策支持]记忆管理架构：
class MemoryManager:
def __init__(self):
self.short_term_memory = ShortTermMemory(capacity=10) # 最近N轮对话
self.long_term_memory = VectorMemory() # 向量数据库
self.episodic_memory = EpisodicMemory() # 事件序列记忆
async def store_interaction(self, interaction):
# 1. 短期记忆存储
self.short_term_memory.add(interaction)
# 2. 提取关键信息到长期记忆
key_info = await self.extract_key_information(interaction)
# 3. 向量化存储
vector = await self.vectorize(key_info)
await self.long_term_memory.store(vector, key_info)
# 4. 更新情景记忆
await self.episodic_memory.append(interaction)
async def retrieve_relevant(self, query, top_k=5):
# 1. 查询向量化
query_vector = await self.vectorize(query)
# 2. 相似度检索
candidates = await self.long_term_memory.similarity_search(
query_vector, top_k=top_k
)
# 3. 短期记忆整合
recent_context = self.short_term_memory.get_recent()
# 4. 记忆融合与排序
return self.integrate_memories(query, candidates, recent_context)3.3 多回合交互假设（Multi-turn Interaction）
定义： Agent能够维护对话状态，进行上下文感知的连续交互，支持对话历史的管理和状态持久化。
sequenceDiagram
participant U as 用户
participant A as Agent
participant M as 记忆系统
participant T as 工具系统
U->>A: 用户输入
A->>M: 检索相关记忆
M-->>A: 历史上下文
A->>T: 制定执行计划
T-->>A: 执行结果
A->>M: 更新状态
A-->>U: 响应输出
Note over U,A: 下一轮交互
U->>A: 后续问题
A->>M: 检索完整对话历史
M-->>A: 上下文信息
A->>T: 继续执行
T-->>A: 状态更新
A-->>U: 连贯响应对话状态管理实现：
class ConversationState:
def __init__(self, session_id):
self.session_id = session_id
self.dialogue_history = []
self.current_task = None
self.user_profile = {}
self.system_state = {}
def add_turn(self, user_input, agent_output, metadata=None):
turn = DialogueTurn(
user_input=user_input,
agent_output=agent_output,
timestamp=datetime.now(),
metadata=metadata or {}
)
self.dialogue_history.append(turn)
# 更新任务状态
if metadata and 'task_status' in metadata:
self.current_task = metadata['task_status']
def get_context(self, window_size=5):
# 获取最近N轮的对话上下文
recent_turns = self.dialogue_history[-window_size:]
return {
'dialogue': recent_turns,
'task_state': self.current_task,
'user_profile': self.user_profile,
'system_state': self.system_state
}LLM Agent核心架构模式概览
在设计LLM Agent系统时，架构模式的选择直接影响系统的性能、可维护性、扩展性和成本效益。本节将深入分析四类核心架构模式：单Agent与多Agent架构、单步与多步执行架构、开放环与闭环架构，以及不同的部署模式，为架构师提供全面的选型参考。
1. 单Agent vs. 多Agent架构
1.1 单Agent架构模式
单Agent架构是最直接也是最常见的模式，所有功能集成在一个Agent中，通过工具调用完成复杂任务。
特点分析：
- 简单性：架构清晰，部署维护成本低
- 一致性：统一的知识表示和决策逻辑
- 延迟优化：减少Agent间通信开销
- 状态管理：单一状态空间，避免状态同步问题
适用场景：
- 单一领域专业任务（如代码生成、文档摘要）
- 资源受限的边缘部署场景
- 对实时性要求较高的应用
class SingleAgent:
def __init__(self, llm, tools, memory):
self.llm = llm # 大语言模型
self.tools = tools # 工具注册表
self.memory = memory # 记忆组件
self.state = {"current_task": None, "context": {}}
async def process_request(self, user_input):
# 统一的处理流程
context = await self.memory.retrieve(user_input)
plan = await self.llm.generate_plan(user_input, context)
for step in plan:
if step["type"] == "tool_call":
result = await self.call_tool(step)
await self.memory.update(result)
elif step["type"] == "llm_reasoning":
result = await self.llm.reason(step["prompt"])
return await self.generate_final_response()
async def call_tool(self, step):
tool_name = step["tool"]
args = step["args"]
return await self.tools[tool_name](**args)1.2 多Agent协作架构
多Agent架构通过专门化的Agent协作完成复杂任务，每个Agent专注于特定领域或功能。
协作模式：
- 层次化协作：Router → Specialist → Synthesizer
- 并行协作：多个Worker并行处理子任务
- 循环协作：Planner → Executor → Critic循环优化
优势与权衡：
- 专业化：每个Agent深入特定领域，能力更强
- 扩展性：新增领域只需添加专门的Agent
- 容错性：单个Agent故障不影响整体系统
- 复杂性：协调开销增加，系统调试困难
class MultiAgentOrchestrator:
def __init__(self):
self.router = RouterAgent()
self.workers = {
"code": CodeAgent(),
"analysis": AnalysisAgent(),
"search": SearchAgent(),
"synthesis": SynthesisAgent()
}
self.critic = CriticAgent()
self.memory = SharedMemory()
async def process_request(self, user_input):
# 第一步：任务分解与路由
task_plan = await self.router.decompose(user_input)
# 第二步：并行执行子任务
results = {}
for task in task_plan["subtasks"]:
agent = self.workers[task["agent_type"]]
result = await agent.execute(task, self.memory)
results[task["id"]] = result
# 第三步：批判性检查
critique = await self.critic.evaluate(task_plan, results)
if critique["needs_revision"]:
results = await self.revise_tasks(task_plan, results, critique)
# 第四步：综合结果
final_response = await self.synthesize_response(results)
return final_response
async def synthesize_response(self, results):
synthesis_prompt = f"""
基于以下并行执行的结果，生成综合回答：
{json.dumps(results, indent=2)}
"""
return await self.workers["synthesis"].execute(synthesis_prompt)架构对比表：
| 维度 | 单Agent架构 | 多Agent架构 |
|---|---|---|
| 复杂度 | 低 | 高 |
| 维护成本 | 低 | 中高 |
| 扩展性 | 中 | 高 |
| 专业化程度 | 中 | 高 |
| 容错性 | 低 | 高 |
| 开发效率 | 高 | 中 |
| 调试难度 | 低 | 高 |
2. 单步 vs. 多步/工具链架构
2.1 单步执行架构
单步架构通过一次LLM调用直接生成最终结果，适合简单、确定性的任务。
适用场景：
- 文本生成、翻译、摘要等直接输出任务
- 规则明确、无需外部数据验证的任务
- 实时性要求极高的场景
class SingleStepAgent:
def __init__(self, llm):
self.llm = llm
async def generate(self, prompt, context=""):
full_prompt = f"{context}\n\n用户输入：{prompt}"
return await self.llm.generate(full_prompt)
def get_cost_estimate(self):
return {"latency": "~500ms", "token_cost": "~1000 tokens"}2.2 多步工具链架构
多步架构将复杂任务分解为多个执行步骤，每步调用特定工具或进行推理。
执行模式：
- 线性执行链：任务按固定顺序执行
- 条件分支执行：根据中间结果决定下一步
- 并行执行：多个子任务并行处理
- 循环优化执行：通过反馈循环改进结果
class MultiStepAgent:
def __init__(self, llm, tools, memory):
self.llm = llm
self.tools = tools
self.memory = memory
self.execution_graph = self._build_execution_graph()
def _build_execution_graph(self):
return {
"step1": {"action": "plan", "next": ["step2"]},
"step2": {"action": "search", "condition": "needs_external_data", "next": ["step3", "step4"]},
"step3": {"action": "analyze", "next": ["step5"]},
"step4": {"action": "reason", "next": ["step5"]},
"step5": {"action": "synthesize", "next": ["end"]}
}
async def execute(self, user_input):
context = await self.memory.retrieve(user_input)
current_step = "step1"
results = {}
while current_step != "end":
step_config = self.execution_graph[current_step]
if step_config["action"] == "plan":
plan = await self._generate_plan(user_input, context)
results["plan"] = plan
elif step_config["action"] == "search":
query = self._extract_search_query(plan, user_input)
search_results = await self.tools["search"](query)
results["search"] = search_results
elif step_config["action"] == "analyze":
analysis = await self.llm.analyze(search_results)
results["analysis"] = analysis
elif step_config["action"] == "reason":
reasoning = await self.llm.reason(user_input, analysis)
results["reasoning"] = reasoning
elif step_config["action"] == "synthesize":
final_result = await self._synthesize(results)
return final_result
current_step = step_config["next"][0] # 简化为单分支
async def _generate_plan(self, user_input, context):
planning_prompt = f"""
基于以下任务制定详细执行计划：
任务：{user_input}
上下文：{context}
请按照以下格式输出计划：
1. 任务分解
2. 所需工具
3. 执行顺序
"""
return await self.llm.generate(planning_prompt)执行模式对比：
graph TD
A[用户输入] --> B{选择执行模式}
B -->|单步| C[直接LLM生成]
B -->|多步| D[任务分解]
D --> E[工具调用]
E --> F{需要更多步骤?}
F -->|是| G[继续执行]
F -->|否| H[生成最终输出]
H --> I[返回结果]