Agent
2026/2/21大约 14 分钟
Agent
Agent 是 LangChain 最强大的特性之一，它让 LLM 能够自主决策、循环思考、调用工具，从"单次问答"进化为"自主解决复杂任务"。
一、Agent 的本质
1.1 什么是 Agent
普通 LLM 调用是单向执行：问题 → LLM → 答案，结束。
Agent 是循环决策系统：
用户提问
│
▼
┌──────────────────┐
│ Agent 循环开始 │
│ │
│ LLM 思考： │
│ 1. 需要工具吗？ │◄──┐
│ 2. 用哪个工具？ │ │
│ 3. 参数是什么？ │ │
│ │ │
│ 执行工具 → 观察结果 │ │
│ │ │
│ 判断：够了吗？ │ │
│ - 不够 ──────────┘ │
│ - 够了 ↓ │
└──────────────────┘
│
▼
最终答案💡 核心区别：Agent 可以自己决定"做什么、怎么做、做几次"，而不是一次性完成所有事。
1.2 Agent 的三个核心要素
| 要素 | 说明 |
|---|---|
| LLM（大脑） | 负责推理、决策、判断 |
| Tools（手臂） | 外部能力：搜索、计算、查库、调 API |
| Memory（记忆） | 历史对话、中间步骤、工具结果 |
二、Agent 的工作原理
2.1 ReAct 范式（Reasoning + Acting）
最经典的 Agent 模式，交替进行"推理"和"行动"：
Thought（思考）: 我需要先查询当前天气
Action（行动）: get_weather
Action Input: {"city": "北京"}
Observation（观察）: 北京今天晴，22°C
Thought: 现在我知道温度了，可以给出穿衣建议
Action: None
Final Answer: 北京今天 22°C，建议穿薄外套或长袖 T 恤...2.2 执行流程图
┌─────────────────────────────────────────┐
│ AgentExecutor │
│ │
│ 1. 接收用户输入 │
│ ↓ │
│ 2. 构建 Prompt（含工具描述 + 历史） │
│ ↓ │
│ 3. LLM 推理 → 输出 Action │
│ ↓ │
│ 4. 解析 Action，执行对应 Tool │
│ ↓ │
│ 5. 把 Observation 追加到历史 │
│ ↓ │
│ 6. 判断：是 Final Answer 吗？ │
│ ├─ 是 → 返回答案，结束 │
│ └─ 否 → 回到步骤 2（继续循环） │
│ │
│ ⚠️ 防护：max_iterations 防死循环 │
└─────────────────────────────────────────┘三、Agent 的类型
LangChain 提供多种 Agent 类型，适用不同场景。
3.1 类型对比表
| Agent 类型 | 驱动方式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| create_react_agent | 文本格式解析 | 通用性强，支持所有 LLM | 解析易出错 | 通用对话 Agent |
| create_tool_calling_agent | Function Calling | 稳定性好，格式标准 | 需模型支持 | GPT-4/Claude 等现代模型 |
| create_structured_chat_agent | 结构化输出 | 支持多输入工具 | Prompt 较长 | 复杂工具场景 |
| create_openai_functions_agent | OpenAI Functions | OpenAI 原生支持 | 绑定 OpenAI | OpenAI 专用 |
| create_json_agent | JSON 输出 | 结构化解析 | 灵活性低 | 严格格式场景 |
📌 推荐选择：新项目优先用
create_tool_calling_agent（GPT-4o/Claude）或create_react_agent（通用模型）。
四、创建第一个 Agent
4.1 准备工具
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""获取指定城市的当前天气。
Args:
city: 城市名称，如"北京"、"上海"
"""
# 实际项目中调用天气 API
weather_map = {
"北京": "晴天，22°C",
"上海": "多云，18°C",
"深圳": "阴天，25°C",
}
return weather_map.get(city, f"{city}的天气信息未找到")
@tool
def calculator(expression: str) -> str:
"""计算数学表达式的值。
Args:
expression: 数学表达式字符串，如 "2 + 3 * 4"
"""
try:
result = eval(expression)
return f"计算结果：{result}"
except Exception as e:
return f"计算错误：{str(e)}"
tools = [get_weather, calculator]4.2 创建 Tool Calling Agent（推荐）
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个helpful助手，可以使用工具来回答问题。
工具使用原则：
1. 当需要实时数据或计算时，使用相应工具
2. 如果工具返回"未找到"，告诉用户你没有该信息
3. 给出最终答案前，确保已获取所有必要信息"""),
MessagesPlaceholder("chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad"), # Agent 的思考过程记录
])
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # 打印详细执行过程
max_iterations=5, # 最多循环 5 次
max_execution_time=30, # 最多执行 30 秒
handle_parsing_errors=True,# 自动处理解析错误
)
# 调用
result = agent_executor.invoke({"input": "北京今天天气怎么样？顺便帮我算一下 123 * 456"})
print(result["output"])执行过程（verbose=True 时打印）：
> Entering new AgentExecutor chain...
Invoking: `get_weather` with `{'city': '北京'}`
晴天，22°C
Invoking: `calculator` with `{'expression': '123 * 456'}`
计算结果：56088
北京今天是晴天，气温 22°C。123 乘以 456 等于 56088。
> Finished chain.4.3 创建 ReAct Agent（兼容性更好）
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub
# 从 hub 拉取标准 ReAct prompt 模板
react_prompt = hub.pull("hwchase17/react")
agent = create_react_agent(
llm=llm,
tools=tools,
prompt=react_prompt,
)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5,
)
result = agent_executor.invoke({"input": "先查北京天气，再根据温度推荐穿衣"})五、Agent 与记忆结合
Agent 默认是无状态的，每次调用相互独立。加入记忆后可实现多轮对话。
5.1 基于 RunnableWithMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
# 存储：session_id → 历史记录
store = {}
def get_session_history(session_id: str):
if session_id not in store:
store[session_id] = ChatMessageHistory()
return store[session_id]
agent_with_memory = RunnableWithMessageHistory(
agent_executor,
get_session_history,
input_messages_key="input",
history_messages_key="chat_history",
)
# 多轮对话
def chat(message: str, session_id: str = "default"):
result = agent_with_memory.invoke(
{"input": message},
config={"configurable": {"session_id": session_id}}
)
return result["output"]
# 第一轮
print(chat("我叫小明，帮我查北京天气"))
# → 北京今天晴天 22°C
# 第二轮（Agent 会记住你的名字）
print(chat("我刚才叫你查哪个城市的天气？"))
# → 你让我查了北京的天气
# 第三轮
print(chat("帮我算一下 100 + 200"))
# → 计算结果是 3005.2 手动管理历史（更灵活）
from langchain_core.messages import HumanMessage, AIMessage
chat_history = []
def chat_manual(user_input: str) -> str:
result = agent_executor.invoke({
"input": user_input,
"chat_history": chat_history,
})
# 追加到历史
chat_history.append(HumanMessage(content=user_input))
chat_history.append(AIMessage(content=result["output"]))
return result["output"]六、高级 Agent 模式
6.1 Plan-and-Execute Agent：规划执行分离
先一次性制定完整计划，再逐步执行，适合复杂多步骤任务。
from langchain_experimental.plan_and_execute import (
PlanAndExecute,
load_agent_executor,
load_chat_planner,
)
# 规划器（用 LLM 制定计划）
planner = load_chat_planner(llm)
# 执行器（按计划执行工具）
executor = load_agent_executor(llm, tools, verbose=True)
agent = PlanAndExecute(planner=planner, executor=executor, verbose=True)
result = agent.invoke({"input": "查询北京、上海、深圳的天气，然后告诉我哪个城市最适合旅游"})执行过程：
[规划阶段]
Plan:
1. 查询北京天气
2. 查询上海天气
3. 查询深圳天气
4. 比较温度和天气状况
5. 给出推荐
[执行阶段]
Step 1: 调用 get_weather("北京") → 晴天，22°C
Step 2: 调用 get_weather("上海") → 多云，18°C
Step 3: 调用 get_weather("深圳") → 阴天，25°C
Step 4-5: 综合分析...6.2 Self-Ask Agent：递归提问
遇到复杂问题时，自动拆解成子问题，逐个回答。
from langchain.agents import create_self_ask_with_search_agent
# 需要一个搜索工具
from langchain_community.tools import DuckDuckGoSearchRun
search = DuckDuckGoSearchRun()
agent = create_self_ask_with_search_agent(
llm=llm,
tools=[search],
verbose=True,
)
agent_executor = AgentExecutor(agent=agent, tools=[search], verbose=True)
result = agent_executor.invoke({"input": "谁是《三体》的作者？他的代表作有哪些？"})执行过程：
Are follow up questions needed: Yes
Follow up: 谁是《三体》的作者？
Intermediate answer: 刘慈欣
Follow up: 刘慈欣的代表作有哪些？
Intermediate answer: 《三体》三部曲、《流浪地球》、《球状闪电》...
Final Answer: 《三体》的作者是刘慈欣，他的代表作包括...6.3 Tool Retrieval：动态选择工具
当工具数量很多（50+）时，把所有工具描述塞进 prompt 会超出 context window。解决方案：先检索最相关的工具，再传给 Agent。
from langchain.agents import create_tool_calling_agent
from langchain.tools.retriever import create_retriever_tool
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
# 假设有 100 个工具
all_tools = [tool1, tool2, ..., tool100]
# 将工具描述向量化存入向量库
tool_descriptions = [
f"{tool.name}: {tool.description}" for tool in all_tools
]
vectorstore = Chroma.from_texts(
tool_descriptions,
OpenAIEmbeddings(),
metadatas=[{"tool_name": tool.name} for tool in all_tools]
)
def get_relevant_tools(query: str, k: int = 5):
"""根据问题检索最相关的 k 个工具"""
results = vectorstore.similarity_search(query, k=k)
tool_names = [doc.metadata["tool_name"] for doc in results]
return [tool for tool in all_tools if tool.name in tool_names]
# Agent 每次只加载相关工具
def agent_with_dynamic_tools(user_input: str):
relevant_tools = get_relevant_tools(user_input)
agent = create_tool_calling_agent(llm, relevant_tools, prompt)
executor = AgentExecutor(agent=agent, tools=relevant_tools)
return executor.invoke({"input": user_input})七、多 Agent 系统
多个专业 Agent 协作，每个负责一个领域。
7.1 简单协作：主 Agent + 专业 Agent
# 专业 Agent 1：数据分析专家
data_tools = [sql_query_tool, data_visualization_tool]
data_agent = create_tool_calling_agent(llm, data_tools, data_prompt)
data_executor = AgentExecutor(agent=data_agent, tools=data_tools)
# 专业 Agent 2：文案专家
writing_tools = [grammar_check_tool, seo_optimize_tool]
writing_agent = create_tool_calling_agent(llm, writing_tools, writing_prompt)
writing_executor = AgentExecutor(agent=writing_agent, tools=writing_tools)
# 包装成工具，供主 Agent 调用
@tool
def analyze_data(query: str) -> str:
"""数据分析专家，可查询数据库、生成图表"""
result = data_executor.invoke({"input": query})
return result["output"]
@tool
def write_content(topic: str) -> str:
"""文案专家，可撰写、优化文章"""
result = writing_executor.invoke({"input": topic})
return result["output"]
# 主 Agent（协调者）
coordinator_tools = [analyze_data, write_content]
coordinator = create_tool_calling_agent(llm, coordinator_tools, coordinator_prompt)
main_agent = AgentExecutor(agent=coordinator, tools=coordinator_tools)
result = main_agent.invoke({
"input": "分析上个月的销售数据，然后写一篇数据报告"
})7.2 LangGraph 实现复杂协作（更强大）
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
# 定义共享状态
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
current_task: str
data_result: str
writing_result: str
# 节点函数
def data_agent_node(state: AgentState):
result = data_executor.invoke({"input": state["current_task"]})
return {"data_result": result["output"]}
def writing_agent_node(state: AgentState):
prompt = f"根据以下数据分析结果，写一篇报告：\n{state['data_result']}"
result = writing_executor.invoke({"input": prompt})
return {"writing_result": result["output"]}
def supervisor_node(state: AgentState):
"""主管决定：数据分析完了吗？需要继续写作吗？"""
if not state.get("data_result"):
return "data_agent"
elif not state.get("writing_result"):
return "writing_agent"
else:
return END
# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("data_agent", data_agent_node)
workflow.add_node("writing_agent", writing_agent_node)
workflow.set_entry_point("supervisor")
workflow.add_conditional_edges(
"supervisor",
lambda x: x,
{
"data_agent": "data_agent",
"writing_agent": "writing_agent",
END: END
}
)
workflow.add_edge("data_agent", "supervisor")
workflow.add_edge("writing_agent", "supervisor")
app = workflow.compile()
result = app.invoke({
"messages": [],
"current_task": "分析上个月销售数据并写报告",
})八、Agent 的控制与安全
8.1 限制执行时间和次数
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
max_iterations=10, # 最多循环 10 次
max_execution_time=60, # 最多执行 60 秒
early_stopping_method="generate", # 超时后生成部分答案
)8.2 工具权限控制
@tool
def delete_database(table_name: str) -> str:
"""删除数据库表（危险操作）"""
# ⚠️ 生产环境必须做权限校验
if not user_has_permission("delete_table"):
return "权限不足，无法执行删除操作"
# 执行删除...
return f"表 {table_name} 已删除"
# 只给特定 Agent 危险工具
safe_tools = [search, calculator]
admin_tools = [search, calculator, delete_database]
user_agent = create_tool_calling_agent(llm, safe_tools, prompt)
admin_agent = create_tool_calling_agent(llm, admin_tools, prompt)8.3 输入/输出验证
from pydantic import BaseModel, Field, validator
class SafeCalculatorInput(BaseModel):
expression: str = Field(description="数学表达式")
@validator("expression")
def validate_expression(cls, v):
# 白名单：只允许数字和基本运算符
allowed = set("0123456789+-*/()., ")
if not all(c in allowed for c in v):
raise ValueError("表达式包含非法字符")
return v
@tool(args_schema=SafeCalculatorInput)
def safe_calculator(expression: str) -> str:
"""安全的计算器，防止代码注入"""
try:
result = eval(expression)
return f"结果：{result}"
except Exception as e:
return f"计算错误：{str(e)}"8.4 人工审批节点
@tool
def execute_with_approval(action: str, details: str) -> str:
"""需要人工审批的操作"""
print(f"\n[等待审批]")
print(f"操作：{action}")
print(f"详情：{details}")
approval = input("是否批准？(y/n): ")
if approval.lower() == 'y':
# 执行实际操作
return f"已执行：{action}"
else:
return "操作已被拒绝"九、Agent 调试与监控
9.1 逐步调试
from langchain.callbacks import StdOutCallbackHandler
# 详细打印每个工具调用
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
callbacks=[StdOutCallbackHandler()],
)
# 手动逐步执行
for step in agent_executor.iter({"input": "复杂任务"}):
print(f"\n[Step {step.get('step_number')}]")
print(f"Action: {step.get('action')}")
print(f"Observation: {step.get('observation')}")
input("按回车继续...")9.2 LangSmith 追踪（生产推荐）
import os
# 配置 LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-project"
# 正常使用，自动上传追踪数据到 LangSmith
result = agent_executor.invoke({"input": "..."})
# 在 LangSmith 网页端可以看到：
# - 每次工具调用的参数和返回值
# - LLM 的推理过程
# - 执行时间、token 消耗
# - 错误堆栈9.3 自定义 Callback
from langchain.callbacks.base import BaseCallbackHandler
class MyAgentCallback(BaseCallbackHandler):
def on_tool_start(self, serialized, input_str, **kwargs):
print(f"🔧 开始执行工具：{serialized.get('name')}")
print(f" 输入：{input_str}")
def on_tool_end(self, output, **kwargs):
print(f"✅ 工具执行完成")
print(f" 输出：{output}")
def on_tool_error(self, error, **kwargs):
print(f"❌ 工具执行失败：{error}")
def on_agent_action(self, action, **kwargs):
print(f"💭 Agent 决策：{action.tool}, {action.tool_input}")
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
callbacks=[MyAgentCallback()],
)十、实战案例：智能客服 Agent
综合运用工具、记忆、多步推理，构建一个完整的客服 Agent。
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
# ========== 工具定义 ==========
@tool
def query_order_status(order_id: str) -> str:
"""查询订单状态
Args:
order_id: 订单编号
"""
# 模拟查询数据库
orders = {
"12345": {"status": "已发货", "tracking": "SF1234567890"},
"67890": {"status": "处理中", "tracking": None},
}
order = orders.get(order_id)
if not order:
return f"订单 {order_id} 不存在"
if order["tracking"]:
return f"订单 {order_id} {order['status']}，快递单号：{order['tracking']}"
return f"订单 {order_id} {order['status']}"
@tool
def search_knowledge_base(question: str) -> str:
"""搜索知识库
Args:
question: 用户问题
"""
# 模拟向量检索
kb = {
"退货": "退货政策：7天内未使用商品可申请退货...",
"发票": "发票开具：订单完成后可在"我的订单"中申请电子发票...",
"积分": "积分规则：消费 1 元获得 1 积分，积分可抵扣...",
}
for key, value in kb.items():
if key in question:
return value
return "抱歉，知识库中没有找到相关信息"
@tool
def create_ticket(issue: str) -> str:
"""创建工单（人工客服介入）
Args:
issue: 问题描述
"""
ticket_id = "T" + str(hash(issue))[-6:]
return f"已为您创建工单 {ticket_id}，客服将在 2 小时内联系您"
# ========== Agent 配置 ==========
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [query_order_status, search_knowledge_base, create_ticket]
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个专业的在线客服助手，目标是快速解决用户问题。
工作流程：
1. 识别问题类型（订单查询/政策咨询/投诉建议）
2. 优先使用工具获取准确信息
3. 如果知识库和订单系统都无法解决，创建工单转人工
4. 保持礼貌、专业、高效
注意事项：
- 查询订单前，先确认用户提供了订单号
- 回答政策问题时，先搜索知识库
- 不要编造信息，不确定的内容明确告知"""),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5,
)
# ========== 记忆管理 ==========
store = {}
def get_session_history(session_id: str):
if session_id not in store:
store[session_id] = ChatMessageHistory()
return store[session_id]
agent_with_memory = RunnableWithMessageHistory(
agent_executor,
get_session_history,
input_messages_key="input",
history_messages_key="chat_history",
)
# ========== 对话函数 ==========
def chat_with_agent(message: str, session_id: str = "default") -> str:
result = agent_with_memory.invoke(
{"input": message},
config={"configurable": {"session_id": session_id}}
)
return result["output"]
# ========== 测试对话 ==========
if __name__ == "__main__":
print("客服 Agent 已启动，输入 'quit' 退出\n")
while True:
user_input = input("用户: ")
if user_input.lower() == 'quit':
break
response = chat_with_agent(user_input)
print(f"客服: {response}\n")测试对话：
用户: 你好，我想查一下订单状态
客服: 您好！请提供您的订单编号，我来帮您查询。
用户: 12345
客服: 您的订单 12345 已发货，快递单号：SF1234567890
用户: 你们的退货政策是什么？
客服: 我们的退货政策如下：7天内未使用商品可申请退货...
用户: 我的商品有质量问题，但是快递一直不来取件
客服: 非常抱歉给您带来不便。我已为您创建工单 T923841，专业客服将在 2 小时内联系您处理快递取件问题。十一、最佳实践与注意事项
11.1 Prompt 设计原则
# ✅ 好的 Agent Prompt
system_prompt = """你是数据分析助手。
能力：
- 查询数据库（query_database）
- 生成图表（create_chart）
- 统计分析（statistical_analysis）
工作流程：
1. 理解用户需求
2. 确定需要查询的数据
3. 调用工具获取数据
4. 分析并给出结论
限制：
- 只使用已提供的工具
- 不确定时明确说明，不要猜测
- 每次只调用一个工具"""
# ❌ 不好的 Prompt
system_prompt = "你是助手，帮用户做事" # 太模糊，Agent 不知道该干什么11.2 工具设计原则
# ✅ 好的工具设计
@tool
def get_user_info(user_id: str) -> str:
"""获取用户基本信息
Args:
user_id: 用户 ID，格式为 U + 6 位数字，如 U123456
Returns:
JSON 格式的用户信息：姓名、邮箱、注册时间
"""
# 清晰的描述、参数说明、返回格式
# ❌ 不好的工具设计
@tool
def do_something(data):
"""做一些事""" # 描述不清晰，Agent 不知道何时该用11.3 错误处理
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
handle_parsing_errors=True, # 自动处理解析错误
max_iterations=10,
return_intermediate_steps=True, # 返回中间步骤，便于调试
)
try:
result = agent_executor.invoke({"input": user_input})
except Exception as e:
# 记录详细错误日志
logger.error(f"Agent 执行失败：{str(e)}", exc_info=True)
# 给用户友好提示
return "抱歉，系统出现问题，请稍后重试或联系人工客服"11.4 性能优化
# 1. 使用流式输出，提升用户体验
for chunk in agent_executor.stream({"input": "复杂问题"}):
print(chunk, end="", flush=True)
# 2. 并行执行独立工具（需要自定义 Agent）
# 3. 缓存工具结果
from functools import lru_cache
@lru_cache(maxsize=100)
@tool
def expensive_api_call(param: str) -> str:
"""调用昂贵的外部 API（结果会缓存）"""
...十二、总结速查
Agent 核心流程：
思考 → 决策工具 → 执行 → 观察 → 判断是否继续 → 最终答案
Agent 类型选择：
GPT-4/Claude → create_tool_calling_agent（推荐）
通用模型 → create_react_agent
复杂任务 → Plan-and-Execute
递归问题 → Self-Ask
Agent 三要素：
LLM（大脑） → 推理、决策
Tools（手臂） → 执行外部能力
Memory（记忆） → 保持上下文
控制机制：
max_iterations → 防止死循环
max_execution_time → 限制总时长
handle_parsing_errors → 容错
人工审批 → 高危操作
调试工具：
verbose=True → 打印详细过程
LangSmith → 可视化追踪
自定义 Callback → 监控特定事件⚠️ 最后提醒：Agent 是强大的，但也引入了不确定性（LLM 可能做出错误决策）。生产环境务必做好错误处理、权限控制、监控告警。从简单场景开始，逐步增加复杂度。