LangChain Agents 代理
Agents 是 LangChain 的核心功能，让 LLM 能够使用工具、做出决策，并完成复杂任务。
#什么是 Agent？
Agent 是一个能够：
- 理解用户意图
- 选择合适的工具
- 执行多步骤任务
- 根据结果调整策略
用户: "北京今天天气怎么样？适合户外运动吗？"
↓
Agent 思考: 需要先获取天气信息
↓
Agent 调用: get_weather("北京")
↓
Agent 收到: "晴，25°C，空气质量良好"
↓
Agent 回答: "北京今天晴朗，25°C，空气质量良好，非常适合户外运动！"
#快速开始
#创建简单 Agent
pythonfrom langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate from langchain_core.tools import tool # 1. 定义工具 @tool def get_weather(city: str) -> str: """获取指定城市的天气信息""" # 模拟天气 API weather_data = { "北京": "晴，25°C", "上海": "多云，28°C" } return weather_data.get(city, f"{city}天气数据暂不可用") @tool def calculate(expression: str) -> str: """执行数学计算""" try: return str(eval(expression)) except: return "计算错误" # 2. 创建 Prompt prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个有帮助的助手，可以使用工具来回答问题。"), ("placeholder", "{chat_history}"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}") ]) # 3. 创建 Agent llm = ChatOpenAI(model="gpt-4o") tools = [get_weather, calculate] agent = create_tool_calling_agent(llm, tools, prompt) # 4. 创建 AgentExecutor agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # 5. 运行 result = agent_executor.invoke({ "input": "北京天气怎么样？另外帮我算一下 123 * 456" }) print(result["output"])
#定义工具
#使用 @tool 装饰器
pythonfrom langchain_core.tools import tool @tool def search_web(query: str) -> str: """搜索网络获取信息 Args: query: 搜索关键词 Returns: 搜索结果摘要 """ # 实现搜索逻辑 return f"关于 '{query}' 的搜索结果..." @tool def send_email(to: str, subject: str, body: str) -> str: """发送电子邮件 Args: to: 收件人邮箱 subject: 邮件主题 body: 邮件正文 Returns: 发送状态 """ # 实现发送逻辑 return f"邮件已发送给 {to}"
#使用 StructuredTool
pythonfrom langchain_core.tools import StructuredTool from pydantic import BaseModel, Field class SearchInput(BaseModel): query: str = Field(description="搜索关键词") max_results: int = Field(default=5, description="最大结果数") def search_function(query: str, max_results: int = 5) -> str: return f"搜索 '{query}'，返回 {max_results} 条结果" search_tool = StructuredTool.from_function( func=search_function, name="search", description="搜索信息", args_schema=SearchInput )
#内置工具
python# 搜索工具 from langchain_community.tools import DuckDuckGoSearchRun search = DuckDuckGoSearchRun() # Wikipedia from langchain_community.tools import WikipediaQueryRun from langchain_community.utilities import WikipediaAPIWrapper wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper()) # 计算器 from langchain_community.tools import LLMMathChain math_tool = LLMMathChain.from_llm(llm) # Python REPL from langchain_experimental.tools import PythonREPLTool python_repl = PythonREPLTool()
#Agent 类型
#Tool Calling Agent (推荐)
pythonfrom langchain.agents import create_tool_calling_agent agent = create_tool_calling_agent(llm, tools, prompt)
#ReAct Agent
pythonfrom langchain.agents import create_react_agent from langchain import hub # 使用预定义的 ReAct prompt prompt = hub.pull("hwchase17/react") agent = create_react_agent(llm, tools, prompt)
#OpenAI Functions Agent
pythonfrom langchain.agents import create_openai_functions_agent agent = create_openai_functions_agent(llm, tools, prompt)
#Agent 配置
#AgentExecutor 选项
pythonagent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, # 显示详细日志 max_iterations=10, # 最大迭代次数 max_execution_time=60, # 最大执行时间（秒） early_stopping_method="generate", # 早停方式 handle_parsing_errors=True, # 处理解析错误 return_intermediate_steps=True # 返回中间步骤 )
#流式输出
pythonasync for event in agent_executor.astream_events( {"input": "你的问题"}, version="v1" ): kind = event["event"] if kind == "on_chat_model_stream": content = event["data"]["chunk"].content if content: print(content, end="")
#带记忆的 Agent
pythonfrom langchain_community.chat_message_histories import ChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory # 创建记忆存储 store = {} def get_session_history(session_id: str): if session_id not in store: store[session_id] = ChatMessageHistory() return store[session_id] # 添加记忆 agent_with_history = RunnableWithMessageHistory( agent_executor, get_session_history, input_messages_key="input", history_messages_key="chat_history" ) # 使用（同一 session_id 会保持对话历史） result = agent_with_history.invoke( {"input": "我叫小明"}, config={"configurable": {"session_id": "user-123"}} ) result = agent_with_history.invoke( {"input": "我叫什么名字？"}, config={"configurable": {"session_id": "user-123"}} ) # 会记住你叫小明
#实用示例
#研究助手 Agent
pythonfrom langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate from langchain_core.tools import tool from langchain_community.tools import DuckDuckGoSearchRun # 工具 search = DuckDuckGoSearchRun() @tool def take_notes(content: str) -> str: """保存研究笔记""" with open("research_notes.txt", "a") as f: f.write(content + "\n\n") return "笔记已保存" @tool def read_notes() -> str: """读取之前的研究笔记""" try: with open("research_notes.txt", "r") as f: return f.read() except FileNotFoundError: return "暂无笔记" # Prompt prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个研究助手。你可以： 1. 搜索网络获取信息 2. 保存重要发现到笔记 3. 回顾之前的笔记 请系统性地进行研究，并记录重要发现。"""), ("placeholder", "{chat_history}"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}") ]) # 创建 Agent llm = ChatOpenAI(model="gpt-4o") tools = [search, take_notes, read_notes] agent = create_tool_calling_agent(llm, tools, prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # 使用 result = agent_executor.invoke({ "input": "研究一下 2024 年 AI 发展趋势，并记录关键发现" })
#代码助手 Agent
pythonfrom langchain_experimental.tools import PythonREPLTool from langchain_core.tools import tool python_repl = PythonREPLTool() @tool def read_file(filepath: str) -> str: """读取文件内容""" with open(filepath, "r") as f: return f.read() @tool def write_file(filepath: str, content: str) -> str: """写入文件""" with open(filepath, "w") as f: f.write(content) return f"已写入 {filepath}" prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个 Python 编程助手。你可以： 1. 执行 Python 代码 2. 读取和写入文件 3. 帮助用户完成编程任务 执行代码前请仔细检查，确保安全。"""), ("human", "{input}"), ("placeholder", "{agent_scratchpad}") ]) tools = [python_repl, read_file, write_file] agent = create_tool_calling_agent(llm, tools, prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # 使用 result = agent_executor.invoke({ "input": "帮我写一个读取 CSV 文件并计算平均值的 Python 脚本" })
#最佳实践
#1. 清晰的工具描述
python@tool def search(query: str) -> str: """搜索网络获取最新信息。 适用场景： - 查找最新新闻和事件 - 获取实时数据 - 验证事实信息 不适用于： - 需要深度分析的问题 - 历史文档查询 Args: query: 搜索关键词，尽量具体明确 Returns: 搜索结果的摘要 """ pass
#2. 限制迭代次数
pythonagent_executor = AgentExecutor( agent=agent, tools=tools, max_iterations=5, # 防止无限循环 max_execution_time=30 # 超时限制 )
#3. 错误处理
python@tool def risky_operation(param: str) -> str: """可能出错的操作""" try: # 执行操作 return "成功" except Exception as e: return f"操作失败: {str(e)}"
#4. 安全考虑
python# 限制代码执行 python_repl = PythonREPLTool( # 在沙箱环境中运行 ) # 验证输入 @tool def safe_search(query: str) -> str: """安全搜索""" # 过滤敏感内容 if any(bad in query.lower() for bad in ["hack", "attack"]): return "搜索被拒绝" return do_search(query)
#下一步
提示：Agent 功能强大但需要谨慎使用，尤其是涉及代码执行和文件操作时。
