代理（Agents） #
代理是 LangChain 中最强大的功能之一，它让 LLM 能够自主决策、调用工具、执行复杂任务。
代理概述 #
text
┌─────────────────────────────────────────────────────────────┐
│ 什么是代理？ │
├─────────────────────────────────────────────────────────────┤
│ │
│ 普通链： │
│ 输入 ───> 固定流程 ───> 输出 │
│ │
│ 代理： │
│ 输入 ───> LLM 思考 ───> 决策 ───> 执行工具 ───> 观察 │
│ │ │
│ └──> 循环直到完成任务 │
│ │
│ 代理可以： │
│ ✅ 自主选择使用哪个工具 │
│ ✅ 决定是否需要更多信息 │
│ ✅ 多步骤完成任务 │
│ ✅ 处理意外情况 │
│ │
└─────────────────────────────────────────────────────────────┘
代理工作原理 #
text
┌─────────────────────────────────────────────────────────────┐
│ ReAct 模式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户: "北京今天天气怎么样？温度是多少？" │
│ │
│ 思考: 我需要查询北京的天气信息 │
│ 行动: get_weather("北京") │
│ 观察: 北京今天晴，温度 25°C，湿度 45% │
│ │
│ 思考: 我已经获得了天气信息，可以回答用户了 │
│ 回答: 北京今天天气晴朗，温度是 25°C，湿度 45%。 │
│ │
└─────────────────────────────────────────────────────────────┘
创建代理 #
使用 LangGraph（推荐） #
LangGraph 是 LangChain 新一代的代理框架，提供更灵活的控制：
python
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini")
# 定义工具
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""获取指定城市的天气信息"""
weather_data = {
"北京": "晴，25°C",
"上海": "多云，28°C",
"广州": "雨，30°C"
}
return weather_data.get(city, f"未找到{city}的天气信息")
@tool
def calculate(expression: str) -> float:
"""计算数学表达式"""
try:
return eval(expression)
except:
return "计算错误"
# 创建代理
tools = [get_weather, calculate]
agent = create_react_agent(model, tools)
# 运行
result = agent.invoke({
"messages": [("user", "北京今天天气怎么样？")]
})
print(result["messages"][-1].content)
使用 AgentExecutor #
python
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有帮助的助手"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
tools = [get_weather, calculate]
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = agent_executor.invoke({"input": "北京天气怎么样？"})
print(result["output"])
text
┌─────────────────────────────────────────────────────────────┐
│ 代理执行流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 接收用户输入 │
│ │ │
│ ▼ │
│ 2. LLM 分析任务，决定是否需要工具 │
│ │ │
│ ┌────┴────┐ │
│ │ │ │
│ 不需要 需要工具 │
│ │ │ │
│ │ ▼ │
│ │ 3. 选择并执行工具 │
│ │ │ │
│ │ ▼ │
│ │ 4. 获取工具输出 │
│ │ │ │
│ │ ▼ │
│ │ 5. LLM 分析结果，决定下一步 │
│ │ │ │
│ │ ┌────┴────┐ │
│ │ │ │ │
│ │ 继续 任务完成 │
│ │ │ │ │
│ └────┼─────────┘ │
│ │ │
│ ▼ │
│ 6. 返回最终答案 │
│ │
└─────────────────────────────────────────────────────────────┘
代理类型 #
1. ReAct Agent #
推理+行动模式：
python
from langchain.agents import create_react_agent
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个有帮助的助手。
你可以使用以下工具：
{tools}
使用以下格式：
Question: 用户的问题
Thought: 你应该思考做什么
Action: 要使用的工具，必须是 [{tool_names}] 之一
Action Input: 工具的输入
Observation: 工具的输出
... (这个 Thought/Action/Action Input/Observation 可以重复 N 次)
Thought: 我现在知道最终答案了
Final Answer: 最终答案"""),
("human", "{input}\n{agent_scratchpad}")
])
agent = create_react_agent(model, tools, prompt)
2. Function Calling Agent #
使用模型的函数调用能力：
python
from langchain.agents import create_tool_calling_agent
# OpenAI、Claude 等模型原生支持函数调用
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)
result = agent_executor.invoke({
"input": "计算 123 * 456"
})
3. Structured Chat Agent #
处理复杂结构化输入：
python
from langchain.agents import create_structured_chat_agent
@tool
def search_product(
name: str,
category: str = None,
min_price: float = None,
max_price: float = None
) -> str:
"""搜索产品
Args:
name: 产品名称
category: 产品类别
min_price: 最低价格
max_price: 最高价格
"""
return f"搜索结果: {name}, 类别: {category}, 价格范围: {min_price}-{max_price}"
agent = create_structured_chat_agent(model, [search_product], prompt)
4. OpenAI Tools Agent #
专门为 OpenAI 模型优化：
python
from langchain.agents import create_openai_tools_agent
agent = create_openai_tools_agent(model, tools, prompt)
工具定义 #
使用 @tool 装饰器 #
python
from langchain_core.tools import tool
@tool
def search_web(query: str) -> str:
"""搜索网络获取信息
Args:
query: 搜索关键词
"""
return f"搜索结果: {query}"
@tool
def get_current_time() -> str:
"""获取当前时间"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""发送邮件
Args:
to: 收件人邮箱
subject: 邮件主题
body: 邮件正文
"""
return f"邮件已发送至 {to}"
使用 StructuredTool #
python
from langchain.tools import StructuredTool
from pydantic import BaseModel, Field
class CalculatorInput(BaseModel):
a: float = Field(description="第一个数字")
b: float = Field(description="第二个数字")
operation: str = Field(description="运算类型: add, subtract, multiply, divide")
def calculator_func(a: float, b: float, operation: str) -> float:
operations = {
"add": a + b,
"subtract": a - b,
"multiply": a * b,
"divide": a / b if b != 0 else "除数不能为零"
}
return operations.get(operation, "未知操作")
calculator = StructuredTool(
name="calculator",
description="执行数学运算",
func=calculator_func,
args_schema=CalculatorInput
)
工具类 #
python
from langchain.tools import BaseTool
from typing import Optional, Type
class WeatherTool(BaseTool):
name = "get_weather"
description = "获取指定城市的天气信息"
def _run(self, city: str) -> str:
return f"{city}今天晴，温度 25°C"
async def _arun(self, city: str) -> str:
return self._run(city)
内置工具 #
搜索工具 #
python
from langchain_community.tools import (
DuckDuckGoSearchRun,
GoogleSearchAPIWrapper,
BingSearchAPIWrapper,
)
# DuckDuckGo（免费）
search = DuckDuckGoSearchRun()
result = search.run("LangChain 是什么")
# Google Search（需要 API Key）
from langchain_community.utilities import GoogleSearchAPIWrapper
search = GoogleSearchAPIWrapper()
Python REPL #
python
from langchain_experimental.tools import PythonREPLTool
python_repl = PythonREPLTool()
result = python_repl.run("print([x**2 for x in range(10)])")
文件操作 #
python
from langchain_community.tools import (
ReadFileTool,
WriteFileTool,
ListDirectoryTool,
)
read_file = ReadFileTool()
write_file = WriteFileTool()
list_dir = ListDirectoryTool()
数据库查询 #
python
from langchain_community.tools.sql_database.tool import (
QuerySQLDataBaseTool,
InfoSQLDatabaseTool,
ListSQLDatabaseTool,
)
from langchain_community.utilities import SQLDatabase
db = SQLDatabase.from_uri("sqlite:///mydb.db")
query_tool = QuerySQLDataBaseTool(db=db)
info_tool = InfoSQLDatabaseTool(db=db)
list_tool = ListSQLDatabaseTool(db=db)
高级代理模式 #
多工具协作 #
python
@tool
def search_database(query: str) -> str:
"""搜索数据库"""
return f"数据库结果: {query}"
@tool
def search_api(query: str) -> str:
"""调用 API"""
return f"API 结果: {query}"
@tool
def combine_results(db_result: str, api_result: str) -> str:
"""合并多个来源的结果"""
return f"合并结果:\n数据库: {db_result}\nAPI: {api_result}"
tools = [search_database, search_api, combine_results]
agent = create_react_agent(model, tools)
带记忆的代理 #
python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent
checkpointer = MemorySaver()
agent = create_react_agent(
model,
tools,
checkpointer=checkpointer
)
# 使用会话 ID 保持记忆
config = {"configurable": {"thread_id": "user-123"}}
result1 = agent.invoke(
{"messages": [("user", "我叫张三")]},
config=config
)
result2 = agent.invoke(
{"messages": [("user", "我叫什么？")]},
config=config
)
人工确认 #
python
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt
@tool
def send_email_with_confirmation(to: str, subject: str, body: str) -> str:
"""发送邮件（需要人工确认）"""
# 中断等待人工确认
approved = interrupt({
"question": f"确认发送邮件至 {to}？",
"subject": subject,
"body": body
})
if approved:
return f"邮件已发送至 {to}"
else:
return "邮件发送已取消"
checkpointer = MemorySaver()
agent = create_react_agent(model, [send_email_with_confirmation], checkpointer=checkpointer)
# 运行到需要确认时暂停
result = agent.invoke(
{"messages": [("user", "发送邮件给 test@example.com")]},
config={"configurable": {"thread_id": "session-1"}}
)
# 人工确认后继续
result = agent.invoke(
{"messages": [("user", "确认")]}, # 或 "取消"
config={"configurable": {"thread_id": "session-1"}}
)
text
┌─────────────────────────────────────────────────────────────┐
│ 人工确认流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户: "发送邮件给 test@example.com" │
│ │ │
│ ▼ │
│ Agent: 选择 send_email_with_confirmation 工具 │
│ │ │
│ ▼ │
│ ┌─────────────────────┐ │
│ │ 等待人工确认 │ │
│ │ "确认发送邮件？" │ │
│ └──────────┬──────────┘ │
│ │ │
│ ┌──────┴──────┐ │
│ │ │ │
│ 确认 取消 │
│ │ │ │
│ ▼ ▼ │
│ 发送邮件 取消发送 │
│ │
└─────────────────────────────────────────────────────────────┘
完整示例 #
智能助手 #
python
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from datetime import datetime
model = ChatOpenAI(model="gpt-4o-mini")
@tool
def get_current_time() -> str:
"""获取当前时间"""
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@tool
def get_weather(city: str) -> str:
"""获取城市天气"""
weather_data = {
"北京": "晴，25°C",
"上海": "多云，28°C",
"广州": "雨，30°C"
}
return weather_data.get(city, f"未找到{city}的天气")
@tool
def calculate(expression: str) -> str:
"""计算数学表达式"""
try:
result = eval(expression)
return str(result)
except Exception as e:
return f"计算错误: {e}"
@tool
def search_web(query: str) -> str:
"""搜索网络"""
return f"关于'{query}'的搜索结果..."
@tool
def create_reminder(task: str, time: str) -> str:
"""创建提醒"""
return f"已创建提醒: {task} 于 {time}"
tools = [get_current_time, get_weather, calculate, search_web, create_reminder]
agent = create_react_agent(model, tools)
# 测试
queries = [
"现在几点了？",
"北京今天天气怎么样？",
"计算 123 * 456",
"搜索 LangChain 教程",
"提醒我下午3点开会"
]
for query in queries:
result = agent.invoke({"messages": [("user", query)]})
print(f"问题: {query}")
print(f"回答: {result['messages'][-1].content}")
print("-" * 50)
调试与监控 #
详细输出 #
python
# AgentExecutor 方式
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True # 打印详细执行过程
)
# LangGraph 方式
result = agent.invoke(
{"messages": [("user", "问题")]},
config={"callbacks": [handler]}
)
使用 LangSmith #
python
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-key"
# 所有执行会自动记录到 LangSmith
最佳实践 #
1. 清晰的工具描述 #
python
@tool
def search_product(query: str, category: str = None) -> str:
"""搜索产品信息
根据关键词搜索产品数据库。
可以指定类别进行过滤。
Args:
query: 搜索关键词，如产品名称或描述
category: 可选，产品类别，如 '电子产品'、'服装'
Returns:
匹配的产品列表
"""
pass
2. 错误处理 #
python
@tool
def api_call(endpoint: str) -> str:
"""调用 API"""
try:
# API 调用逻辑
return "成功"
except Exception as e:
return f"API 调用失败: {str(e)}"
3. 限制迭代次数 #
python
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
max_iterations=5, # 最大迭代次数
handle_parsing_errors=True # 处理解析错误
)
下一步 #
最后更新：2026-03-30