1. 智能代理架构设计模式解析
在构建现代AI系统时,我们实际上是在组合各种代理设计模式。每种模式都有其独特的阶段、构建方法和评估标准。本文将深入剖析17种高层架构模式,这些模式捕捉了代理系统可能采用的主要形式。
1.1 核心架构模式概览
让我们先快速了解几种基础架构模式:
- 多代理系统(Multi-Agent System):多个工具和代理协同解决问题
- 集成决策系统(Ensemble Decision System):多个代理各自提出答案,通过投票选出最佳方案
- 思维树(Tree-of-Thoughts):代理探索多种推理路径,选择最有前景的方向
- 反思方法(Reflexive approach):代理能够识别并承认自身知识的局限性
- ReAct循环:代理在思考、行动、再思考之间交替,不断完善其过程
这些模式各有优势,适用于不同场景。接下来我们将逐一拆解每种架构的实现细节。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 环境准备与基础设置
2.1 核心工具选择
在开始构建前,我们需要搭建开发环境。以下是行业标准工具链:
python复制import os
from typing import List, Dict, Any, Optional, Annotated, TypedDict
from dotenv import load_dotenv # 从.env文件加载环境变量
# Pydantic用于数据建模/验证
from pydantic import BaseModel, Field
# LangChain & LangGraph组件
from langchain_nebius import ChatNebius # Nebius LLM封装
from langchain_tavily import TavilySearch # Tavily搜索工具集成
from langchain_core.prompts import ChatPromptTemplate # 用于结构化提示词
from langgraph.graph import StateGraph, END # 构建状态机图
from langgraph.prebuilt import ToolNode, tools_condition # 预构建节点和条件
# 美化输出
from rich.console import Console # 控制台样式
from rich.markdown import Markdown # 在终端渲染Markdown
2.2 工具链分工
这套工具链各司其职:
- LangChain:提供核心构建块(提示词、工具定义、LLM封装)
- LangGraph:编排引擎,连接复杂流程(含循环和分支)
- LangSmith:调试器,可视化追踪代理每一步
2.3 环境变量配置
创建.env文件存放敏感信息:
env复制# Nebius LLM的API密钥
NEBIUS_API_KEY="your_nebius_api_key_here"
# LangSmith的API密钥
LANGCHAIN_API_KEY="your_langsmith_api_key_here"
# Tavily搜索工具的API密钥
TAVILY_API_KEY="your_tavily_api_key_here"
加载环境变量并验证:
python复制load_dotenv() # 从.env文件加载环境变量
# 启用LangSmith追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "Implementing 17 Agentic Architectures"
# 验证必需API密钥
for key in ["NEBIUS_API_KEY", "LANGCHAIN_API_KEY", "TAVILY_API_KEY"]:
if not os.environ.get(key):
print(f"{key} not found. Please create a .env file and set it.")
3. 反射架构(Reflection)实现
3.1 反射模式原理
反射模式让代理能够审视自身工作并改进。它特别适合输出质量至关重要的场景,如生成复杂代码或撰写技术报告。
反射流程分为三个阶段:
- 生成(Generate):创建初始草案
- 批评(Critique):分析草案找出缺陷
- 精炼(Refine):基于批评生成改进版本
3.2 数据结构定义
使用Pydantic模型确保结构化输出:
python复制class DraftCode(BaseModel):
"""初始代码草案的Schema"""
code: str = Field(description="解决用户请求的Python代码")
explanation: str = Field(description="代码工作原理的简要说明")
class Critique(BaseModel):
"""代码自评的Schema"""
has_errors: bool = Field(description="代码是否有潜在错误")
is_efficient: bool = Field(description="代码是否高效")
suggested_improvements: List[str] = Field(description="具体的改进建议")
critique_summary: str = Field(description="评审摘要")
class RefinedCode(BaseModel):
"""精炼后代码的Schema"""
refined_code: str = Field(description="最终改进的Python代码")
refinement_summary: str = Field(description="基于评审所做的更改摘要")
3.3 节点实现
生成节点:
python复制def generator_node(state):
"""生成代码初稿"""
console.print("--- 1. 生成初始草案 ---")
generator_llm = llm.with_structured_output(DraftCode)
prompt = f"""你是一名专业的Python程序员。编写一个Python函数来解决以下请求。
提供一个简单清晰的实现和解释。
请求: {state['user_request']}
"""
draft = generator_llm.invoke(prompt)
return {"draft": draft.model_dump()}
批评节点:
python复制def critic_node(state):
"""评审生成的代码"""
console.print("--- 2. 评审草案 ---")
critic_llm = llm.with_structured_output(Critique)
code_to_critique = state['draft']['code']
prompt = f"""你是一名资深代码评审专家。你的任务是对以下代码进行全面评审。
分析代码的:
1. **错误和缺陷**:是否有潜在的运行时错误、逻辑缺陷或未处理的边界情况?
2. **效率和最佳实践**:这是解决问题的最有效方式吗?是否符合Python标准规范(PEP 8)?
提供结构化的评审意见和具体的改进建议。
待评审代码:
```python
{code_to_critique}
```
"""
critique = critic_llm.invoke(prompt)
return {"critique": critique.model_dump()}
精炼节点:
python复制def refiner_node(state):
"""基于评审改进代码"""
console.print("--- 3. 精炼代码 ---")
refiner_llm = llm.with_structured_output(RefinedCode)
draft_code = state['draft']['code']
critique_suggestions = json.dumps(state['critique'], indent=2)
prompt = f"""你是一名专业Python程序员,负责根据评审意见改进代码。
你的目标是重写原始代码,实现评审中的所有改进建议。
**原始代码:**
```python
{draft_code}
```
**评审和建议:**
{critique_suggestions}
请提供最终的改进代码和所做更改的摘要。
"""
refined_code = refiner_llm.invoke(prompt)
return {"refined_code": refined_code.model_dump()}
3.4 工作流构建
定义状态图和节点连接:
python复制class ReflectionState(TypedDict):
"""反射图的状态表示"""
user_request: str
draft: Optional[dict]
critique: Optional[dict]
refined_code: Optional[dict]
# 初始化状态图
graph_builder = StateGraph(ReflectionState)
# 添加节点
graph_builder.add_node("generator", generator_node)
graph_builder.add_node("critic", critic_node)
graph_builder.add_node("refiner", refiner_node)
# 定义工作流边
graph_builder.set_entry_point("generator")
graph_builder.add_edge("generator", "critic")
graph_builder.add_edge("critic", "refiner")
graph_builder.add_edge("refiner", END)
# 编译图
reflection_app = graph_builder.compile()
3.5 测试与评估
测试斐波那契数列生成函数:
python复制user_request = "编写一个Python函数来查找第n个斐波那契数。"
initial_input = {"user_request": user_request}
# 运行工作流
final_state = None
for state_update in reflection_app.stream(initial_input, stream_mode="values"):
final_state = state_update
输出对比:
plaintext复制--- 初始草案 ---
def fibonacci(n):
if n <= 0:
return 0
elif n == 1:
return 1
else:
return fibonacci(n-1) + fibonacci(n-2)
--- 评审意见 ---
1. 函数没有正确处理负数输入
2. 时间复杂度高(O(2^n)),应考虑动态规划或记忆化
3. 不符合PEP 8规范
--- 精炼代码 ---
def fibonacci(n):
"""计算第n个斐波那契数"""
if n < 0:
raise ValueError("n必须是非负整数")
elif n == 0:
return 0
elif n == 1:
return 1
else:
fib = [0, 1]
for i in range(2, n + 1):
fib.append(fib[i-1] + fib[i-2])
return fib[n]
量化评估:
python复制class CodeEvaluation(BaseModel):
"""代码评估Schema"""
correctness_score: int = Field(description="代码逻辑正确性评分(1-10)")
efficiency_score: int = Field(description="算法效率评分(1-10)")
style_score: int = Field(description="代码风格和可读性评分(1-10)")
justification: str = Field(description="评分理由")
# 评估初始草案
initial_draft_evaluation = evaluate_code(final_state['draft']['code'])
# 评估精炼代码
refined_code_evaluation = evaluate_code(final_state['refined_code']['refined_code'])
评估结果:
plaintext复制--- 初始草案评估 ---
{
'correctness_score': 2,
'efficiency_score': 4,
'style_score': 2,
'justification': '函数时间复杂度为O(2^n)...'
}
--- 精炼代码评估 ---
{
'correctness_score': 8,
'efficiency_score': 6,
'style_score': 9,
'justification': '代码正确...时间复杂度为O(n)...'
}
4. 工具使用(Tool Use)架构
4.1 工具使用模式原理
当代理需要外部信息时,工具使用模式必不可少。它连接代理推理和真实世界数据,使代理能够查询新数据并与外部世界交互。
核心流程:
- 接收查询
- 决定是否需要工具
- 调用工具获取数据
- 综合工具结果生成最终答案
4.2 工具定义
python复制# 初始化搜索工具(限制结果数量)
search_tool = TavilySearchResults(max_results=2)
search_tool.name = "web_search"
search_tool.description = "用于搜索互联网获取最新信息的工具,包括新闻、事件和时事"
tools = [search_tool]
4.3 代理状态定义
python复制class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
4.4 代理节点实现
python复制def agent_node(state: AgentState):
"""调用LLM决定下一步行动"""
console.print("--- 代理: 思考中... ---")
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
# 工具节点(使用LangGraph预构建)
tool_node = ToolNode(tools)
4.5 路由逻辑
python复制def router_function(state: AgentState) -> str:
"""根据代理的最后消息决定下一步"""
last_message = state["messages"][-1]
if last_message.tool_calls:
console.print("--- 路由器: 决定调用工具 ---")
return "call_tool"
else:
console.print("--- 路由器: 决定结束 ---")
return "__end__"
4.6 工作流构建
python复制graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("call_tool", tool_node)
graph_builder.set_entry_point("agent")
graph_builder.add_conditional_edges("agent", router_function)
graph_builder.add_edge("call_tool", "agent")
tool_agent_app = graph_builder.compile()
4.7 测试案例
python复制user_query = "苹果最新WWDC活动的主要公告是什么?"
initial_input = {"messages": [("user", user_query)]}
for chunk in tool_agent_app.stream(initial_input, stream_mode="values"):
chunk["messages"][-1].pretty_print()
执行追踪:
plaintext复制================================= 用户消息 =================================
苹果最新WWDC活动的主要公告是什么?
------ 代理: 思考中... ---
--- 路由器: 决定调用工具 ---
================================== AI消息 ==================================
工具调用: web_search (call_abc123)
参数:
query: 苹果WWDC最新公告
--- 工具消息 ---
名称: web_search
[{"title": "WWDC 2025: 我们所知的一切...", "url": "...", "content": "苹果活动持续了一小时...我们总结了所有公告...iOS 26, iPadOS 26, macOS Tahoe..."}]
------ 代理: 思考中... ---
--- 路由器: 决定结束 ---
================================== AI消息 ==================================
苹果最新WWDC活动的主要公告包括将影响未来十年iOS、iPadOS和macOS开发的新设计...
4.8 工具使用评估
python复制class ToolUseEvaluation(BaseModel):
"""工具使用评估Schema"""
tool_selection_score: int = Field(description="代理选择正确工具的评分(1-5)")
tool_input_score: int = Field(description="工具输入格式良好的评分(1-5)")
synthesis_quality_score: int = Field(description="代理综合工具输出的评分(1-5)")
justification: str = Field(description="评分理由")
# 评估工具使用性能
evaluation = evaluate_tool_use_performance(user_query, agent_output)
评估结果:
plaintext复制{
'tool_selection_score': 5,
'tool_input_score': 5,
'synthesis_quality_score': 4,
'justification': "AI代理正确使用了网页搜索工具...工具输出格式良好且相关...综合信息方面还有改进空间..."
}
5. ReAct架构实现
5.1 ReAct模式原理
ReAct(Reason + Act)模式创建了一个动态推理循环,使代理能够处理需要多步推理的复杂问题。
核心流程:
- 接收目标
- 思考(Reason):生成下一步想法
- 行动(Act):执行行动(如调用工具)
- 观察(Observe):获取工具结果
- 重复:用新信息回到思考步骤
5.2 代理节点调整
python复制def react_agent_node(state: AgentState):
"""思考并决定下一步"""
console.print("--- REACT代理: 思考中... ---")
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
5.3 工作流构建
python复制react_graph_builder = StateGraph(AgentState)
react_graph_builder.add_node("agent", react_agent_node)
react_graph_builder.add_node("tools", tool_node) # 重用之前的工具节点
react_graph_builder.set_entry_point("agent")
react_graph_builder.add_conditional_edges("agent", router_function)
react_graph_builder.add_edge("tools", "agent") # 关键循环
react_agent_app = react_graph_builder.compile()
5.4 多步查询测试
python复制multi_step_query = "创作科幻电影《沙丘》的公司现任CEO是谁?该公司最近一部电影的预算是多少?"
console.print(f"测试ReAct代理处理多步查询: '{multi_step_query}'\n")
for chunk in react_agent_app.stream({"messages": [("user", multi_step_query)]}, stream_mode="values"):
chunk['messages'][-1].pretty_print()
执行追踪:
plaintext复制--- REACT代理: 思考中... ---
--- 路由器: 调用工具... ---
================================== AI消息 ==================================
工具调用: web_search...
参数: query: 创作《沙丘》的公司CEO
--- 工具消息 ---
[{"title": "沙丘: 第三部 - 维基百科", "content": "传奇CEO Joshua Grode..."}]
--- REACT代理: 思考中... ---
--- 路由器: 结束... ---
================================== AI消息 ==================================
创作《沙丘》的公司CEO是Joshua Grode... 最近电影预算未找到...
5.5 任务评估
python复制class TaskEvaluation(BaseModel):
"""任务评估Schema"""
task_completion_score: int = Field(description="任务完成度评分(1-10)")
reasoning_quality_score: int = Field(description="推理过程质量评分(1-10)")
justification: str = Field(description="评分理由")
evaluation = evaluate_agent_output(multi_step_query, agent_output)
评估结果:
plaintext复制{
'task_completion_score': 8,
'reasoning_quality_score': 9,
'justification': "代理正确地将问题分解为多个步骤...成功识别了公司和CEO...虽然未能找到最近电影的预算,但其推理过程合理..."
}
6. 规划架构(Planning)实现
6.1 规划模式原理
规划模式在解决方案路径可预测时,提供了比纯反应式更结构化和高效的方法。
核心流程:
- 计划(Plan):创建完整的子任务列表
- 执行(Execute):按顺序执行每个子任务
- 综合(Synthesize):组装最终答案
6.2 数据结构定义
python复制class Plan(BaseModel):
"""回答用户查询的工具调用计划"""
steps: List[str] = Field(description="执行后将回答查询的工具调用列表")
class PlanningState(TypedDict):
user_request: str
plan: Optional[List[str]]
intermediate_steps: List[str] # 存储工具输出
final_answer: Optional[str]
6.3 规划节点实现
python复制def planner_node(state: PlanningState):
"""生成回答用户请求的行动计划"""
console.print("--- 规划器: 分解任务... ---")
planner_llm = llm.with_structured_output(Plan)
prompt = f"""你是一名专业规划师。你的工作是创建一个分步计划来回答用户的请求。
计划中的每一步必须是对`web_search`工具的单一调用。
**用户请求:**
{state['user_request']}
"""
plan_result = planner_llm.invoke(prompt)
console.print(f"--- 规划器: 生成计划: {plan_result.steps} ---")
return {"plan": plan_result.steps}
6.4 执行节点实现
python复制def executor_node(state: PlanningState):
"""执行计划中的下一步"""
console.print("--- 执行器: 运行下一步... ---")
next_step = state["plan"][0]
query = next_step.replace("web_search('", "").replace("')", "")
result = search_tool.invoke({"query": query})
return {
"plan": state["plan"][1:], # 移除已执行的步骤
"intermediate_steps": state["intermediate_steps"] + [result]
}
6.5 路由逻辑
python复制def planning_router(state: PlanningState):
"""根据计划状态路由"""
if not state["plan"]:
console.print("--- 路由器: 计划完成。转到综合器。 ---")
return "synthesize"
else:
console.print("--- 路由器: 计划还有步骤。继续执行。 ---")
return "execute"
6.6 工作流构建
python复制planning_graph_builder = StateGraph(PlanningState)
planning_graph_builder.add_node("plan", planner_node)
planning_graph_builder.add_node("execute", executor_node)
planning_graph_builder.add_node("synthesize", synthesizer_node)
planning_graph_builder.set_entry_point("plan")
planning_graph_builder.add_conditional_edges("plan", planning_router)
planning_graph_builder.add_conditional_edges("execute", planning_router)
planning_graph_builder.add_edge("synthesize", END)
planning_agent_app = planning_graph_builder.compile()
6.7 测试案例
python复制plan_centric_query = """找出法国、德国和意大利首都城市的人口,然后计算它们的总和。"""
initial_planning_input = {"user_request": plan_centric_query, "intermediate_steps": []}
final_planning_output = planning_agent_app.invoke(initial_planning_input)
执行追踪:
plaintext复制--- 规划器: 分解任务... ---
--- 规划器: 生成计划: ["web_search('巴黎人口')", "web_search('柏林人口')", "web_search('罗马人口')"] ---
--- 路由器: 计划还有步骤。继续执行。 ---
--- 执行器: 运行下一步... ---
...
6.8 过程评估
python复制class ProcessEvaluation(BaseModel):
"""过程评估Schema"""
task_completion_score: int = Field(description="任务完成度评分(1-10)")
process_efficiency_score: int = Field(description="代理过程的效率和直接性评分(1-10)")
justification: str = Field(description="评分理由")
evaluation = evaluate_process(plan_centric_query, agent_output)
评估结果:
plaintext复制{
'task_completion_score': 8,
'process_efficiency_score': 9,
'justification': "代理预先创建了一个清晰、最优的计划...执行过程高效直接..."
}
7. PEV架构实现
7.1 PEV模式原理
PEV(Planner-Executor-Verifier)架构在规划模式基础上增加了质量控制层,使系统能够自我修正。
核心流程:
- 计划(Plan):创建步骤序列
- 执行(Execute):执行下一步
- 验证(Verify):检查工具输出
- 路由和迭代:
- 成功:继续下一步
- 失败:重新计划
- 完成:生成最终答案
7.2 有缺陷的工具模拟
python复制def flaky_web_search(query: str) -> str:
"""故意对特定查询返回错误的搜索工具"""
console.print(f"--- 工具: 搜索 '{query}'... ---")
if "员工人数" in query.lower():
console.print("--- 工具: [bold red]模拟API失败![/bold red] ---")
return "错误: 无法获取数据。API端点当前不可用。"
else:
return search_tool.invoke({"query": query})
7.3 验证节点实现
python复制class VerificationResult(BaseModel):
"""验证结果Schema"""
is_successful: bool = Field(description="工具执行是否成功且数据有效")
reasoning: str = Field(description="验证决定的理由")
def verifier_node(state: PEVState):
"""检查最后工具结果是否有错误"""
console.print("--- 验证器: 检查最后工具结果... ---")
verifier_llm = llm.with_structured_output(VerificationResult)
prompt = f"验证以下工具输出是成功有效的结果还是错误消息。任务是 '{state['user_request']}'.\n\n工具输出: '{state['last_tool_result']}'"
verification = verifier_llm.invoke(prompt)
console.print(f"--- 验证器: 判断为 '{'成功' if verification.is_successful else '失败'}' ---")
if verification.is_successful:
return {"intermediate_steps": state["intermediate_steps"] + [state['last_tool_result']]}
else:
return {"plan": [], "intermediate_steps": state["intermediate_steps"] + [f"验证失败: {state['last_tool_result']}"]}
7.4 工作流构建
python复制class PEVState(TypedDict):
user_request: str
plan: List[str]
intermediate_steps: List[str]
last_tool_result: Optional[str]
final_answer: Optional[str]
pev_graph_builder = StateGraph(PEVState)
pev_graph_builder.add_node("plan", planner_node)
pev_graph_builder.add_node("execute", executor_node)
pev_graph_builder.add_node("verify", verifier_node)
pev_graph_builder.add_node("synthesize", synthesizer_node)
pev_graph_builder.set_entry_point("plan")
def pev_router(state: PEVState):
if not state["plan"]:
if any("验证失败" in step for step in state["intermediate_steps"]):
console.print("--- 路由器: 计划失败。重新规划。 ---")
return "plan"
else:
console.print("--- 路由器: 计划完成。综合结果。 ---")
return "synthesize"
else:
console.print("--- 路由器: 执行下一步。 ---")
return "execute"
pev_graph_builder.add_conditional_edges("plan", lambda _: "execute")
pev_graph_builder.add_conditional_edges("execute", lambda _: "verify")
pev_graph_builder.add_conditional_edges("verify", pev_router)
pev_graph_builder.add_edge("synthesize", END)
pev_agent_app = pev_graph_builder.compile()
7.5 测试案例
python复制failure_test_query = "找出创作《沙丘》的公司员工人数和最近一部电影的预算"
initial_pev_input = {
"user_request": failure_test_query,
"plan": [],
"intermediate_steps": [],
"last_tool_result": None
}
final_pev_output = pev_agent_app.invoke(initial_pev_input)
执行追踪:
plaintext复制--- 规划器: 分解任务... ---
--- 规划器: 生成计划: ["web_search('创作《沙丘》的公司员工人数')", "web_search('创作《沙丘》的公司最近电影预算')"] ---
--- 路由器: 执行下一步。 ---
--- 执行器: 运行下一步... ---
--- 工具: 搜索 '创作《沙丘》的公司员工人数'... ---
--- 工具: 模拟API失败! ---
--- 验证器: 检查最后工具结果... ---
--- 验证器: 判断为 '失败' ---
--- 路由器: 计划失败。重新规划。 ---
--- 规划器: 分解任务... ---
--- 规划器: 生成计划: ["web_search('传奇娱乐公司信息')"] ---
...
7.6 健壮性评估
python复制class RobustnessEvaluation(BaseModel):
"""健壮性评估Schema"""
error_recovery_score: int = Field(description="从错误中恢复的能力评分(1-10)")
final_accuracy_score: int = Field(description="最终答案准确性评分(1-10)")
justification: str = Field(description="评分理由")
evaluation = evaluate_robustness(failure_test_query, agent_output)
评估结果:
plaintext复制{
'error_recovery_score': 9,
'final_accuracy_score': 7,
'justification': "代理成功检测到API失败并重新规划...虽然未能获取所有请求信息,但在困难条件下表现良好..."
}
8. 架构组合与系统集成
8.1 模式组合策略
在实际系统中,我们通常需要组合多种架构模式。以下是常见组合方式:
- 规划+PEV:用于需要健壮性的结构化任务
- ReAct+工具使用:用于探索性任务
- 反射+多代理:用于高质量输出生成
8.2 组合示例
python复制def hybrid_agent(query: str):
"""组合多种架构的混合代理"""
# 第一步:确定任务类型
task_type = classify_task(query)
if task_type == "structured":
# 使用规划+PEV
return pev_agent_app.invoke({"user_request": query, ...})
elif task_type == "exploratory":
# 使用ReAct
return react_agent_app.invoke({"messages": [("user", query)]})
else:
# 默认使用工具使用+反射
draft = reflection_app.invoke({"user_request": query})
return tool_agent_app.invoke({"messages": [("user", draft['refined_code'])]})
8.3 系统集成注意事项
- 状态管理:确保不同架构间的状态转换清晰
- 错误处理:实现统一的错误处理机制
- 性能监控:跟踪各组件性能指标
- 资源分配:根据任务类型动态分配资源
9. 性能优化技巧
9.1 缓存策略
python复制from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache
# 启用内存缓存
set_llm_cache(InMemoryCache())
# 或使用Redis缓存
from langchain.cache import RedisCache
set_llm_cache(RedisCache(redis_url="redis://localhost:6379"))
9.2 并行执行
python复制from langchain.runnables import RunnableParallel
parallel = RunnableParallel({
"step1": chain1,
"step2": chain2
})
# 并行执行多个链
results = parallel.invoke(input)
9.3 批处理
python复制# 使用batch处理多个输入
inputs = [{"query": q} for q in queries]
results = chain.batch(inputs)
10. 评估与监控
10.1 评估指标
- 任务完成率:代理成功完成的任务比例
- 工具使用准确率:正确选择和使用工具的比例
- 响应时间:从查询到回答的时间
- 用户满意度:人工评估回答质量
10.2 LangSmith监控
python复制# 配置LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "Production-Monitoring"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
10.3 自定义评估器
python复制def custom_evaluator(run, example):
"""自定义评估逻辑"""
# 检查工具使用
tool_used = any(op["op"] == "tool" for op in run.ops)
# 检查响应质量
response = run.outputs["output"]
quality_score = llm.evaluate_response(response)
return {
"tool_used": tool_used,
"quality_score": quality_score
}
11. 实际应用建议
- 渐进式复杂化:从简单架构开始,逐步增加复杂性
- 模块化设计:保持各组件独立可替换
- 持续评估:建立自动化评估流程
- 用户反馈循环:收集真实用户反馈指导改进
12. 常见问题排查
12.1 工具调用失败
症状:代理无法正确调用工具
解决方案:
- 检查工具描述是否清晰
- 验证工具参数格式
- 测试工具独立于代理
12.2 循环逻辑
症状:代理陷入无限循环
解决方案:
- 设置最大迭代次数
- 添加循环检测逻辑
- 优化提示词减少模糊性
12.3 低质量输出
症状:回答不准确或无关
解决方案:
- 增强反射机制
- 改进验证步骤
- 提供更详细的示例
13. 未来发展方向
- 自适应架构选择:根据任务动态选择最佳架构
- 多模态扩展:整合视觉、音频等模态
- 长期记忆:实现跨会话知识保持
- 自我优化:自动调整内部参数
14. 总结与资源
本文详细介绍了17种智能代理架构模式中的5种核心模式,包括:
- 反射架构(Reflection)
- 工具使用架构(Tool Use)
- ReAct架构
- 规划架构(Planning)
- PEV架构(Planner-Executor-Verifier)
每种模式都有其适用场景和优势,实际系统中往往需要组合使用多种模式。
完整代码可在GitHub获取:
GitHub仓库 - all-agentic-architectures
进一步学习资源:
- LangChain官方文档
- LangGraph示例库
- 最新AI代理研究论文
在实际应用中,建议从简单模式开始,逐步增加复杂性,并建立完善的评估监控体系。记住,没有放之四海而皆准的解决方案,最佳架构总是取决于具体应用场景和需求。
