Agent 设计模式
大型语言模型(LLM)的 Agent 设计模式是构建复杂、智能应用的关键。这些模式赋予 LLM 规划、反思、使用工具以及与其他 Agent 协作的能力。
1. Tool Use Pattern(工具使用模式)
核心思想: Agent 能够调用外部工具来扩展自身能力,弥补 LLM 无法进行实时计算、访问最新信息或执行特定操作的局限。这是最基础也是最普遍的 Agent 能力。
浅入深: 从调用单个简单工具开始,逐步过渡到调用多个不同类型的工具。
from langchain_community.llms import OpenAI
from langchain.agents import AgentExecutor, create_react_agent, Tool
from langchain import PromptTemplate
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper# 1. 定义工具
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
tools = [Tool(name="Wikipedia",func=wikipedia.run,description="用于查询维基百科的工具。当你需要了解历史事件、人物、科学概念等知识时,可以使用它。")
]# 2. 定义 LLM
llm = OpenAI(temperature=0) # 实际使用时请替换为你的LLM客户端和API Key# 3. 定义提示模板 (ReAct 风格,这里简化为Tool Use)
template = """
你是一个智能助手,可以回答用户的问题。
你可以使用以下工具:
{tools}请严格按照以下格式回答:
问题: 用户的问题
思考: 你需要思考如何回答这个问题。
工具: 如果你需要使用工具,请指定工具名称和输入参数。例如: Wikipedia[输入参数]
工具输出: 工具的输出
思考: 对工具输出进行总结并生成最终答案。
最终答案: 最终答案问题: {input}
{agent_scratchpad}
"""
prompt = PromptTemplate.from_template(template)# 4. 创建 Agent (使用 create_react_agent 简化创建过程)
agent = create_react_agent(llm, tools, prompt)# 5. 创建 Agent 执行器
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)# 6. 运行 Agent
response = agent_executor.invoke({"input": "秦始皇的出生日期和去世日期是什么?"})
print("\n--- 最终响应 ---")
print(response["output"])
2. ReAct Pattern(思考-行动模式)
核心思想: ReAct (Reasoning and Acting) 模式结合了 LLM 的推理能力(思考)和外部工具的使用(行动)。Agent 通过内部的“思考”步骤来决定下一步的“行动”,并在观察行动结果后继续“思考”,形成一个循环,直到完成任务。
浅入深: 理解 ReAct 的核心循环,并能识别思考和行动的日志输出。
上面 Tool Use 的代码示例实际上已经包含了 ReAct 的思想,因为 create_react_agent
默认就是使用 ReAct 模式。verbose=True
会打印出每次“思考”和“行动”的日志。
在实际运行上述代码时,会看到类似这样的输出:
> Entering new AgentExecutor chain...
思考: 用户问的是秦始皇的出生日期和去世日期,这些信息可以在维基百科上找到。我需要使用Wikipedia工具来查询“秦始皇”。
工具: Wikipedia[秦始皇]
工具输出: (此处省略维基百科查询结果,包含出生和去世日期)
思考: 我已经从维基百科获取了秦始皇的出生日期和去世日期。我可以根据这些信息生成最终答案了。
最终答案: 秦始皇(嬴政)出生于公元前259年,去世于公元前210年。> Finished chain.
这里“思考”和“工具”就是 ReAct 模式的核心体现。
3. Reflection Pattern(反思模式)
核心思想: Agent 不仅能执行任务,还能对自己或任务的执行过程进行反思、评估,并从错误中学习。这使得 Agent 能够自我纠正、提高任务完成的质量和效率。
浅入深: 从简单的结果评估开始,发展到对整个决策过程的反思。
LangChain 中的 langchain_experimental.autonomous_agents.autogpt
和 langchain.agents.AgentExecutor
的 handle_parsing_errors
参数可以部分体现反思。更复杂的反思通常需要多Agent协同或自定义的循环逻辑。
from langchain_community.llms import OpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParserllm = OpenAI(temperature=0.7)# 1. 任务执行 Agent
task_agent_prompt = PromptTemplate.from_template("""
你是一个内容生成专家。请为主题 "{topic}" 生成一篇短文。
要求:至少包含300字,并且语言生动有趣。
短文:
""")
task_agent_chain = task_agent_prompt | llm | StrOutputParser()# 2. 反思 Agent
reflection_prompt = PromptTemplate.from_template("""
你是一个内容评估专家。请评估以下短文:
---
{article}
---
请根据以下标准给出评价:
1. 字数是否达到300字?
2. 语言是否生动有趣?
3. 内容是否切题?
如果存在不足,请给出改进建议。如果非常满意,请直接说“满意”。评估和建议:
""")
reflection_chain = reflection_prompt | llm | StrOutputParser()def run_with_reflection(topic, max_attempts=3):print(f"--- 开始为 '{topic}' 生成文章 ---")current_article = ""for attempt in range(max_attempts):print(f"\n--- 尝试 {attempt + 1} ---")current_article = task_agent_chain.invoke({"topic": topic})print(f"生成的文章:\n{current_article[:200]}...") # 打印部分内容reflection = reflection_chain.invoke({"article": current_article})print(f"反思评估:\n{reflection}")if "满意" in reflection or "符合要求" in reflection: # 简化判断print("--- 评估满意,任务完成!---")return current_articleelse:print("--- 评估不满意,尝试改进。---")# 在实际应用中,这里需要将 reflection 的建议反馈给 task_agent_chain# 例如:修改 task_agent_prompt,加入“请根据上次的评估:[reflection] 进行改进”# 为简化,本示例仅演示循环print("--- 达到最大尝试次数,任务结束。---")return current_article# 运行带有反思的 Agent
generated_article = run_with_reflection("量子计算的未来影响")
print("\n--- 最终生成的文章 ---")
print(generated_article)
4. Planning Pattern(规划模式)
核心思想: Agent 能够将复杂的任务分解为更小、更易管理的子任务,并为每个子任务制定执行步骤。这使得 Agent 能够处理多步骤、高难度的任务。
浅入深: 从简单的链式规划(一步接一步)到复杂的树状规划(分支和子任务)。
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate# 1. 定义子任务 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)# 2. 定义规划 Agent
planning_prompt = PromptTemplate.from_template("""
你是一个任务规划专家。请将以下复杂任务分解为一系列清晰、可执行的子任务,并按顺序给出。
任务: {task}
输出格式:
1. 子任务一:...
2. 子任务二:...
...
""")
planning_chain = planning_prompt | llm | StrOutputParser()# 3. 定义执行 Agent (简化为根据子任务生成内容)
execution_prompt = PromptTemplate.from_template("""
你是一个执行专家,正在完成一个任务分解后的子任务。
当前任务主题是: {overall_task}
请完成以下子任务: {sub_task}
你的输出应该直接是完成子任务的内容。
""")
execution_chain = execution_prompt | llm | StrOutputParser()def run_planning_agent(overall_task):print(f"--- 任务规划: {overall_task} ---")plan_output = planning_chain.invoke({"task": overall_task})print("\n生成的计划:")print(plan_output)sub_tasks = [line.strip() for line in plan_output.split('\n') if line.strip().startswith(tuple(str(i) + '.' for i in range(1,10)))]results = []print("\n--- 执行子任务 ---")for i, sub_task in enumerate(sub_tasks):print(f"\n执行 {sub_task}")result = execution_chain.invoke({"overall_task": overall_task, "sub_task": sub_task})print(f"结果: {result[:100]}...") # 打印部分结果results.append(result)# 4. 整合结果integration_prompt = PromptTemplate.from_template("""
你是一个结果整合专家。请将以下子任务的成果整合起来,形成关于主题 "{overall_task}" 的最终报告。
子任务成果:
{results}
最终报告:
""")final_report = integration_prompt.invoke({"overall_task": overall_task, "results": "\n".join(results)})print("\n--- 最终报告 ---")print(final_report)# 运行规划 Agent
run_planning_agent("撰写一篇关于人工智能在教育领域应用现状与前景的详细报告")
5. Multi-Agent Pattern(多 Agent 模式)
核心思想: 多个 Agent 各自拥有专门的角色和能力,它们通过协作、沟通来共同解决更复杂、更大规模的任务。每个 Agent 可以是上述任何一种模式的实例。
浅入深: 从两个 Agent 的简单对话协作,到形成团队(如辩论、专家组),再到复杂的Agent组织结构。
多智能体这里只是给出一个简化例子
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser# 初始化 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)# 1. Agent A: 内容专家
content_expert_prompt = ChatPromptTemplate.from_messages([SystemMessage("你是一位内容创建专家,擅长生成引人入胜的营销文案。"),HumanMessage("{input}")
])
content_expert_chain = content_expert_prompt | llm | StrOutputParser()# 2. Agent B: 营销策略专家
marketing_strategy_prompt = ChatPromptTemplate.from_messages([SystemMessage("你是一位营销策略专家,擅长评估文案的营销效果,并给出改进建议,使其更具吸引力。"),HumanMessage("请评估以下营销文案,并给出详细的改进建议:\n{copy}\n\n你的评估和建议:")
])
marketing_strategy_chain = marketing_strategy_prompt | llm | StrOutputParser()# 3. 协调 Agent (Driver Agent)
def multi_agent_workflow(product_info, max_iterations=2):print(f"--- 多 Agent 协作开始:{product_info} ---")current_copy = ""for i in range(max_iterations):print(f"\n--- 迭代 {i + 1} ---")if i == 0:# 第一次迭代由内容专家基于产品信息生成初稿print("内容专家正在生成初稿...")current_copy = content_expert_chain.invoke({"input": f"为以下产品信息生成一份营销文案:{product_info}"})else:# 后续迭代由内容专家根据营销策略专家的建议进行改进print("内容专家正在根据建议改进文案...")current_copy = content_expert_chain.invoke({"input": f"这是现有的营销文案:{current_copy}\n请根据以下建议进行改进:{last_suggestion}"})print(f"当前文案:\n{current_copy[:150]}...")print("营销策略专家正在评估并给出建议...")last_suggestion = marketing_strategy_chain.invoke({"copy": current_copy})print(f"营销策略专家的建议:\n{last_suggestion}")if "完美" in last_suggestion or "无需改进" in last_suggestion: # 简化判断print("--- 评估满意,协作完成!---")breakprint("\n--- 最终营销文案 ---")return current_copy# 运行多 Agent 协作
final_marketing_copy = multi_agent_workflow("一款新型智能家居控制器,特点是语音识别精准、可连接所有品牌家电、隐私保护严格。")
print(final_marketing_copy)
总结
- Tool Use 和 ReAct 是最基本且强大的能力,让 Agent 能够与外部世界交互。
- Reflection 赋予 Agent 自我学习和纠错的能力,提高可靠性。
- Planning 让 Agent 能够处理复杂的多步骤任务。
- Multi-Agent 模式则将个体 Agent 的能力汇聚成协同工作的系统,开启了构建真正智能的自治系统的大门。