作者:互联网 时间: 2026-07-20 08:20:14
在使用 LangChain 时,model 和 agent 都可以使用 invoke 方法,但它们在定位、功能和底层执行逻辑上有本质区别:

model.invoke("你好") 发起一次网络请求,获取文本回复后立刻结束,速度极快。返回通常是单一的 AIMessage。大模型(大脑) + 工具(手脚) + 思考逻辑(工作流)。agent.invoke(...) 背后可能会自主发起多次网络请求(思考 -> 拦截并调工具 -> 观察返回结果 -> 再思考),速度视任务复杂度而定。返回结果通常是一个包含完整对话轨迹的状态数据结构。model 作为大脑,还需要传入 tools 数组来赋予其执行特定动作的能力。在使用 Agent 之前,我们需要准备好大模型实例。在后续所有示例中,我们默认使用以下配置的模型作为基础:
复制代码import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from rich import print as rprintload_dotenv(override=True)# 1. 准备大模型实例
model = init_chat_model(
model="gpt-5.4-mini",
model_provider="openai",
api_key=os.getenv("CLOSEAI_API_KEY"),
base_url=os.getenv("CLOSEAI_BASE_URL")
)
如果不给 Agent 绑定工具,它的能力类似于直接调用大模型,但返回的数据结构是一套完整的图状态(State)。
复制代码# 2. 创建 Agent,仅传入 model
agent = create_agent(
model=model
)# 3. 调用 Agent
response = agent.invoke({
"messages": [
{"role": "system", "content": "你是一个精通数学的老师,擅长以通俗易懂的方式讲解数学问题"},
{"role": "user", "content": "100 + 20 * 3 = ?"}
]
})rprint(response)
我们可以通过 @tool 装饰器将普通的 Python 函数转换为 LangChain 可识别的工具,然后传递给 Agent。
复制代码from langchain_core.tools import tool# 1. 定义一个自定义工具
@tool
def get_weather(city: str) -> str:
"""查询指定城市的天气。"""
# 模拟真实的天气 API 调用
return f"{city}天气晴朗,温度是15℃"# 2. 创建 Agent,将模型和自定义工具绑定
agent = create_agent(
model=model,
tools=[get_weather] # 传入工具列表
)# 3. 调用 Agent
response = agent.invoke({
"messages": [
{"role": "user", "content": "请问北京天气怎么样?"}
]
})rprint(response) # 模型会自动决定去调用 get_weather 工具
LangChain 官方社区提供了非常多开箱即用的工具,例如 TavilySearch,可以轻松让 Agent 具备联网搜索的能力。
复制代码from langchain_tavily import TavilySearch# 1. 实例化内置的联网搜索工具
web_search = TavilySearch(
max_results=2,
tavily_api_key=os.getenv("TAVILY_API_KEY"),
)# 2. 创建 Agent,绑定内置工具
agent = create_agent(
model=model,
tools=[web_search]
)# 3. 调用 Agent
response = agent.invoke({
"messages": [
{"role": "user", "content": "请帮我查询2024年诺贝尔物理学奖得主是谁?"}
]
})rprint(response) # 模型会自动使用搜索引擎工具获取最新信息
Agent 可以同时持有多个不同的工具。模型在遇到复杂问题时,会根据 Prompt 和工具的描述,自主决策使用哪个工具,甚至会组合调用多个工具来分步解决问题。
复制代码# 1. 准备一个工具组合列表
tools = [get_weather, web_search]# 2. 创建 Agent,同时赋予多个工具
agent = create_agent(
model=model,
tools=tools
)# 3. 调用 Agent
response = agent.invoke({
"messages": [
{"role": "user", "content": "北京天气如何?另外,帮我搜一下最新的科技头条新闻。"}
]
})rprint(response) # 模型会先后调用天气工具和搜索工具,最后总结回答
name)在创建 Agent 时,可以通过 name 参数为其命名。这在多 Agent 协作或调试时非常有用,能在日志中清晰区分当前执行任务的 Agent。
复制代码agent = create_agent(
model=model,
name="agent01" # 为 Agent 命名
)
system_prompt 快捷参数create_agent 提供了一个快捷参数 system_prompt,可以直接为其赋予“人设”:
复制代码agent = create_agent(
model=model,
system_prompt="你是一名多才多艺的智能助手,可以调用工具帮助用户解决问题。"
)
ChatPromptTemplate 动态注入变量 (Model 与 Agent 对比)对于复杂的场景,我们需要使用 ChatPromptTemplate 动态生成提示词。Model 和 Agent 都可以完美结合模板使用,但由于底层的输入要求不同,写法上略有差异:
对于裸 Model (大脑直接接收提示词):
Model 能够直接接收并理解 PromptValue 对象,所以我们可以分步直接传给它:
复制代码from langchain_core.prompts import ChatPromptTemplateprompt = ChatPromptTemplate.from_messages([
("system", "你是一个精通【{subject}】的智能助手。"),
("user", "我的问题是:{question}")
])# 1. 填入变量,生成提示词对象 (PromptValue)
rendered_prompt = prompt.invoke({
"subject": "物理学",
"question": "相对论是什么?"
})# 2. 直接把提示词对象传给 Model
response = model.invoke(rendered_prompt)
对于 Agent (项目经理需要固定格式的状态字典):
Agent 本质上是一个状态机(比如 LangGraph),它强制要求输入的数据必须是带有 "messages" 键的字典。因此我们需要手动分步进行格式转换:
复制代码# 1. 填入变量,生成提示词对象
rendered_prompt = prompt.invoke({
"subject": "物理学",
"question": "2024年诺贝尔奖得主是谁?"
})# 2. 把提示词对象转换成标准的“消息列表”
messages_list = rendered_prompt.to_messages()# 3. 包装成带有 "messages" 键的字典,最后交给 Agent 执行
response = agent.invoke({
"messages": messages_list
})
控制工具函数的输入(即如何约束工具传参),对于 Model 和 Agent 是一模一样的:都是靠 Pydantic、函数注释(docstring)和类型提示。
但在控制最终的结构化输出上,Agent 因为底层占用了工具调用,不能像裸 Model 一样简单粗暴地绑定 .with_structured_output()。在最新的 LangChain 中,官方为 Agent 提供了 4 种结构化输出策略 (response_format):
ProviderStrategy;如果不兼容,就自动降级用 ToolStrategy。配置示例:
复制代码from langchain_core.messages import SystemMessage
from pydantic import BaseModel, Field
from typing import Literal
from langchain.agents import create_agent
from langchain.agents.structured_output import (
ProviderStrategy,
ToolStrategy,
AutoStrategy
)
from langchain.tools import tool# 1. 定义日常业务工具
@tool
def search_customer_database(query: str) -> str:
"""在客户数据库中搜索信息"""
if "张三" in query:
return "客户记录:张三,VIP客户,最近购买日期:2026-01-15,累计消费:$15,000"
return "无记录"@tool
def send_email(customer: str) -> str:
"""发送感谢邮件"""
return f"已向 {customer} 发送感谢邮件"# 2. 定义我们最终想要的结构化数据 (Pydantic Schema)
class CustomerAnalysis(BaseModel):
"""客户分析报告"""
customer_name: str = Field(None, description="客户姓名")
customer_tier: Literal["潜在客户", "普通客户", "VIP客户", "流失风险"] = Field("潜在客户", description="客户等级")
recent_activity: str = Field(None, description="最近活动")
send_email: bool = Field(False, description="是否已发送感谢邮件")# 3. 创建 Agent
agent = create_agent(
model=model,
tools=[search_customer_database, send_email],
# 强大的 System Prompt:指导 Agent 按既定 SOP 工作
system_prompt=SystemMessage(content=
"请分析指定客户的情况:n"
"1. 先搜索客户数据库了解最新情况n"
"2. 如果是VIP客户,则立刻调用发送感谢邮件工具n"
"3. 基于搜索结果生成最终的结构化分析报告"
),
# 策略 1 (最推荐): 自动判断模型能力,智能选择策略
response_format=AutoStrategy(CustomerAnalysis)
# 策略 2 (手动指定): response_format=ToolStrategy(CustomerAnalysis)
# 策略 3 (手动指定): response_format=ProviderStrategy(CustomerAnalysis)
# 策略 4 (不推荐): response_format=CustomerAnalysis
)response = agent.invoke({"messages": [{"role": "user", "content": "请分析客户张三"}]})
# 解析好的 Pydantic 对象会保存在 structured_response 字段中
print(response["structured_response"])
handle_tool_error)在给 Agent 绑定工具(或使用结构化输出)时,经常会遇到模型传参错误或工具内部报错的情况。我们可以通过为工具配置错误处理参数,让 Agent 拥有自动纠错的能力。
注意命名差异:
@tool)中,该参数名为 handle_tool_error。ToolStrategy 等策略中,为了统一,该参数被命名为 handle_errors。
两者在底层的作用完全一致。不论是普通工具里的 handle_tool_error,还是结构化输出里的 handle_errors,该参数都支持以下几种配置方式:
True (默认行为):当工具报错时,程序不会崩溃。LangChain 会捕获真正的 Python 报错信息(如 ValueError),并将报错文本发还给大模型,让大模型自我反思并重试调用。False:遇到错误不拦截,直接让 Python 程序抛出异常并终止。代码示例(结构化输出场景):
复制代码from pydantic import BaseModel, Field
from typing import Union
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
# 假设 model 已经初始化
# model = init_chat_model(...)class ContactInfo(BaseModel):
"""个人联系信息"""
name: str = Field(description="姓名")
email: str = Field(description="电子邮箱")class EventDetails(BaseModel):
"""活动详情"""
event_name: str = Field(description="活动名称")
date: str = Field(description="活动日期")agent = create_agent(
model=model,
response_format=ToolStrategy(
Union[ContactInfo, EventDetails],
tool_message_content="提取完成!",
handle_errors=True # 核心:设置为 True 以开启自动重试
)
)result = agent.invoke({
"messages": [{
"role": "user",
"content": "请提取以下文本中内容:姓名:张三,电子邮箱:[email protected],活动名称:公司年会,活动日期:2026-07-15"
}]
})
print(result["structured_response"])
这是实战中非常实用的技巧,主要适用于以下场景:
代码示例(普通工具场景):
复制代码from langchain.tools import tool
from langchain.agents import create_agent
# 假设 model 已经初始化
# model = init_chat_model(...)# 核心:使用 fixed string 向大模型隐藏真实的 ValueError,并精准引导纠错
@tool(handle_tool_error="参数错误!订单号必须是纯数字,不要带字母或特殊字符!请检查后重试!")
def get_order_status(order_id: int) -> str:
"""查询订单状态。必须传入纯数字的订单号。"""
# 模拟业务逻辑:如果大模型提取了字母,类型不匹配会抛出 ValueError 或 TypeError。
# 因为配置了 handle_tool_error,程序不会崩溃,而是把上述报错信息替换为固定字符串发回给模型。
if int(order_id) == 12345:
return "订单 12345 状态:已发货"
return "订单不存在"agent = create_agent(
model=model,
tools=[get_order_status]
)# 故意给一段带有干扰的文本
response = agent.invoke({
"messages": [{"role": "user", "content": "帮我查一下订单状态,订单号好像是 O-12345。"}]
})
# 大模型第一次很可能会把 "O-12345" 当作参数传进去导致报错。
# 随后它会收到固定字符串提示,从而自我纠正提取出正确的 12345 再试一次。
print(response["messages"][-1].content)
例如 handle_tool_error=ValueError。这意味着只有发生 ValueError 时才让模型重试,如果发生断网(NetworkError)等其他类型的错误,程序该崩溃还是崩溃。
支持传入一个自定义的 Python 函数,接收 Exception 对象,返回给模型一段特定的错误提示字符串:
复制代码def my_custom_handler(error: Exception) -> str:
# 可以在这里记录日志、发报警
return f"模型老兄,你犯了 {type(error)} 类型的错误,仔细检查一下哦!"@tool(handle_tool_error=my_custom_handler)
def some_tool(param: str) -> str:
pass
在实际业务中(特别是网页对话框),Agent 执行任务往往需要耗费较长时间。为了避免前端长时间“假死”无响应,我们需要使用 agent.stream() 来实时输出 Agent 的思考和执行进度。
在最新的 LangChain/LangGraph 架构中,stream_mode 支持多达 7 种模式,满足从前端展示到后端调试的各种需求。
messages 模式 (最核心,打字机效果的唯一选择):
values 模式 (全局快照):
updates 模式 (增量补丁 - 开发者最爱):
values 完全一样,但它极其精简,只打印当前这一步操作带来的增量变化(例如刚刚新增的那一条消息)。checkpoints 模式 (持久化存档):
(注:除此之外还包含 tasks、debug 和 custom 模式,一般仅用于底层极限调试或定制极其细粒度的进度条推送)
messages 模式)下面展示如何捕获 messages 模式下的流式 Token,实现最经典的打字机效果:
复制代码# 假设 agent 已创建,并配备了某些工具
# agent = create_agent(model=model, tools=[...])# 调用 stream,并指定 stream_mode="messages"
stream = agent.stream(
{"messages": [{"role": "user", "content": "请查一下北京天气,然后根据天气写一首诗"}]},
stream_mode="messages"
)# 遍历流式产生的数据块 (chunk)
# 这里的 chunk 包含了 (message_chunk, metadata) 两个元素
for chunk in stream:
# 取出真正的内容片段
msg_chunk = chunk[0]
# 1. 打印普通的文本内容 (比如大模型正在作诗)
if msg_chunk.content:
print(msg_chunk.content, end="", flush=True)
# 2. 如果你想监听大模型准备调什么工具,可以打印这段底层的 tool_calls 字典流
elif msg_chunk.tool_call_chunks:
print(f"n[系统提示:大模型正在疯狂敲代码准备调工具 {msg_chunk.tool_call_chunks[0]['name']}...]n")
学完前面的所有高级概念后,我们可以把它们全部拼装在一起,打造一个极具工程水准的 Agent。
这个终极例子完美融合了以下 6 大核心概念:
name="xxx"),方便日志调试。ChatPromptTemplate),实现动态变量注入和系统人设分离。prompt.invoke() 渲染变量生成消息,再直接传给 Agent,拒绝晦涩难懂的 LCEL 语法。@tool(handle_tool_error=...)),利用固定字符串精准引导纠错。AutoStrategy + handle_errors=True),强制约束最终输出并自带重试光环。stream_mode="messages"),实现丝滑的打字机前端效果。 复制代码import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain.agents import create_agent
from langchain.agents.structured_output import AutoStrategy
from pydantic import BaseModel, Fieldload_dotenv(override=True)# ================= 1. 初始化模型 =================
model = init_chat_model(
model="gpt-4o-mini",
model_provider="openai",
api_key=os.getenv("CLOSEAI_API_KEY"),
base_url=os.getenv("CLOSEAI_BASE_URL")
)# ================= 2. 定义业务工具 (带精准容错机制) =================
@tool(handle_tool_error="参数错误!天气工具只能接受纯中文城市名,请不要带英文或拼音,重试!")
def get_weather(city: str) -> str:
"""查询国内城市的天气状况"""
# 模拟业务容错拦截:如果大模型提取了 "Beijing" 而非 "北京",抛出异常交由外层纠错
import re
if re.search(r'[a-zA-Z]', city):
raise ValueError("Invalid City Name")
return f"{city}的天气是:晴朗,25度"@tool(handle_tool_error="表达式错误!请检查你的数学公式格式是否支持 eval。")
def calculate_math(expression: str) -> str:
"""计算数学表达式的值,如 '2 + 2'"""
return str(eval(expression))# ================= 3. 定义结构化输出模型 (Pydantic) =================
class FinalResponse(BaseModel):
"""智能助手的最终格式化回答"""
thought_process: str = Field(description="详细的推理和思考过程")
final_answer: str = Field(description="直接给用户的最终答案")
confidence_score: int = Field(description="对答案的自信度打分(0-100)")# ================= 4. 构建 Agent 引擎 =================
# 将多个特性融为一体:命名、挂载工具、设置强约束的结构化策略并开启自动重试
ultimate_agent = create_agent(
model=model,
tools=[get_weather, calculate_math],
name="SuperAssistant",
response_format=AutoStrategy(
schema=FinalResponse,
handle_errors=True # 强力兜底:格式不对自己重做
)
)# ================= 5. 定义动态提示词模板 =================
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个全能的智能助理,名叫 {bot_name}。请尽力解答用户的问题。"),
("user", "{user_input}")
])# ================= 6. 渲染提示词模板 =================
print("正在连线 SuperAssistant...")# 先用模板填入动态变量,生成最终的消息列表
messages = prompt.invoke({
"bot_name": "贾维斯",
"user_input": "Beijing现在的天气怎么样?帮我把它的温度乘以 5 告诉我结果。"
}).to_messages()# ================= 7. 触发流式运行 (打字机效果) =================
# 把消息列表放进字典的 "messages" 字段,直接传给 agent 开启流式监听
stream = ultimate_agent.stream(
{"messages": messages},
stream_mode="messages"
)# 实时捕获并打印 Token 块
for chunk in stream:
msg = chunk[0]
# 如果大模型在思考时生成了普通文本内容
if msg.content:
print(msg.content, end="", flush=True)
# 如果大模型决定调用工具,把动作打印出来
elif msg.tool_call_chunks:
print(f"n[动作] -> 准备调用工具: {msg.tool_call_chunks[0]['name']} ...")