LangChain
- langChain Site
- langChain Github
- langChain English Docs
- langChain Chinese Docs
- langChain API Search
- langChain vs langGraph vs Deep Agents
- langSmith Site
环境配置 & Start
PyCharm
anaconda
下载安装后需配置环境变量，在path下新增，也就是安装地址。环境变量配置完成之后通过命令conda --version查看是否配置成功
txt
D:\soft\Miniconda\install
D:\soft\Miniconda\install\Scripts
D:\soft\Miniconda\install\Library\binConda 基础命令对照表
| 命令 | 描述 |
|---|---|
| conda -help | 查看帮助 |
| conda info | 查看 conda 信息 |
| conda --version | 查看conda版本 |
| conda update conda | 更新 Conda（慎用） |
| conda create --name env_name python=3.8 | 创建新环境 |
| conda activate env_name | 激活环境 |
| conda deactivate | 退出当前环境 |
| conda remove --name env_name --all | 删除环境 |
| conda list | 列出当前环境包 |
| conda search package_name | 搜索包 |
| conda install package_name | 安装包 |
| conda update package_name | 更新包 |
| conda update conda | 更新conda |
| conda info --envs | 列出所有环境 |
| conda init | 初始化conda |
创建conda环境：LangChain 1.2版本要求Python版本为 3.10+以上 ，这里我们使用 python3.13.12 版本
cmd
# 查看anconda安装好的python环境
conda env list
# 创建一个名为langchain1.2的环境，指定Python版本是3.13.12
conda create --name langchain1.2 python=3.13.12
# 初始化虚拟环境 (执行完此指令，重新启动命令行窗口)
conda init
# 在命令行窗口切换到某python环境
conda activate langchain1.2
# 检查环境
python -V
# 退出环境
conda deactivate
# 删除环境
conda remove --name langchain1.2 --all下载langChain包，先用conda进行安装，实在不行就用pip安装
cmd
# 安装指定版本。比如1.2.2
conda install langchain==1.2.12
# 指定频道（如 conda-forge）
conda install -c conda-forge langchain==1.2.12
# 使用pip安装
pip install langchain==1.2.12
# 查看已安装的包
conda list
pip list在 PyCharm 当中进行环境配置，选择setting--> Python --> Interpreter，在这里可以看到当前项目指定的conda的虚拟环境以及所有安装的第三方依赖，直接输出尝试一下
- 如果是创建的新工程，在 Interpreter type 这里选择 Custom 自定义环境，下面就可以指定conda
py
import langchain
print(langchain.__version__)Model
先提前安装一下所有依赖
bash
pip install -r ./requirements.txt使用模型提供商库
LangChain为一些大模型供应商提供了专门的Model类，以deepseek为例，所需要的依赖包：langchain-deepseek、python-dotenv
- 支持的Model类
langchain-deepseek依赖于langchain-openai，安装前者，pip会自动从pypi拉取元数据解析依赖，后者也会被安装python-dotenv是用于环境管理的包- 创建
.env文件，填入deepseek的key和baseUrl - 可以不传api_key和base_url，在ChatDeepSeek类中会自动去env文件中找这两个配置
- 大多数API平台都支持
OpenAI API接口规范，所以基本都可以通过 ChatOpenAI 集成
env
DEEPSEEK_API_KEY=sk-crazyThursdayvme50
DEFAULT_API_BASE=https://api.deepseek.compython
import os
from dotenv import load_dotenv
from langchain_deepseek import ChatDeepSeek
from langchain_openai import ChatOpenAI
load_dotenv(override=True)
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEFAULT_API_BASE = os.getenv("DEFAULT_API_BASE")
# 可以不传api_key和base_url，在ChatDeepSeek类中会自动去env文件中找这两个配置
model = ChatDeepSeek(
model="deepseek-v4-flash",
# api_key=DEEPSEEK_API_KEY,
# base_url=DEFAULT_API_BASE
)
print(model.invoke("你是谁？"))python
# ChatZhipuAI 智谱 https://www.bigmodel.cn/
# ChatTongyi 阿里云百炼 https://bailian.console.aliyun.com/python
openAiModel = ChatOpenAI(
model="deepseek-v4-flash",
api_key=DEEPSEEK_API_KEY,
base_url=DEFAULT_API_BASE,
)
print(openAiModel.invoke("你是谁？"))init_chat_model
init_chat_model 是 LangChain 1.x 中推出的用于初始化聊天模型的统一接口。只要是LangChain支持的模型都可以处理，它会根据模型名称自动选择对应的模型类初始化实例
python
model = init_chat_model(
model="deepseek-v4-flash",
model_provider="deepseek",
# model="deepseek:deepseek-v4-flash",
api_key=DEEPSEEK_API_KEY,
base_url=DEFAULT_API_BASE,
)
print(model.invoke("你是谁？"))模型调用
提供了几种核心的调用方式，主要是invoke()、stream()和batch()方法，以及它们的异步版本ainvoke()、astream()和 abatch()
invoke()：阻塞式，一次性返回完整结果问答、批处理任务、无需实时反馈的场景ainvoke()：非阻塞式，提高系统吞吐量高并发Web应用、IO密集型任务stream()：流式输出，实时返回每个token聊天机器人、长文本生成、需要提升用户体验的交互应用asteam()：非阻塞式，提高系统吞吐量高并发Web应用、IO密集型任务batch()：批量处理多个输入高并发场景，需要同时处理大量请求abatch()：非阻塞式，提高系统吞吐量高并发Web应用、IO密集型任务
invoke
invoke的工作模式是阻塞式的，即程序会等待模型完全生成整个响应后，再一次性将结果返回给用户
参数说明
| 参数 | 类型 | 说明 | 是否必填 | 默认值 |
|---|---|---|---|---|
| input | string、list[dict]、list[Message] | 你要发送给模型的内容 | 是 | 无 |
| config | dict | 高级配置（回调函数、元数据、标签等) | 否 | None |
- 字典列表
- role角色说明：
system、user、assistant- system：设定 AI 的行为、角色、规则
- user：用户的输入/问题
- assistant：AI 的历史回复（用于对话上下文）
- role角色说明：
- 对象列表
- Message对象：
SystemMessage、HumanMessage、AIMessage
- Message对象：
返回值说明：返回一个AIMessage对象
| 参数 | 说明 | |||
|---|---|---|---|---|
| id | 模型生成的唯一标识 | |||
| model_provider | 模型供应商 | |||
| model_name | 使用的具体模型版本 | |||
| content | 模型生成的最终文本答案 | |||
| service_tier | 服务层级（如按量付费或订阅） | |||
| system_fingerprint | 系统指纹，用于追踪模型后端的配置变更 | |||
| finish_reason | 停止原因：stop（自然结束）、length（长度受限） | |||
| logprobs | 对数概率（通常用于分析词汇选择的可能性） | |||
| additional_kwargs | ||||
| refusal | 模型拒绝回答的情况（如触碰安全策略），None 表示正常回答 | |||
| response_metadata | 响应元数据（API 返回的详细原始数据） | |||
| token_usage | ||||
| completion_tokens | 生成回答消耗的 Token 数（输出） | |||
| prompt_tokens | 用户输入消耗的 Token 数（输入） | |||
| total_tokens | 本次交互总共消耗的 Token | |||
| completion_tokens_details | ||||
| accepted_prediction_tokens | 预测性生成的 Token 数 | |||
| audio_tokens | 音频生成消耗（如有） | |||
| reasoning_tokens | 推理模型思考过程消耗的 Token | |||
| rejected_prediction_tokens | 被拒绝的预测 Token | |||
| prompt_tokens_details | ||||
| audio_tokens | 输入中的音频 Token 数 | |||
| cached_tokens | 命中的缓存 Token 数（能省钱/提速） | |||
| latency_checkpoint | 延迟性能监控（单位：毫秒 ms） | |||
| engine_tbt_ms | 引擎 Token 间平均间隔时间 | |||
| engine_ttft_ms | 引擎生成首个 Token 的时间 | |||
| engine_ttlt_ms | 引擎生成最后一个 Token 的时间 | |||
| pre_inference_ms | 推理前的预处理耗时(安全审核、Token 化等预处理) | |||
| service_tbt_ms | 服务端token与token之间生成的间隔时间，决定了打字机效果是否丝滑 | |||
| service_ttft_ms | 服务端接收到请求到输出首字的总时间 | |||
| service_ttlt_ms | 服务端完成全部输出的总时间 | |||
| total_duration_ms | 本次请求在系统中记录的总持续时长 | |||
| user_visible_ttft_ms | 用户看到第一个字跳出来等待的时间 |
python
model.invoke("你是谁？")python
messages = [
{"role": "system", "content": "你是一个专业的翻译员"},
{"role": "user", "content": "帮我把你好世界翻译成英语"}
]
response1 = model.invoke(messages)
print(f"AI的回复：{response1.content}")
messages.append({"role": "assistant", "content": response1.content})
messages.append({"role": "user", "content": "我刚刚问了你什么问题？"})
response2 = model.invoke(messages)
print(f"AI的回复：{response2.content}")python
messages = [
SystemMessage("你是一个专业的翻译员"),
HumanMessage("帮我把您吃了吗翻译成英语")
]
response1 = model.invoke(messages)
print(f"AI的回复：{response1.content}")
messages.append(AIMessage(response1.content))
messages.append(HumanMessage("我刚刚问了你什么问题?"))
response2 = model.invoke(messages)
print(f"AI的回复：{response2.content}")python
from rich import print as rprint
response = model.invoke("你是谁？")
# 格式化打印
# rprint(response)
metadata = response.response_metadata
print(f"使用的模型: {metadata['model_name']}")
print(f"结束原因: {metadata['finish_reason']}")
print(f"模型提供商：{metadata['model_provider']}\n")
usage = metadata.get('token_usage', {})
print(f"输入 tokens: {usage.get('prompt_tokens')}")
print(f"输出 tokens: {usage.get('completion_tokens')}")
print(f"总计 tokens: {usage.get('total_tokens')}")
print(f"消息 ID: {response.id}")stream
py
for chunk in model.stream("用一句话介绍langchain"):
# 逐token输出
print(chunk.text, end="", flush=True)batch
python
messages = [
"中国一线城市有哪些?直接输出城市名称",
"1+1=?",
"45*24=?",
]
# batch：等待所有请求处理完毕，按原始输入顺序返回结果列表
# responses = model.batch(messages)
# 允许应用在收到第一个结果后立即返回响应，而不会等待批次内所有任务完成才响应
responses = model.batch_as_completed(messages)
for response in responses:
print(response)config参数
config参数：允许在调用模型时，动态地配置和控制模型的行为，而无需在初始化时就固定所有参数，这为应用带来了极大的灵活性和可维护性
- 参数
run_name、tags、callbacks主要用在LangSmith中，用于追踪、筛选和调试 metadata可以配置用户指定的一些信息，在工作流开发中，当整个流程被包装为Runnable链时，可以将这些参数传递给后续的链节点使用configurable中可配置的参数与init_chat_model初始化模型参数一样，与在初始化模型时设置的参数（如temperature=0.7）的关键区别在于：init_chat_model初始化参数：模型的 默认设置 ，适用于该模型实例的大部分场景- 运行时 config： 单次调用的特定设置 ，优先级更高，针对本次调用进行的特殊调整
- 配置
configurable覆盖默认参数时需要在init_chat_model初始化模型中指定configurable_fields参数来指定模型运行时可替换的参数有哪些
输出美化
python
response = model.invoke(conversation)
response.pretty_print()python
from rich import print as rprint
rprint(response)模型配置信息profile
LangChain1.1及更高版本可以通过 profile属性 查看模型的配置信息
python
print(model.profile)LangSmith
在LangSmith官网的setting当中创建一个API key。随后在项目当中新增环境变量，随后进行测试
env
# 是否启用
LANGSMITH_TRACING=true
# 监控web地址
LANGSMITH_ENDPOINT=https://api.smith.LangChain.com
LANGSMITH_API_KEY=lsv2_pt_crazyThursdayvme50
# 自定义项目名称，可以在Langsmith WebUI监控页面根据名称查看对应的运行记录
LANGSMITH_PROJECT="my-langchain"Message
这里用invoke的时候已经使用过消息了，这里再对其进行部分补充
HumanMessage
metadata：元数据，可以有很多，自定义
- Open AI API手册
- 模型加载了name传递的信息，在多人对话场景很有用
AIMessage
response_metadata：特有属性，LLM的响应中附加元数据，根据不同模型会有不同，如可能会包含本次token使用量等信息tool_calls：特有属性，表示工具调用信息。当LLM决定调用工具时，在AIMessage 中就会包含这个属性，没有工具调用则为空
ToolMessage
参数说明
- content ：文件内容
- name ：工具名称
- tool_call_id ：工具调用唯一ID，ToolMessage必须紧邻匹配的AIMessage，和前者tool_calls中的id一致
python
model = init_chat_model(
model="deepseek:deepseek-v4-flash",
api_key=DEEPSEEK_API_KEY,
base_url=DEFAULT_API_BASE,
# 需要禁用推理
extra_body={"thinking": {"type": "disabled"}}
)
def get_weather(city: str):
return city + '阳光明媚'
model_with_tools = model.bind_tools([get_weather])
ai_message = AIMessage(
content=[],
tool_calls=[{
"name": "get_weather",
"args": {"city": "长沙"},
"id": "call_00_nUD2NC9QRN5Cg1GaoIkBJQ4s"
}]
)
tool_message = ToolMessage(
content=get_weather("长沙"),
tool_call_id="call_00_nUD2NC9QRN5Cg1GaoIkBJQ4s"
)
message = [
HumanMessage(content="长沙天气如何"),
ai_message,
tool_message
]
response = model_with_tools.invoke(message)
response.pretty_print()AIMessage 中的 tool_calls 只是模型的"意图声明"，它表示模型想说：
"我想调用 get_weather 函数，参数是 city=长沙"
但它不会自动帮你执行这个函数。LangChain 把工具调用拆分成了明确的步骤，需要你手动串联：
用户提问 → 模型决定调用哪个工具(AIMessage.tool_calls) ↓ 你手动执行函数，拿到结果 ← 这一步需要你自己做 ↓ 把结果包装成 ToolMessage 发回给模型 ↓ 模型根据工具结果生成最终回答
对话优化
python
"""
message：已经进行过的对话
max_pairs：最多保留的对话对数
"""
def keep_recent_messages(messages, max_pairs=3):
system_messages = [msg for msg in messages if isinstance(msg, SystemMessage)]
other_messages = [msg for msg in messages if not isinstance(msg, SystemMessage)]
recent_messages = other_messages[-(max_pairs * 2):]
return system_messages + recent_messagespython
messages = [
SystemMessage("你是一个翻译大师")
]
# 这个函数直接拿过来
# def keep_recent_messages
# 第index论对话
index = 1
print("欢迎使用翻译大师！输入你的问题，或者输入" + EXIT_WORD + "退出对话")
while True:
print("\n", f"=======================第${index}轮对话开始=========================")
user_input = input("请输入你的问题：")
if EXIT_WORD == user_input.lower():
print("对话已结束，欢迎下次使用！")
break
messages.append(HumanMessage(user_input))
memory_message = keep_recent_messages(messages, MAX_PAIRS_HISTORY)
reply_content = ""
for chunk in model.stream(memory_message):
if (chunk.content):
print(chunk.content, end="", flush=True)
reply_content += chunk.content
messages.append(AIMessage(reply_content))
print("\n", f"=======================第${index}轮对话结束=========================")
index += 1content & content_blocks
- content可以理解为数据内容，是弱类型的，支持字符串和列表，content传列表也就是为了实现一个多模态
- 多模态指的是处理不同形式数据的能力，例如文本、音频、图像和视频
- 多模态 content API
- API参考
- 在
LangChain 1.x中，content_blocks是消息对象（BaseMessage）的一项重大升级。它的核心目标是提供一种跨模型供应商、标准化的多模态数据结构 - 同时还支持输出格式化
Prompt
ChatPromptTemplate
| 特性 | PromptTemplate | ChatPromptTemplate |
|---|---|---|
| 输出格式 | 纯文本字符串 | 消息列表 |
| 角色支持 | 无 | system/user/assistant |
| 对话历史 | × | √ |
| 适用场景 | 简单提示 | 聊天、对话、多轮交互 |
实例化：from_messages
其底层，也是调用的类的 __init()__ 方法
python
chat_template = ChatPromptTemplate.from_messages(
[
("system", "你是一个有帮助的AI机器人，你的名字是{name}。"),
("human", "你好，最近怎么样？"),
("ai", "我很好，谢谢！"),
("human", "{user_input}"),
]
)python
chat_template1 = ChatPromptTemplate(
[
("system", "你是一个有帮助的AI机器人，你的名字是{name}。"),
("human", "你好，最近怎么样？"),
("ai", "我很好，谢谢！"),
("human", "{user_input}"),
]
)调用
- 基本调用
- invoke：返回ChatPromptValue
- format：返回字符串
- format_messages：返回消息列表
- 注：
SystemMessage、HumanMessage... 这种不支持传参，需要用SystemMessagePromptTemplate...
- 变量预填充
- 消息占位符：可以使用json格式和MessagesPlaceholder方式
- 模板库复用
python
from langchain_core.prompts import ChatPromptTemplate
chat_template = ChatPromptTemplate.from_messages(
[
("system", "你是一个有帮助的AI机器人，你的名字是{name}。"),
("human", "你好，最近怎么样？"),
("ai", "我很好，谢谢！"),
("human", "{user_input}"),
]
)
invoke_prompt = chat_template.invoke({"name": "Jane", "user_input": "我很好"})
# --------------------------------------------------------------------------------------------
format_prompt = chat_template.format(name="Jane", user_input="我很好")
# --------------------------------------------------------------------------------------------
format_message_prompt = chat_template.format_messages(name="Jane", user_input="我很好")python
template = ChatPromptTemplate.from_messages([
("system", "你是{role}，目标用户是{audience}"),
("user", "{task}")
])
# 部分变量预填充 partial
custom_template = template.partial(role="手机销售", audience="张三")
print(custom_template.format_messages(task="这个咋卖？"))python
# 消息占位符
placeholder_template = ChatPromptTemplate.from_messages(
[
("system", "你是一个有用的AI助手"),
("placeholder", "{conversation}"),
MessagesPlaceholder("conversation")
]
)
prompt_value = placeholder_template.invoke({
"conversation": [
HumanMessage("你好"),
AIMessage("你好"),
HumanMessage("1+1=?"),
AIMessage("2"),
HumanMessage("我刚才问了什么问题？"),
AIMessage("你刚才问的是1+1=？")
]
})Tools
在LangChain中，工具（Tools）实际上是指明确定义了输入和输出的 可调用函数 。因此，工具调用(Tool Calling) 也被称为 函数调用( Function Calling)
大模型能根据对话上下文决定何时调用工具以及传递哪些参数，其中整体流转是HumanMessage -> AIMessage 取tools_call，再手动调用工具，返回ToolMessage，再给到AIMessage，大语言模型是不支持直接调用tool的
创建 & 调用
- 使用
@tool表示该方法是一个tool工具，里面需要加上docstringdocstring风格：Google 风格 docstring 说明- 参数说明
description：工具描述，优先级最高parse_docstring：docstring会被解析，填充到相应的字段描述中name_or_callable：更改方法名称【开发中，习惯使用函数名作为工具名称，不推荐自定义工具名称】
- 执行
model.bind_tools([get_weather])，底层最终会调用convert_to_openai_tool生成工具描述- type：定义当前数据节点必须是什么数据类型。常见类型有 string, number, integer, boolean, object, array, null。object即是json对象
- properties：用于定义JSON 对象（Object）中可以包含哪些属性（键），以及每个属性对应的值类型和说明
- required：当 type为 "object"时使用，是一个数组，列出了对象中必须存在的属性名
- 为什么不使用@tool装饰器修饰的函数，也可以理解为工具
- 查看
convert_to_openai_tool底层源码
- 查看
python
@tool
def get_weather(city: str) -> str:
"""获取指定城市的天气信息"""
return city + '阳光明媚'
# @tool其他参数配置
@tool(description="根据城市名称查询当日天气的工具")
@tool(parse_docstring=True)
@tool(name_or_callable="get_weather_tool")python
rprint(convert_to_openai_tool(get_weather))自定义args_schema
- Pydantic
- 主要优势在于能够精确控制工具参数的格式和验证规则
- 通过继承核心基类 BaseModel 定义数据模型，从而声明字段结构、类型约束、默认值以0及校验规则
Field：用来定制字段，可用于设置默认值、描述等Literal：限定参数为固定选项- 通过args_schema进行关联
- Json Schema
- 可以基于数据库配置或用户输入在 运行时动态生成，所以适合参数结构需要动态生成的场景
python
class WeatherInput(BaseModel):
city: str = Field(
description="城市名称",
default="北京"
)
dt: str = Field(
description="日期",
default="2026-07-24"
)
unit: Literal["c", "f"] = Field(
description="温度单位",
default="c"
)
@tool(description="获取指定城市的天气信息", args_schema=WeatherInput)
def get_weather(city: str, dt: str, unit: Literal["c", "f"]):
return f"日期：{dt}，{city}阳光明媚，单位：{unit}"python
weather_schema = {
'type': 'function',
'function': {
'name': 'get_weather_tools',
'description': '获取指定城市的天气信息',
'parameters': {
'properties': {
'city': {
'default': '北京',
'description': '城市名称',
'type': 'string'
},
'dt': {
'default': '2026-07-24',
'description': '日期',
'type': 'string'
},
'unit': {
'default': 'c',
'description': '温度单位',
'enum': ['c', 'f'],
'type': 'string'
}
},
'type': 'object'
}
}
}
@tool(args_schema=weather_schema)model call tool
前面说过，模型不能够自动的调用工具，他只能识别需要调用哪些工具，在这里第一个invoke会得到一个AIMessage，这里面包含了tool_calls，也就是他需要调用哪些工具， 所以在后面就直接开始循环所有工具进行匹配调用，这一步之后会得到一个ToolMessage，再将其给到模型，得到最终结果
- bind_tools方法
- 底层最终会调用
convert_to_openai_tool生成工具描述 - tool_choice：用于控制是否强制使用工具，该字段最终会作为 payload 的 tool_choice 字段传递给模型
- 可选值：
auto【默认】、none、required
- 可选值：
- 底层最终会调用
python
@tool
def get_weather(city: str) -> str:
"""获取指定城市的天气信息"""
return city + '阳光明媚'
messages = [
HumanMessage("北京天气如何？")
]
model_with_tools = model.bind_tools([get_weather])
response = model_with_tools.invoke(messages)
if response.tool_calls:
print('AI想调用的工具：', response.tool_calls)
else:
print('AI的回答：', response.content)
messages.append(response)
tool_calls = response.tool_calls
for tool_call in tool_calls:
if tool_call["name"] == "get_weather":
# 返回的是ToolMessage类型消息
tool_response = get_weather.invoke(tool_call)
print('调用get_weather后返回的数据类型：', type(tool_response))
messages.append(tool_response)
for msg in messages:
print("每一条消息：----", msg)
final_response = model_with_tools.invoke(messages)
print(f"最终的回答: \n{final_response}")一个Tool应该具备
- 清晰的描述【docstring、args_schema、return】
- 功能单一【一个工具只做一件事】
- 如何处理工具失败
- 工具内部处理，加上try except捕获异常
- Agent 级重试（使用 prompt），prompt=如果工具调用失败，尝试使用其他方法解决问题。
- 调用级重试，加上：
@retry(stop=stop_after_attempt(3))如果失败，最多尝试 3 次
- 返回字符串
- 大模型（LLM）的本质只吃“文本”
- 避免大模型“胡思乱想”（乱码与格式问题）
return json.dumps(user, ensure_ascii=False)
- 同步 vs 异步
- 同步工具 ：简单场景，CPU 密集型任务
- 异步工具 ：IO 密集型（API 调用、数据库、文件操作）
Agent
创建 & 调用
create_agent- 底层基于 LangGraph 实现
- 完整参数说明： create_agent_parameters
- model：模型，可以直接传字符串或者
init_chat_model初始化的模型 - tools：工具列表
- langChain内置的工具列表
- 内置工具：Tavily Search，登录之后创建一个API_KEY
- config：可以设置
recursion_limit限制工具调用次数 - system_prompt：系统提示词，提示词为Agent提供了任务背景、行为准则和操作指南
- model：模型，可以直接传字符串或者
agent.invoke()- 输入：传入的参数为字典类型，字典内通过 messages字段传递消息列表
- 输出：通过invoke调用Agent，底层可能会经历多轮交互，返回的是完整的 消息列表
- 重试机制
- 可以在messages里面的SystemMessage加上提示词：
若工具调用失败，请尝试再次调用3次，如果3次都失败，请返回错误信息
- 可以在messages里面的SystemMessage加上提示词：
- 流式输出stream：通过
stream_mode可以指定输出方式- 可选类型：
Literal["values", "updates", "checkpoints", "tasks", "debug", "messages", "custom"]
- 可选类型：
python
create_agent(
model: str | BaseChatModel,
tools: Sequence[BaseTool | Callable[..., Any] | dict[str, Any]] | None = None,
*,
system_prompt: str | SystemMessage | None = None,
middleware: Sequence[AgentMiddleware[StateT_co, ContextT]] = (),
response_format: ResponseFormat[ResponseT] | type[ResponseT] | dict[str, Any] | None = None,
state_schema: type[AgentState[ResponseT]] | None = None,
context_schema: type[ContextT] | None = None,
checkpointer: Checkpointer | None = None,
store: BaseStore | None = None,
interrupt_before: list[str] | None = None,
interrupt_after: list[str] | None = None,
debug: bool = False,
name: str | None = None,
cache: BaseCache[Any] | None = None,
transformers: Sequence[TransformerFactory] | None = None
) -> CompiledStateGraph[AgentState[ResponseT], ContextT, InputAgentState, OutputAgentState[ResponseT]]python
load_dotenv(override=True)
agent = create_agent("deepseek:deepseek-v4-flash")
# ---------------------------------------------------------------------
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEFAULT_API_BASE = os.getenv("DEFAULT_API_BASE")
model = init_chat_model(
model="deepseek:deepseek-v4-flash",
api_key=DEEPSEEK_API_KEY,
base_url=DEFAULT_API_BASE
)
agent2 = create_agent(model)python
@tool(parse_docstring=True)
def get_weather(city: str):
"""
用来获取城市的天气信息
Args:
city:城市名称
"""
return f"{city}的天气是晴天"
agent = create_agent(model=model, tools=[get_weather])
messages = [
SystemMessage("你是一个天气助手，可以根据城市名称获取天气信息"),
HumanMessage("广州的天气咋样")
]
result = agent.invoke({"messages": messages})python
web_search = TavilySearch(
tavily_api_key=os.getenv("TAVILY_API_KEY"),
max_results=2
)
agent = create_agent(name='联网搜索Agent', model=model, tools=[web_search], system_prompt="")
messages = [
SystemMessage("你是一个联网查询助手"),
HumanMessage("2018年S冠军是那个战队")
]
result = agent.invoke({"messages": messages})python
for chunk in agent.stream(
{"messages": messages},
# stream_mode="values"
stream_mode="updates"
):
rprint(chunk)
print("*" * 50)流式输出：stream_mode总结：
- 实现实时对话交互 --> 优先选择messages
- 观察Agent的 思考与执行步骤 --> 优先选择updates
- 需要查看每一步状态 --> 优先选择values/tasks/debug
- 在工具执行时输出自定义业务 --> 日志优先选择custom
| 模式 | 输出内容 | 使用场景 |
|---|---|---|
| values | 每个步骤执行后，都会输出完整的状态信息 | 适用于每一步都要获取完整状态、状态持久化场景 |
| updates（默认） | 每个步骤执行后，只增量更新状态中发生变化的内容 | 用于监控Agent 执行进度 例如观察Agent决定调用工具、工具执行结果等步骤 |
| messages | 输出流式返回的Token以及相关的元数据（如：来自哪个节点model/tool） | 实现类似ChatGPT 的打字机效果 为聊天机器人等交互式应用提供最佳的实时体验 |
| tasks | 输出当前task任务开始和结束的时间，包含任务的结果和错误信息 | 该模式用于监控任务的生命周期 |
| debug | 与tasks模式类似，比task模式多输出 任务步骤、时间戳、task类型（task/task_result） | 该模式用于调试、监控task任务的生命周期 |
| checkpoints | 当检查点（checkpoint）被创建时会触发输出，输出包含检查点中的状态 | 用于需要状态持久化 工作流恢复或分布式执行跟踪的高级场景 |
| custom | 通过get_stream_writer在工具或节点内部自定义发送的数据 | 用于输出业务逻辑相关的进度信息 （如“已处理10/100条记录”）、自定义日志或指标 |
工具调用流程：当用户提出一个复杂需求时，Agent会像人类一样，先理解任务、规划步骤、使用合适的工具（如搜索 网络、查询数据库、执行计算）获取信息，Agent 会在一个循环中 反复调用模型和工具 ，直到某次模型输出中 不再包含工具调用 则结束，最后综合所有信息给出最终答案
ReAct = Reasoning（推理思考） + Acting（执行行动）
核心循环：Thought → Action → Observation（循环往复）
Thought（思考）
模型先推理：我现在需要什么信息？应该调用哪个工具？怎么拆解问题？纯文本内在思考，不输出外部动作
Action（行动）
根据思考结果，调用外部工具（搜索、计算器、代码解释器、API 等）
Observation（观察）
获取工具返回的结果，作为新信息交给模型。
结构化输出 ToolStrategy
- 如果使用的deepseek v4的模型需要关闭推理模式：
extra_body={"thinking": {"type": "disabled"}}并且其不支持ProviderStrategy，使用ToolStrategy - 联合模式Union：
response_format=ToolStrategy(Union[ContactInfo, EventInfo])- LLM能够根据输入文本的内容，智能地选择 最合适的一个 数据模型（Schema）来生成结构化输出，但是最终会只有一种类型输出
- tool_message_content：定制其消息内容，将指定的内容写入对话历史的提示信息
- handle_errors：错误处理
python
class UserInfo(BaseModel):
name: str = Field(..., description="用户姓名")
age: int = Field(..., description="用户年龄")
city: str = Field(..., description="用户所在城市")
# deepseek-v4-flash 模型不支持 response_format 参数
# agent = create_agent(name='信息提取Agent', model=model, response_format=ProviderStrategy(UserInfo))
agent = create_agent(name='信息提取Agent', model=model, response_format=ToolStrategy(UserInfo))
# agent = create_agent(name='信息提取Agent', model=model, response_format=AutoStrategy(UserInfo))
# agent = create_agent(name='信息提取Agent', model=model, response_format=None)
messages = [
HumanMessage("用户信息提取，张三今年16岁来自上海")
]
result = agent.invoke({"messages": messages})
rprint(result)python
# 自定义错误处理函数
def custom_error_handler(error: Exception) -> str:
"""自定义错误处理器"""
error_str = str(error)
print(f"捕获到错误类型：{type(error).__name__}")
print(f"错误详情：{error_str}")
if isinstance(error, StructuredOutputValidationError):
return "数据格式有误，请检查字段是否符合要求。"
elif isinstance(error, MultipleStructuredOutputsError):
return "检测到多个响应，请选择最相关的一个进行返回。"
else:
return f"Error: {error_str}"
agent = create_agent(name='信息提取Agent', model=model, response_format=ToolStrategy(UserInfo, handle_errors=True))
# handle_errors: bool | str | type[Exception] | tuple[type[Exception], ...]
# handle_errors=True handle_errors=False
# handle_errors="请检查输入数据"
# handle_errors=(MultipleStructuredOutputsError,StructuredOutputValidationError) 对指定异常类型进行捕获
# handle_errors=custom_error_handlerMiddleware
Agent 执行过程中的钩子函数
内置中间件
SummarizationMiddleware
- API Docs
- 作用：对历史消息列表进行摘要&总结 ，达到压缩上下文的效果
- 原理：在达到触发条件时，调用大模型对历史消息进行摘要，将摘要的结果作为HumanMessage，放到消息列表最开始的位置
- 参数说明
- model：用于指定生成摘要的模型
- trigger：摘要的触发条件
- tokens：历史token的累计数量达到该值触发摘要
- messages：历史消息条数达到该值触发摘要
- fraction：历史token的累计数量达到模型的
max_input_tokens*fraction触发摘要- 如果设置了fraction，则要求模型的profile包含
max_input_tokens
- 如果设置了fraction，则要求模型的profile包含
- keep：摘要时保留的原始消息，支持三种条件
- tokens：摘要时保留的token数量
- messages：摘要时保留的历史消息条数
- fraction：摘要时保留
max_input_tokens*fraction个token
- token_counter：统计token数量的函数
- summary_prompt：摘要时的自定义提示词
- trim_token_to_summarize：摘要时历史消息的最大token数
使用SummarizationMiddleware
python
model = init_chat_model(
model="deepseek:deepseek-v4-flash",
api_key=DEEPSEEK_API_KEY,
base_url=DEFAULT_API_BASE,
profile={
"max_input_tokens": 128_000
}
)
agent = create_agent(
model,
middleware=[
SummarizationMiddleware(
model=model,
trigger=[
("tokens", 100),
("messages", 6),
("fraction", 0.001)
],
keep=("messages", 2),
summary_prompt="对历史消息摘要，消息列表如下\n{messages}"
)
]
)
messages = [
SystemMessage("你是个非常友好的AI助手"),
HumanMessage("一句话介绍LangChain"),
AIMessage(
"LangChain是一个用于构建和编排由大型语言模型（LLM）驱动的应用开发框架，通过将复杂任务链式组合，让AI能像搭积木一样调用工具、记忆和外部数据。"),
HumanMessage("推荐其的js版本还是py版本进行开发"),
AIMessage("如果你是纯后端开发或做AI应用原型，无脑选 Python。如果你是前端/全栈工程师，且不想维护两套语言，选 JS 版。"),
HumanMessage("推不推荐java版本，一句话回答")
]
result = agent.invoke({"messages": messages})
for msg in result["messages"]:
msg.pretty_print()HumanInTheLoopMiddleware
- API Docs
- 作用：在工具调用前中断Agent运行，等待用户对工具调用请求决策。可选的决策有
- approve（同意执行）
- edit（编辑调用配置后执行）
- reject（拒绝执行）
- 参数说明
- interrupt_on：工具名和中断策略的映射
- True表示所有决策（approve, edit, reject） 都可以选择
- False表示不中断，即无需审批即可执行
- InterruptOnConfig是一个TypedDict的子类，可以用字典直接赋值
- allowed_decisions：精细控制中断后允许的决策
- description：特定工具的中断描述信息，优先级高于
description_prefix
- description_prefix：自定义中断描述
- 默认为"Tool execution requires approval"
- interrupt_on：工具名和中断策略的映射
使用HumanInTheLoopMiddleware
python
@tool
def get_weather(location: str) -> str:
"""
根据位置获取天气信息.
Args:
location:城市、地点名称
"""
return f"{location}阳光明媚"
@tool
def get_news(location: str) -> str:
"""根据位置获取新闻信息."""
return f"{location}新闻更新"
@tool
def read_email_tool(email_id: str) -> str:
"""根据邮件id读取邮件内容."""
return f"邮件内容：{email_id}"
@tool
def send_email_tool(email_id: str) -> str:
"""根据邮件id发送邮件."""
return f"邮件已发送：{email_id}"
agent = create_agent(
model,
tools=[get_weather, get_news, read_email_tool, send_email_tool],
checkpointer=InMemorySaver(),
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={
"get_weather": True,
"get_news": True,
"read_email_tool": False,
"send_email_tool": InterruptOnConfig(
allowed_decisions=["approve", "reject"],
),
},
description_prefix="中断啦"
)
]
)
messages = [
HumanMessage(
"帮我查询广州的天气，获取深圳的新闻信息，读取邮件id为1的邮件内容，给邮件id为2的邮箱发送邮件，同时帮我做这四件事"),
]
config = RunnableConfig(configurable={"thread_id": "1"})
result = agent.invoke({"messages": messages}, config=config)
# rprint(result)
weather_decision = {
"type": "edit",
"edited_action": {
"name": "get_weather",
"args": {"location": "佛山"}
}
}
news_decision = {
"type": "approve",
}
send_email_decision = {
"type": "approve"
}
decisions = {
"decisions": []
}
interrupts = result.get("__interrupt__", [])
action_requests = interrupts[0].value["action_requests"]
for action_request in action_requests:
if action_request["name"] == "get_weather":
decisions["decisions"].append(weather_decision)
if action_request["name"] == "get_news":
decisions["decisions"].append(news_decision)
if action_request["name"] == "send_email_tool":
decisions["decisions"].append(send_email_decision)
if interrupts:
# 审批通过
resumed_response = agent.invoke(
Command(resume=decisions),
config=config, # 必须是同一个 thread_id
)
for msg in resumed_response["messages"]:
msg.pretty_print()PIIMiddleware
- API Docs
- 用于检测和处理对话中的个人身份信息（Personally Identifiable Information，PII），支持自定义处理策略
- 产说说明
- pii_type：检测的PII数据类型，可以是内置类型或自定义类型，自定义类型有
- strategy：处理PII信息的策略
- detector：自定义PII检测函数或者正则表达式
- apply_to_input：是否在调用模型前检测【True】
- apply_to_output：是否在模型调用后检测【False】
- apply_to_tool_results：是否在工具调用后检测其输出【False】
使用PIIMiddleware
python
# 自定义检测函数
def detect_phone_number(content: str):
return [
{
"text": m.group(0), # 提取出具体匹配到的 11 位数字文本（例如"13800138000"）
"start": m.start(), # 这段数字在原文本中的“起始索引位置”（从 0 开始算）
"end": m.end() # 这段数字在原文本中的“结束索引位置”
}
for m in re.finditer(r"[0-9]{11}", content)
]
agent = create_agent(
model,
middleware=[
PIIMiddleware("email", strategy="redact", apply_to_input=True),
PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
PIIMiddleware("url", strategy="hash", apply_to_input=True),
PIIMiddleware("mac_address", strategy="mask", apply_to_input=True),
PIIMiddleware("ip", strategy="block", apply_to_input=True),
PIIMiddleware("api_key", strategy="hash", apply_to_input=True, detector=r"sk-[a-zA-Z0-9]+"),
PIIMiddleware("phone_number", strategy="mask", apply_to_input=True, detector=detect_phone_number)
]
)
try:
response = agent.invoke({
"messages": [
# HumanMessage("帮我向 156168188@qq.com 发送一封邮件，同时查看银行卡号： 5105-1051-0510-5100 的余额，"
# "访问 https://localhost:12345，确认这是不是 MAC地址： 11-11-11-11-11-11"),
# HumanMessage("看看这个 IP 能不能 ping 通：192.168.10.1")
HumanMessage("试一试我这个sk-crazyThursdayvme50是否有效，给张三打个电话，他的手机号是：13800138000")
]
})
for msg in response["messages"]:
msg.pretty_print()
except Exception as e:
print(f"检测到IP，抛出异常：{e}")TodoListMiddleware
- API Docs
- 赋予了Agent 任务规划和追踪进度的能力，可以应对复杂的多步任务
- 参数说明
- system_prompt：自定义指导todo列表使用的提示词
- tool_description：自定义write_tools工具的描述信息
ModelCallLimitMiddleware & ToolCallLimitMiddleware
- ModelCallLimitMiddleware
- API Docs
- 限制模型调用次数，避免无限循环，控制调用成本
- ToolCallLimitMiddleware
- API Docs
- 限制工具调用次数，可以限制所有工具调用的总次数，也可以限制特定工具的调用次数
- 退出行为的三种模式
- error：直接抛异常
- end：结束整个会话
- continue：继续运行Agent，这是默认行为，此时Agent会将工具调用超出限制的信息传递给模型，后者自主决定后续行为，如果模型能力不足，可能导致死循环，为了避免这种情况，我们实现的fake server会以20%的概率输出正确响应，从而能终止循环
pythom
ModelCallLimitMiddleware(
thread_limit=2, # 每个线程最多2次模型调用
# run_limit=5, # 每次运行最多5次
exit_behavior="end", # 达到限制后退出
)python
ToolCallLimitMiddleware(
# thread_limit=2, # 每个线程最多2次工具调用
run_limit=2, # 每次运行最多2次
exit_behavior="end",
)Other Middlewares
| Middleware | 说明 |
|---|---|
| ModelFallbackMiddleware | 用于故障转移，当主模型无法访问时，启用备用模型 |
| LLMToolSelectorMiddleware | 智能工具筛选。当工具太多时，用子模型筛选最相关的几个工具 |
| ToolRetryMiddleware | 基于指数退避算法，设置工具调用失败时的重试策略 |
| ModelRetryMiddleware | 模型调用失败时重试，策略和工具调用的重试一样，都是基于指数退避算法 |
| LLMToolEmulator | 某些情况下，工具尚未开发完成，我们希望先测试工具调用，可以用LLM tool emulator模拟工具 |
| ContextEditingMiddleware | 上下文编辑中间件，该中间件提供了上下文管理的一种方式 通过更改发送给模型的消息列表来控制成本 |
| FilesystemFileSearchMiddleware | 基于系统的Glob和Grep检索工具，为Agent赋予本地文件搜索和分析的能力 |
| Shell tool | 为Agent提供一个可以执行命令的Shell环境 |
| Filesystem | 这是源自deepagents（基于LangChain的另一个框架）的中间件 内置了四个工具，分别用于查看目录、读文件、写文件和改文件 |
| Subagent | 来自deepagents的中间件 用于便捷地创建子Agent |
自定义中间件
Hook钩子函数分类
- Node-style hooks：在流程的特定节点运行
- before_agent：在Agent开始运行之前执行
- before_model：在模型调用之前执行
- after_model：在模型调用之后执行
- after_agent：在Agent流程全部完成后执行
- Wrap-style hooks：在模型或工具调用前后运行
- wrap_model_call （包裹模型调用）
- wrap_tool_call （包裹工具调用）
Node-style
- 基于装饰器实现
- 用对应的四个钩子装饰器：before_model、after_model、before_agent、after_agent
- before_model：消息修剪（trim messages）、PII 脱敏、输入验证、条件路由
- after_model：输出验证、格式化响应、统计信息、状态更新
- 如何知道这几个注解下的方法的参数和返回值？
- ctrl点进去一个注解，以
before_model为例，进去直接直接搜这个可以看到其对应的声明，（里面还有self？） - 这是因为其底层是会构造一个AgentMiddleware类
- ctrl点进去一个注解，以
- 入参说明
- state: 是一个AgentState实例，维护Agent运行过程中的状态，这类状态会随着Agent的运行而发生变化，包括消息列表
- runtime: 是一个Runtime实例，维护Agent运行过程中的上下文环境，包括上下文、长期记忆等
- 返回值说明
- None：不修改状态
- 字典：更新状态
- {“jump_to”: “…”}：控制流程
- end：结束 Agent
- tools：跳到工具节点
- 其他自定义节点
- 装饰器参数：
can_jump_to决定了钩子函数可以直接跳转至流程的哪些位置- 取值：
Literal["tools", "model", "end"] - 相当于在before_model进行拦截，先进行一个判断，第一次是不会进去的，之后返回一个人工构造的AIMessage，并且让这个AIMessage去调用工具，此时Agent会触发工具调用
- 这个时候整个消息列表是HumanMessage、AIMessage和ToolMessage
- 再会进来这个before_model进行判断，此时有AIMessage并且其content的内容是我们手动构造的，进入这个判断，之后返回None表示退出这个Hook，基于后面的执行
- 最终会返回4条消息，顺序是HumanMessage、AIMessage、ToolMessage、AIMessage
- 最后一个AIMessage的内容是：你好！今天的天气阳光明媚，是个不错的日子。请问有什么我可以帮你的吗？ -> after_model <- -> after_agent <-
- 相当于进行拦截后强制调用工具，在初始的时候只是一个
HumanMessage("你好")，但最终会返回到这个工具当中得到的天气信息
- 取值：
- 基于类实现
- 需要继承
AgentMiddleware父类，重写对应的方法 - 在类中实现can_jump_to，使用注解
@hook_config传递这个参数即可
- 需要继承
python
@before_model
def before_model_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
state["messages"][-1].content += " -> before_model <- "
return None
@after_model
def after_model_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
state["messages"][-1].content += " -> after_model <- "
return None
@before_agent
def before_agent_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
state["messages"][-1].content += " -> before_agent <- "
return None
@after_agent
def after_agent_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
state["messages"][-1].content += " -> after_agent <- "
return None
agent = create_agent(
model=model,
middleware=[before_model_middleware, after_model_middleware, before_agent_middleware, after_agent_middleware]
)python
def before_model(self, state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None:
"""Logic to run before the model is called.
Args:
state: The current agent state.
runtime: The runtime context.
Returns:
Agent state updates to apply before model call.
"""
def before_model(
func: _CallableWithStateAndRuntime[StateT, ContextT] | None = None,
*,
state_schema: type[StateT] | None = None,
tools: list[BaseTool] | None = None,
can_jump_to: list[JumpTo] | None = None,
name: str | None = None,
) -> (
Callable[[_CallableWithStateAndRuntime[StateT, ContextT]], AgentMiddleware[StateT, ContextT]]
| AgentMiddleware[StateT, ContextT]
):python
class MyMiddleware(AgentMiddleware):
def __init__(self):
super().__init__()
def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
state["messages"][-1].content += " -> before_model <- "
return None
def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
state["messages"][-1].content += " -> after_model <- "
return None
def before_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
state["messages"][-1].content += " -> before_agent <- "
return None
def after_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
state["messages"][-1].content += " -> after_agent <- "
return None
agent = create_agent(
model=model,
middleware=[MyMiddleware()]
)python
@tool
def get_weather() -> str:
"""获取天气信息"""
return "今天的天气阳光明媚"
@before_model(can_jump_to=["tools"])
def before_model_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
# 检查是否已经有工具调用记录（避免死循环）
for msg in state["messages"]:
# 如果已经有人工构造的消息，说明已经跳转过，直接返回
if isinstance(msg, AIMessage) and msg.content.startswith("消息构造"):
return None
# 第一次调用时，跳过模型直接执行工具
return {
"messages": [
AIMessage(
content="消息构造：这是一条消息",
tool_calls=[
{
"name": "get_weather",
"args": {},
"id": "call_force_weather_001",
}
],
)],
"jump_to": "tools",
}python
@hook_config(can_jump_to=["tools", "end"])
def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:Wrap-style
可以同时在模型、工具调用前后做事
- 基于装饰器实现
- 用对应的两个钩子装饰器：
@wrap_model_call和@wrap_tool_call
- 用对应的两个钩子装饰器：
- 基于类实现
- 需要继承
AgentMiddleware父类，重写对应的方法
- 需要继承
python
@wrap_model_call
def wrap_model_call_middleware(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse | AIMessage | ExtendedModelResponse:
request.messages[-1].content += " -> wrap_model_call_before <- "
response = handler(request)
response.result[0].content += " -> wrap_model_call_after <- "
return response
@tool
def get_weather(city: str, is_forcast: bool) -> str:
"""
获取天气信息
Args:
city:城市名称
is_forcast:明天的天气
"""
if is_forcast:
msg = f"{city}今天的天气是晴天，明天的天气是雨天"
else:
msg = f"{city}的天气是晴天"
return msg
@wrap_tool_call
def wrap_tool_call_middleware(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command:
result = handler(request)
print(f"原始参数：{request.tool_call['args']}")
print(f"原始参数调用结果： {result}")
request.tool_call["args"]["is_forcast"] = True
result = handler(request)
print(f"更新后的参数：{request.tool_call['args']}")
print(f"更新参数调用结果： {result}")
return result
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[wrap_model_call_middleware, wrap_tool_call_middleware]
)python
@tool
def get_weather(city: str, is_forcast: bool) -> str:
"""
获取天气信息
Args:
city:城市名称
is_forcast:明天的天气
"""
time.sleep(random.uniform(0.5, 1.5))
if is_forcast:
msg = f"{city}今天的天气是晴天，明天的天气是雨天"
else:
msg = f"{city}的天气是晴天"
return msg
class WrapModelCallMiddleWare(AgentMiddleware):
def __init__(self):
super().__init__()
def wrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse | AIMessage | ExtendedModelResponse:
request.messages[-1].content += " -> wrap_model_call_before <- "
response = handler(request)
response.result[0].content += " -> wrap_model_call_after <- "
return response
def wrap_tool_call(
self,
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command:
tool_name = request.tool_call["name"]
tool_args = request.tool_call.get("args", {})
start_time = time.time()
result = handler(request)
elapsed = time.time() - start_time
print(f"🔧 开始执行工具：{tool_name}， 参数：{tool_args} ✅ 工具执行成功，耗时：{elapsed:.2f}秒")
return result
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[WrapModelCallMiddleWare()]
)hook函数执行顺序
- before_* 钩子函数：从前到后执行
- after_* 钩子函数：从后往前执行
- wrap_* 钩子函数：洋葱架构，前面的包裹后面的
python
agent = create_agent(
model=model,
middleware=[
before_model_middleware1,
before_model_middleware2,
before_model_middleware3,
after_model_middleware1,
after_model_middleware2,
after_model_middleware3,
wrap_model_call_middleware1,
wrap_model_call_middleware2,
wrap_model_call_middleware3
]
)text
================================ Human Message =================================
打个招呼 -> before_model1 <- -> before_model2 <- -> before_model3 <- -> wrap_model_call_before1 <- -> wrap_model_call_before2 <- -> wrap_model_call_before3 <-
================================== Ai Message ==================================
你好！很高兴见到你，有什么可以帮你的吗？😊 -> wrap_model_call_after3 <- -> wrap_model_call_after2 <- -> wrap_model_call_after1 <- -> after_model3 <- -> after_model2 <- -> after_model1 <-Memory
LangChain的上下文工程是基于Agent讨论的，而上下文工程是构建在LangGraph之上的。
LangGraph 提供了三种管理上下文的方法，这些方法结合了可变性和生命周期维度：
| 上下文类型 | 描述 | 可变性 | 生命周期 | 访问方法 |
|---|---|---|---|---|
| 动态运行时上下文 | 在单次运行中会演变的可变数据 | 动态 | 单次运行 | LangGraph state对象 |
| 动态跨会话上下文 | 在对话间共享的持久数据。比如用户偏好、历史洞察、知识条目 | 动态 | 跨对话 | LangGraph store对象 |
| 静态运行时上下文 | 在启动时传入的用户元数据、工具、数据库 连接 | 静态 | 单次运行 | LangGraph context对象 |
LangChain的记忆分类
- 短期记忆：对话流水，批量进 Prompt（可内存、也可持久化）
- 长期记忆：历史事实库，检索按需取用（几乎一定持久化）
短期记忆
使用State（会话内部状态） + checkpointer（持久化机制） + Thread ID（会话作用域）
State：默认存储历史消息列表messages，通过State 管理历史消息Checkpointer：负责将State 作为检查点持久化保存，检查点是某个时刻的State 快照Thread ID：用于唯一标识State，LangChain运行时会按照 thread_id 读写State快照
核心流程：InMemorySaver + thread_id 实现短期记忆：每次调用先根据thread_id读取上一轮完整状态快照，借助add_messages归约器合并本轮用户消息；Agent 生成回答后再次更新状态，最终把最新完整快照存入检查点。
使用PostgresSaver会创建这些表
- checkpoints：这是主表，存每个 thread 在某个时刻的 checkpoint快照
- checkpoint_blobs：这张表专门存不适合直接内联进 checkpoints.checkpoint 的较复杂channel值
- checkpoint_writes：这张表存的是中间写入/ pending writes，不是最终完整 checkpoint
- checkpoint_migrations：这张表不是业务数据表，而是迁移版本表
python
agent = create_agent(
model=model,
checkpointer=InMemorySaver()
)
config: RunnableConfig = {
"configurable": {
"thread_id": "1"
}
}
response = agent.invoke({
"messages": [
HumanMessage("你好，我叫张三")
]
}, config=config)
response_next = agent.invoke({
"messages": [
HumanMessage("我叫什么？")
]
}, config=config)python
POSTGRES_SQL_URL = “postgresql://admin:123456@localhost:5432/langchain_db?sslmode=disable”
with PostgresSaver.from_conn_string(POSTGRES_SQL_URL) as checkpointer:
# 初始化postgre sql数据库
checkpointer.setup()
agent = create_agent(
model=model,
checkpointer=checkpointer
)
config: RunnableConfig = {
"configurable": {
"thread_id": "1"
}
}
response = agent.invoke({
"messages": [
HumanMessage("你好，我叫张三")
]
}, config=config)
response_next = agent.invoke({
"messages": [
HumanMessage("我叫什么？")
]
}, config=config)记忆治理策略（上下文管理）
- 消息裁剪：调用模型前裁剪上下文
- 在调用模型之前，将整个的消息列表进行裁剪，保留第一条，然后判断其长度进行裁剪拼接
- 消息删除：控制模型可以看到的上下文范围
- RemoveMessage到底干了什么
- 追加“墓碑”标记：框架收到
RemoveMessage(id=“1”)后，并不会去内存的数组里把 id=“1” 的对象删掉，而是把这个 RemoveMessage 作为一条新记录追加到当前线程的状态历史中。这个RemoveMessage就像是一个“墓碑” - 运行时过滤合并（Reducer）：当下一次你再次调用
agent.invoke或者大模型要去读取上下文时，框架的内置合并器（Reducer）会把“原始消息”和“墓碑标记”放在一起进行计算
- 追加“墓碑”标记：框架收到
- RemoveMessage到底干了什么
- 摘要
python
@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
messages = state["messages"]
if len(messages) <= 3:
return None
first_msg = messages[0]
recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
new_messages = [first_msg] + recent_messages
return {
"messages": [
RemoveMessage(id=REMOVE_ALL_MESSAGES),
*new_messages
]
}python
@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
messages = state["messages"]
if len(messages) > 5:
to_delete = len(messages) - 5
return {"messages": [RemoveMessage(id=m.id) for m in messages[:to_delete]]}
return None长期记忆
长期记忆的存储是 store -> namespace -> key -> value的四层架构
- 基础API
- put
- namespace：文档所在的层级路径
- key：该路径下的唯一键
- value：要保存的 JSON-like 字典
- index：控制语义检索索引
None(默认选项)：使用 store 初始化时配置的索引配置，如果初始化时没有指定索引策略，则index参数将会被忽略False：不为该 item 建立语义索引list[str]：只对指定字段路径建索引
- ttl：可选，过期时间；是否支持取决于具体 store 实现
- get
- search
- namespace_prefix：命名空间前缀，在该前缀下搜索
- query：语义检索时用于查询的自然语言
- filter：过滤条件，value中的键值对组合，见下文举例
- limit：可以返回item的最大条数，效果等同于SQL中的limit
- offset：返回结果之前跳过的item数量
- refresh_ttl：同上
- put
- InMemoryStore
- PostgresStore
python
store = InMemoryStore()
namespace = ("users",)
user_id = 'user-1'
username = "modify"
store.put(namespace, user_id, {"name": username})
store.put(namespace, "user-2", {"name": "张三"})
store.put(namespace, "user-3", {"name": "李四"})
store.put(namespace, "user-4", {"name": "王五"})
store.put(namespace, "user-5", {"name": "张六"})
store.put(("goods",), "goods-1", {"name": "手机"})
print(store.get(namespace, user_id), "\n")
# search检索
for item in store.search(namespace):
print(item)
print("\n")
for item in store.search(namespace, filter={"name": "张三"}):
print(item)
print("\n")
DB_URL = "postgresql://admin:123456@localhost:5432/langchain_db?sslmode=disable"
with PostgresStore.from_conn_string(DB_URL) as sqlStore:
# 创建数据库
sqlStore.setup()
sqlStore.put(namespace, user_id, {"name": username})
print(sqlStore.get(namespace, user_id))python
# 因为AgentState是TypedDict结构化类型，在运行时阶段是可以通过的，在类型检查时是不认可的，【要求是精确匹配】，这里扩展了一个字段
class CustomState(AgentState):
user_id: NotRequired[str]
@tool(parse_docstring=True)
def save_user_info(name: str, runtime: ToolRuntime) -> str:
"""
将用户信息保存在长期记忆中
Args:
name: 用户名
runtime: 工具的运行时
Returns:
str: 保存状态
"""
assert runtime.store is not None, "store 未配置"
runtime.store.put(("users",), runtime.state["user_id"], {"name": name})
return "saved"
@tool(parse_docstring=True)
def get_user_info(runtime: ToolRuntime) -> str:
"""
从长期记忆中读取用户信息
Args:
runtime: 工具的运行时
Returns:
str: 用户信息
"""
assert runtime.store is not None, "store 未配置"
item = runtime.store.get(("users",), runtime.state["user_id"])
return str(item.value) if item else "unknown"
agent = create_agent(
model=model,
tools=[save_user_info, get_user_info],
store=InMemoryStore(),
system_prompt="用户提及个人信息时及时记录，用户询问个人信息时尝试用工具检索",
state_schema=cast(type[AgentState], CustomState),
)
response_input = agent.invoke({
"messages": [HumanMessage("你好，很高兴认识你，我是小明")],
"user_id": "user-1"
})
response_output = agent.invoke({
"messages": [HumanMessage("我是谁")],
"user_id": "user-1"
})python
with PostgresStore.from_conn_string(POSTGRES_SQL_URL) as store:
agent = create_agent(
model=model,
tools=[save_user_info, get_user_info],
store=store,
system_prompt="用户提及个人信息时及时记录，用户询问个人信息时尝试用工具检索",
state_schema=cast(type[AgentState], CustomState),
)RAG
Document Loaders 文档加载器
LangChain Document loader integrations
- Document对象
- page_content：真正的文档内容，字符串类型
- metadata：文档内容的原数据，字典类型
- JSONLoader使用指定的jq结构来解析 JSON 文件
- BaseLoader作为基类，所有的文档加载器都需要继承BaseLoader，并且实现对应的load方法，加载完成后将其转成Documents对象
python
from langchain_community.document_loaders import TextLoader
text_loader = TextLoader(file_path="filePath.txt", encoding="utf-8")
text = text_loader.load()
print(text)python
from langchain_community.document_loaders.csv_loader import CSVLoader
loader = CSVLoader(file_path="filePath.csv")
csv = loader.load()
print(csv)pythom
from langchain_community.document_loaders import JSONLoader
from rich import print as rprint
json_loader = JSONLoader(
file_path="filePath.json",
# jq_schema=".", ## 提取所有字段
jq_schema=".messages[].content",
text_content=False # 保持原始 JSON 结构，将提取的数据转换为JSON字符串存入page_content字段中
)
json = json_loader.load()
rprint(json)pythom
from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader(
# 文件路径，支持本地文件和在线文件链接
# file_path="filePath.pdf",
file_path="https://arxiv.org/pdf/alg-geom/9202012",
# 提取模式:控制如何从 PDF 文件中解析和提取文本结构。
# plain 提取文本，默认值
# layout 布局感知提取模式，通常会通过插入大量的空格、换行符，来模拟原文档中的多栏、 缩进和间距（适用场景：学术论文（如arXiv论文）、多栏报刊杂志、带有左右分栏的合同）
extraction_mode="plain",
)
pdf = loader.load()
print(pdf)python
loader = UnstructuredWordDocumentLoader(
# 文件路径
file_path="filePath.docx",
# 加载模式:
# single 返回单个Document对象
# elements 按标题等元素切分文档
mode="single",
)
word = loader.load()
print(word)python
from langchain_community.document_loaders import UnstructuredMarkdownLoader
loader = UnstructuredMarkdownLoader(
file_path="filePath.md",
# 加载模式:
# single 返回单个Document对象
# elements 按标题等元素切分文档
mode="single",
# 解析策略：
# "fast"（快速模式），它会以最快的速度提取文本，不进行复杂的版面分析
# "hi_res" 高分辨率模式
strategy="fast"
)
md = loader.load()
print(md)python
from langchain_community.document_loaders import UnstructuredHTMLLoader
# strategy:
# "fast" 解析加载html文件速度是比较快（但可能丢失部分结构或元数据）
# "hi_res": (高分辨率解析) 解析精准（速度慢一些）
# "ocr_only" 强制使用ocr提取文本，仅仅适用于图像（对HTML无效）
# mode ：one of `{'paged', 'elements', 'single'}
# "elements" 按语义元素（标题、段落、列表、表格等）拆分成多个独立的小文档
loader = UnstructuredHTMLLoader(
file_path="filePath.html",
mode="elements",
strategy="fast"
)
htmls = loader.load()
for html in htmls:
print(html)
print(f"\n\n\n")pythom
from langchain_community.document_loaders import DirectoryLoader
from langchain_community.document_loaders import PythonLoader
directory_loader = DirectoryLoader(
path="../asset",
glob="*.py", # 文件匹配模式（过滤器）。使用标准的 Unix 路径通配符。
use_multithreading=True, # 是否启用多线程。填 True 意味着 LangChain 会同时并发读取多个文件。
show_progress=True, # 是否显示进度条。填 True 时，控制台在加载文件时会弹出一个进度条
loader_cls=PythonLoader # 指定底层核心加载器
)
directory = directory_loader.load()
print(directory)Text Splitters 文档切分器
langchain Text Splitters API Docs
python
python
python
python
python
Text Embedding Models 文档嵌入模型
使用硅基流动平台的嵌入模型，选择bge-m3作为嵌入模型
python
python
python
Vector Stores 向量存储
使用Milvus进行向量存储