LangChain学习笔记
·
LangChain学习笔记
参考文档：https://docs.langchain.com/oss/python/langchain/quickstart
1、LangChain介绍
1.1、核心概念
- LLM（大语言模型）：LangChain 的核心是与 LLM 的交互。LLM 接收文本输入，返回文本输出
- Prompt（提示词）：发送给 LLM 的输入文本，决定了 LLM 的输出质量
- Chain（链）：将多个组件串联起来，形成一个完整的工作流
- Memory（记忆）：让 LLM 能够记住之前的对话内容
- Document（文档）：文本数据的基本单位，包含内容和元数据
- Vector Store（向量存储）：存储文档的向量表示，用于语义检索
- Agent（智能体）：能够使用工具、进行推理和决策的 LLM 应用
1.2、LangChain
LangChain 是一个用于开发由大语言模型（LLM）驱动的应用程序的框架。它提供了一系列工具和抽象，让你能够：
- 🔗 将 LLM 与其他数据源连接
- 🧠 构建具有记忆能力的对话系统
- 🤖 创建能够使用工具的智能体（Agent）
- 📚 实现检索增强生成（RAG）系统
- ⚡ 快速原型开发和生产部署
1.3、应用场景
-
RAG（检索增强生成）：将企业内部知识库与 LLM 结合，提供基于上下文的精准回答
-
Agent（智能体）开发：构建能使用工具的自主 AI 系统，连接外部数据源和 API，自动化复杂的多步骤任务
-
聊天机器人：带记忆的多轮对话系统，如客服机器人、虚拟助手
-
数据连接与处理：自然语言查询数据库，生成数据报表和可视化
-
多模态应用：处理文本、图片、音频、视频，多模态检索和生成
-
自动化工作流：测试报告自动生成、文档处理自动化、业务流程智能化
2、环境搭建
这里选用的uv作为包管理工具
2.1、安装uv
macOS/Linux
# 1、官方安装脚本
curl -LsSf https://astral.sh/uv/install.sh | sh
uv --version
# 2、Homebrew
brew install uv
uv --version
windows
# 1、官方安装脚本
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"
uv --version
2.2、版本管理
注意uv 只能卸载它自己安装的 Python**，**不能卸载之前手动安装的 Python
# 列出所有可用的 Python 版本(简介版)
uv python list
# 安装指定版本 Python
uv python install 3.14
# 安装多个版本
uv python install 3.11 3.12 3.13
# 卸载 Python 版本
uv python uninstall 3.11
# 列出本地已安装的所有 Python 版本
uv python list --only-installed
# 设置全局python版本
uv python pin --global 3.13
2.3、项目管理
uv init agent-study
3、快速开始
3.1、依赖安装
uv add "langchain>=1.2.13"
uv add langchain-community
uv add langchain-openai
uv add langchain-anthropic
uv add langchain-deepseek
uv add dotenv
3.2、API KEY配置
openai和claude用的是中转站：https://foxcode.rjj.cc/auth/register?aff=M5APLZE
.env
DEEPSEEK_API_KEY=sk-xxxxxxxxxxxxxxxxx
DEEPSEEK_BASE_URL=https://api.deepseek.com
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxx
OPENAI_BASE_URL=https://code.newcli.com/codex/v1
ANTHROPIC_API_KEY=sk-xxxxxxxxxxxxxxxxx
ANTHROPIC_BASE_URL=https://code.newcli.com/claude
env_util.py
import os
from dotenv import load_dotenv
load_dotenv(verbose=True)
deepseek_api_key = os.getenv("DEEPSEEK_API_KEY")
deepseek_base_url = os.getenv("DEEPSEEK_BASE_URL")
openai_api_key = os.getenv("OPENAI_API_KEY")
openai_base_url = os.getenv("OPENAI_BASE_URL")
anthropic_api_key = os.getenv("ANTHROPIC_API_KEY")
anthropic_base_url = os.getenv("ANTHROPIC_BASE_URL")
3.3、模型初始化
Model Class
最常见的方式，直接导入并实例化特定提供商的模型类
- 类型提示完整，IDE 自动补全友好，可以访问提供商特定的参数和功能
- 切换模型提供商需要修改导入和代码，不适合需要动态选择模型的场景
from langchain_core.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
res = deepseek_llm.invoke([
HumanMessage(content="hi!", )
])
print(res)
if __name__ == "__main__":
main()
init_chat_model
LangChain 提供的统一初始化函数，通过字符串指定模型
- 统一接口，易于切换模型，适合从配置文件读取模型名称
- 类型提示较弱，无法使用提供商特定的高级参数
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
llm = init_chat_model(model="deepseek-chat",
model_provider="deepseek",
api_key=deepseek_api_key,
base_url=deepseek_base_url)
res = llm.invoke([
HumanMessage(content="hi!")
])
print(res.content)
if __name__ == "__main__":
main()
4、基本使用
- SystemMessage：设定AI 的行为、角色和规则，通常放在对话开头
- HumanMessage：用户（人类）发送的消息
- AIMessage：AI 助手的回复
4.1、简单对话
from langchain_core.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
res = deepseek_llm.invoke([
HumanMessage(content="hi!", )
])
print(res)
if __name__ == "__main__":
main()
4.2、多轮对话
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
res = deepseek_llm.invoke([
SystemMessage(content="你是一个 Python 编程助手"),
HumanMessage(content="如何读取文件？")
])
print(res.content)
if __name__ == "__main__":
main()
4.3、流式输出
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
messages = [
SystemMessage(content="你是一个 Python 编程助手"),
HumanMessage(content="使用python写快排算法")
]
for chunk in deepseek_llm.stream(messages):
print(chunk.content, end="", flush=True)
if __name__ == "__main__":
main()
4.4、链式调用
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个{role}"),
("human", "{input}")
])
chain = prompt | deepseek_llm
res = chain.invoke({"role": "代码审查专家", "input": """ 检查下面这段代码有什么问题？
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个{role}"),
("human", "{input}")
])
chain = prompt | deepseek_llm
res = chain.invoke({
"role": "代码审查专家",
"input": "检查下面这段代码有什么问题？"
})
print(res.content)
if __name__ == "__main__":
main()
"""})
print(res.content)
if __name__ == "__main__":
main()
4.5、批量处理
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
responses = deepseek_llm.batch([
[HumanMessage("hi!")],
[HumanMessage("介绍一下你自己")],
[HumanMessage("测试开发工程师终面一般会问些什么？")],
])
for response in responses:
print(response.content)
if __name__ == "__main__":
main()
4.6、异步调用
import asyncio
from langchain_core.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
async def ainvoke(messages):
return await deepseek_llm.ainvoke(messages)
def main():
response = asyncio.run(ainvoke([HumanMessage("hi!")]))
print(response.content)
if __name__ == "__main__":
main()
5、Template
5.1、PromptTemplate
from_template
from langchain_core.prompts import PromptTemplate
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
prompt = PromptTemplate.from_template("你是一个{role}, 请用{language}回答：{question}")
# 格式化生成最终提示词
# print(prompt.format(role="Python专家", language="中文", question="二分法示例代码"))
# 格式化生成最终提示词
# print(prompt.format_prompt(role="Python专家", language="中文", question="二分法示例代码"))
chain = prompt | deepseek_llm
res = chain.invoke({"role": "Python专家", "language": "中文", "question": "二分法示例代码"})
print(res.content)
if __name__ == "__main__":
main()
partial
import datetime
from langchain_core.prompts import PromptTemplate
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def dt():
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
template = PromptTemplate.from_template("你是一个{role}, 现在时间是{dt}, 请用{language}回答：{question}")
# 动态填充&静态填充
.partial(dt=dt, role="Python专家")
prompt = template_partial.format(language="中文", question="二分法示例代码")
res = deepseek_llm.invoke(prompt)
print(res.content)
if __name__ == "__main__":
main()
5.2、ChatPromptTemplate
from_template
简单模板，默认创建 human 消息（可以使用role指定角色）
from langchain_core.messages import HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
prompt = ChatPromptTemplate.from_template("将{text}翻译为英文")
# message = prompt.invoke({"text": "你好，世界！"})
# 返回消息列表
# print(message.to_messages())
# 返回字符串
# print(message.to_string())
# 返回字符串
# print(prompt.format(text="你好，世界！"))
# 返回消息列表
# print(prompt.format_messages(text="你好，世界！"))
chain = prompt | deepseek_llm | StrOutputParser()
content = chain.invoke({"text": "你好，世界！"})
print(content)
if __name__ == "__main__":
main()
from_messages
多轮对话模板，用于构建完整的对话结构（system是系统提示，human是用户消息，ai是LLM的回复消息，placeholder占位符用于插入消息队列）
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个{role}"),
("human", "{input}")
])
# message = prompt.invoke({"role": "代码专家", "input": "如何读取文件？"})
# 返回消息列表
# print(message.to_messages())
# 返回字符串
# print(message.to_string())
# 返回字符串
# print(prompt.format(role="你好，世界！", input="如何读取文件？"))
# 返回消息列表
# print(prompt.format_messages(role="你好，世界！", input="如何读取文件？"))
chain = prompt | deepseek_llm
res = chain.invoke({"role": "Python专家", "input": "如何读取文件？"})
print(res.content)
if __name__ == "__main__":
main()
partial
预先填充某些变量
import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def dt():
strftime = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(strftime)
return strftime
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
prompt = (ChatPromptTemplate.from_messages([
("system", "你是一个{role}， 今天是{dt}"),
("human", "{input}")
])
# 动态填充
.partial(dt=dt))
# 静态填充
prompt_partial = prompt.partial(role="Python专家")
chain = prompt_partial | deepseek_llm
res = chain.invoke({"input": "如何读取文件？"})
print(res.content)
if __name__ == "__main__":
main()
placeholder
插入消息列表
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个{role}"),
("placeholder", "{chat_history}"),
("human", "{input}")
])
chat_history = [
HumanMessage(content="我是眼眸，职位是高级测试开发工程师"),
AIMessage(content="好的"),
]
prompt_partial = prompt.partial(role="AI助手")
chain = prompt_partial | deepseek_llm
res = chain.invoke({"chat_history": chat_history, "input": "我的岗位职责是什么"})
print(res.content)
if __name__ == "__main__":
main()
6、结构化输出
6.1、pydantic
- 自动数据验证（类型、范围、格式等）
- 丰富的字段配置（默认值、别名、验证器）
- 返回类型安全的对象，有代码提示
- 可以定义自定义验证逻辑
from typing import Optional
from pydantic import BaseModel, Field
class Person(BaseModel):
name: str = Field(description="姓名")
age: int = Field(description="年龄")
city: str = Field(description="城市")
email: Optional[str] = Field(description="邮件（可选）")
positions: str = Field(description="职位")
import datetime
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from model.person import Person
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个{role}，当前时间{dt}"),
("placeholder", "{chat_history}"),
("human", "{input}")
])
chat_history = [
HumanMessage(
content="我是眼眸，2000年出生，目前就职于上海哔哩哔哩科技有限公司，职位是高级测试开发工程师，邮箱是xumeng03@bilibili.com"),
AIMessage(content="好的"),
]
prompt_partial = prompt.partial(dt=datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), role="AI助手")
structured_output = deepseek_llm.with_structured_output(Person)
chain = prompt_partial | structured_output
person = chain.invoke({"chat_history": chat_history, "input": "提取Person信息"})
print(person.model_dump())
if __name__ == "__main__":
main()
6.2、JSON Schema
pydantic模型可以直接转为json schema解构：schema = Person.model_json_schema()
- 跨语言通用标准
- 不依赖 Python 特定库
- 支持复杂的验证规则
import datetime
import json
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from model.person import Person
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个{role}，当前时间{dt}"),
("placeholder", "{chat_history}"),
("human", "{input}")
])
chat_history = [
HumanMessage(
content="我是眼眸，2000年出生，目前就职于上海哔哩哔哩科技有限公司，职位是高级测试开发工程师，邮箱是xumeng03@bilibili.com"),
AIMessage(content="好的"),
]
prompt_partial = prompt.partial(dt=datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), role="AI助手")
schema = {
"properties": {
"name": {
"description": "姓名",
"title": "Name",
"type": "string"
},
"age": {
"description": "年龄",
"title": "Age",
"type": "integer"
},
"city": {
"description": "城市",
"title": "City",
"type": "string"
},
"email": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"description": "邮件（可选）",
"title": "Email"
},
"positions": {
"description": "职位",
"title": "Positions",
"type": "string"
}
},
"required": [
"name",
"age",
"city",
"email",
"positions"
],
"title": "Person",
"type": "object"
}
structured_output = deepseek_llm.with_structured_output(schema)
chain = prompt_partial | structured_output
person = chain.invoke({"chat_history": chat_history, "input": "提取Person信息"})
print(json.dumps(person, indent=4, ensure_ascii=False))
if __name__ == "__main__":
main()
6.3、StrOutputParser
LLM 的原始输出是 AIMessage 对象，如果只想要文本内容，需要手动提取res.content，StrOutputParser 自动完成这个转换
from langchain_core.messages import HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
chain = deepseek_llm | StrOutputParser()
content = chain.invoke([
HumanMessage(content="hi!", )
])
print(content)
if __name__ == "__main__":
main()
6.4、输出解析器
import json
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_deepseek import ChatDeepSeek
from model.person import Person
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
parser = PydanticOutputParser(pydantic_object=Person)
# 格式说明，根据模型生成的一段固定的文本
format_instructions = parser.get_format_instructions()
prompt = PromptTemplate(
template="提取以下信息：{query}\n\n{format_instructions}",
# 使用时才提供的变量
input_variables=["query"],
# 创建时就固定的变量
partial_variables={"format_instructions": format_instructions},
)
chain = prompt | deepseek_llm | parser
person = chain.invoke({
"query": "眼眸，2000年出生，目前就职于上海哔哩哔哩科技有限公司，职位是高级测试开发工程师，邮箱是xumeng03@bilibili.com"})
print(person.model_dump())
if __name__ == "__main__":
main()
7、Chains
LCEL（LangChain Expression Language）使用 | 管道操作符将组件串联，每个组件都是Runnable 对象，数据从左到右流动，最常见的组合就是 prompt | llm | parser
7.1、常见的 Runnable 类型
- PromptTemplate/ChatPromptTemplate：
prompt = PromptTemplate.from_template("...") - LLM/ChatModel：
llm = ChatDeepSeek(...) - OutputParser：
parser = StrOutputParser() - Retriever：
retriever = vectorstore.as_retriever() - RunnableLambda：
RunnableLambda(lambda x: x.upper())
7.2、基础链
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
prompt = ChatPromptTemplate.from_template("将{text}翻译为英文")
chain = prompt | deepseek_llm | StrOutputParser()
# 1. invoke - 单次调用
content = chain.invoke({"text": "你好，世界！"})
print(content)
# 2. batch - 批量调用
contents = chain.batch([
{"text": "你好"},
{"text": "再见"}
])
for content in contents:
print(content)
# 3. stream - 流式输出
for chunk in chain.stream({"text": "这是一很久很久以前的故事"}):
print(chunk, end="", flush=True)
if __name__ == "__main__":
main()
7.3、顺序链
基本使用
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url,
timeout=60,
max_retries=2
)
outline_prompt = ChatPromptTemplate.from_template("为主题{topic}写一个故事大纲")
story_prompt = ChatPromptTemplate.from_template("基于大纲写一个完整的故事：{outline}")
appraise_prompt = ChatPromptTemplate.from_template("评价这个故事\n{story}")
chain = outline_prompt | deepseek_llm | StrOutputParser() | (
lambda outline: {"outline": outline}) | story_prompt | deepseek_llm | StrOutputParser() | (
lambda story: {"story": story}) | appraise_prompt | deepseek_llm | StrOutputParser()
content = chain.invoke({"topic": "时空穿越"})
print(content)
if __name__ == "__main__":
main()
保留中间结果
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url,
timeout=60,
max_retries=2
)
outline_prompt = ChatPromptTemplate.from_template("为主题{topic}写一个故事大纲")
story_prompt = ChatPromptTemplate.from_template("基于大纲写一个完整的故事：{outline}")
appraise_prompt = ChatPromptTemplate.from_template("评价这个故事\n{story}")
chain = (
# 输入: {"topic": "..."}
RunnablePassthrough.assign(
outline=outline_prompt | deepseek_llm | StrOutputParser()
) |
# 现在有: {"topic": "...", "outline": "..."}
RunnablePassthrough.assign(
story=story_prompt | deepseek_llm | StrOutputParser()
) |
# 现在有: {"topic": "...", "outline": "...", "story": "..."}
RunnablePassthrough.assign(
appraise=appraise_prompt | deepseek_llm | StrOutputParser()
)
# 最终: {"topic": "...", "outline": "...", "story": "...", "appraise": "..."}
)
result = chain.invoke({"topic": "时空穿越"})
print(result["topic"])
print(result["outline"])
print(result["story"])
print(result["appraise"])
if __name__ == "__main__":
main()
多输入顺序链
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url,
timeout=60,
max_retries=2
)
outline_prompt = ChatPromptTemplate.from_template("为主题{topic}写一个故事大纲，字数{topic_limit}字以内")
story_prompt = ChatPromptTemplate.from_template("基于大纲[{outline}]写一个{story_limit}字以内故事")
appraise_prompt = ChatPromptTemplate.from_template("评价这个故事\n{story}")
chain = (
# 输入: {"topic": "..."}
RunnablePassthrough.assign(
outline=outline_prompt | deepseek_llm | StrOutputParser()
) |
# 现在有: {"topic": "...", "outline": "..."}
RunnablePassthrough.assign(
story=story_prompt | deepseek_llm | StrOutputParser()
) |
# 现在有: {"topic": "...", "outline": "...", "story": "..."}
RunnablePassthrough.assign(
appraise=appraise_prompt | deepseek_llm | StrOutputParser()
)
# 最终: {"topic": "...", "outline": "...", "story": "...", "appraise": "..."}
)
result = chain.invoke({"topic": "时空穿越", "topic_limit": 100, "story_limit": 1000})
print(result["topic"])
print(result["outline"])
print(result["story"])
print(result["appraise"])
if __name__ == "__main__":
main()
7.4、数据转换链
RunnableLambda
RunnableLambda 可以将任何Python 函数转换为 Runnable 对象，融入链中
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def uppercase(s: str) -> str:
return s.upper()
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
transform = RunnableLambda(uppercase)
prompt = ChatPromptTemplate.from_template("将{text}翻译为英文")
chain = prompt | deepseek_llm | StrOutputParser() | transform
content = chain.invoke({"text": "你好，世界！"})
print(content)
if __name__ == "__main__":
main()
lambda
可以直接使用 lambda，会自动包装为 RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def uppercase(s: str) -> str:
return s.upper()
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
transform = RunnableLambda(uppercase)
prompt = ChatPromptTemplate.from_template("将{text}翻译为英文")
chain = prompt | deepseek_llm | StrOutputParser() | transform | (lambda s: s.lower())
content = chain.invoke({"text": "你好，世界！"})
print(content)
if __name__ == "__main__":
main()
RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def uppercase(s: str) -> str:
return s.upper()
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
transform = RunnableLambda(uppercase)
prompt = ChatPromptTemplate.from_template("将{text}翻译为英文")
chain = (
RunnablePassthrough.assign(
normal=prompt | deepseek_llm | StrOutputParser() | transform
) |
RunnablePassthrough.assign(
lower=lambda x: x["text"].lower(),
)
)
x = chain.invoke({"text": "你好，世界！"})
print(x)
if __name__ == "__main__":
main()
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def uppercase(s: str) -> str:
return s.upper()
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
transform = RunnableLambda(uppercase)
prompt = ChatPromptTemplate.from_template("将{text}翻译为英文")
chain = (
RunnablePassthrough.assign(
normal=prompt | deepseek_llm | StrOutputParser() | transform
) |
RunnablePassthrough.assign(
lower=lambda x: x["text"].lower(),
) |
RunnablePassthrough().pick(keys=["normal", "lower"])
)
x = chain.invoke({"text": "你好，世界！"})
print(x)
if __name__ == "__main__":
main()
7.5、条件顺序链
RunnableBranch
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
chain = (
RunnableBranch(
(
lambda x: x["type"] == "code",
ChatPromptTemplate.from_template("解释这段代码: \n{content}") | deepseek_llm | StrOutputParser()
),
(
lambda x: x["type"] == "text",
ChatPromptTemplate.from_template("总结这段内容: \n{content}") | deepseek_llm | StrOutputParser()
),
ChatPromptTemplate.from_template("原样输出用户输入的{content}") | deepseek_llm | StrOutputParser()
)
)
res = chain.invoke({"type": "13213", "content": "ChatPromptTemplate.from_template(\"解释这段代码：{content}\")"})
print(res)
if __name__ == "__main__":
main()
多步骤条件链
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch, RunnablePassthrough
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
chain = (
RunnablePassthrough.assign(
fruit=ChatPromptTemplate.from_template(
"根据用户描述分析用户更喜欢哪种水果（回复：apple/grape）: \n{content}") | deepseek_llm | StrOutputParser()
) |
RunnableBranch(
(
lambda x: x["fruit"] == "apple",
ChatPromptTemplate.from_template("介绍一下水果苹果") | deepseek_llm | StrOutputParser()
),
(
lambda x: x["fruit"] == "grape",
ChatPromptTemplate.from_template("介绍一下水果葡萄") | deepseek_llm | StrOutputParser()
),
ChatPromptTemplate.from_template("原样输出用户输入的{content}") | deepseek_llm | StrOutputParser()
)
)
res = chain.invoke({"content": "我喜欢吃葡萄，但是如果跟苹果相比，我更喜欢后者"})
print(res)
if __name__ == "__main__":
main()
8、Memory
LLM 本身是无状态的，每次调用都是独立的，Memory 让对话具有连续性
8.1、完整记忆
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
store = {}
def get_session_history(session_id: str):
if session_id not in store:
store[session_id] = ChatMessageHistory()
return store[session_id]
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个AI助手"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
chain1 = prompt | deepseek_llm | StrOutputParser()
chain2 = RunnableWithMessageHistory(
chain1,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
res = chain2.invoke({"input": "我是眼眸，职位是高级测试开发工程师"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我的岗位职责是什么"}, config={"configurable": {"session_id": "user1"}})
print(res)
if __name__ == "__main__":
main()
8.2、窗口记忆
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
store = {}
def get_session_history(session_id: str, k: int = 2):
if session_id not in store:
store[session_id] = ChatMessageHistory()
history = store[session_id]
messages = history.messages
if len(messages) > k:
trimmed = messages[-k:]
history.clear()
for trim in trimmed:
history.add_message(trim)
return history
def main():
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个AI助手"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
chain1 = prompt | deepseek_llm | StrOutputParser()
chain2 = RunnableWithMessageHistory(
chain1,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
res = chain2.invoke({"input": "我是眼眸，职位是高级测试开发工程师"},
config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我最喜欢的水果是苹果"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我最喜欢的颜色是青色"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我最喜欢的颜色是青色"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我的岗位是什么？"}, config={"configurable": {"session_id": "user1"}})
print(res)
if __name__ == "__main__":
main()
8.3、Token限制记忆
trim_messages 的 token_counter 参数需要模型实现 get_num_tokens_from_messages() 方法，但 DeepSeek 没有实现
from langchain_core.messages import trim_messages
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_deepseek import ChatDeepSeek
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
store = {}
def get_session_history(session_id: str, k: int = 1000):
if session_id not in store:
store[session_id] = ChatMessageHistory()
history = store[session_id]
messages = history.messages
trimmed = trim_messages(messages, max_tokens=k, strategy="last", token_counter=deepseek_llm)
if len(trimmed) > len(history.messages):
history.clear()
for trim in trimmed:
history.add_message(trim)
print("===============start==================")
print(history)
print("================end===================")
return history
def main():
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个AI助手"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
chain1 = prompt | deepseek_llm | StrOutputParser()
chain2 = RunnableWithMessageHistory(
chain1,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
res = chain2.invoke({"input": "我是眼眸，职位是高级测试开发工程师"},
config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我最喜欢的水果是苹果"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我最喜欢的颜色是青色"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我最喜欢的颜色是青色"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我的岗位是什么？"}, config={"configurable": {"session_id": "user1"}})
print(res)
if __name__ == "__main__":
main()
8.4、摘要记忆
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_deepseek import ChatDeepSeek
class SummaryMemory(BaseChatMessageHistory):
def __init__(self, llm: ChatDeepSeek, max_messages: int = 10):
super().__init__()
self.llm = llm
self.max_messages = max_messages
self.summary = ""
self.recent_messages = []
@property
def messages(self) -> list[BaseMessage]:
"""返回所有消息（摘要 + 最近消息）"""
result = []
if self.summary:
result.append(SystemMessage(content=f"对话摘要：{self.summary}"))
result.extend(self.recent_messages)
return result
def add_user_message(self, message: str) -> None:
"""添加用户消息"""
self.recent_messages.append(HumanMessage(content=message))
self._check_and_summarize()
def add_ai_message(self, message: str) -> None:
"""添加 AI 消息"""
self.recent_messages.append(AIMessage(content=message))
self._check_and_summarize()
def add_message(self, message: BaseMessage) -> None:
"""添加消息"""
self.recent_messages.append(message)
self._check_and_summarize()
def clear(self) -> None:
"""清空历史"""
self.summary = ""
self.recent_messages = []
def _check_and_summarize(self):
"""检查是否需要生成摘要"""
if len(self.recent_messages) > self.max_messages:
self._create_summary()
def _create_summary(self):
"""生成摘要并保留最近的消息"""
old_messages = self.recent_messages[:-4]
formatted_msgs = "\n".join([
f"{'用户' if isinstance(msg, HumanMessage) else 'AI'}: {msg.content}"
for msg in old_messages
])
summary_prompt = f"""当前摘要：{self.summary if self.summary else '无'}
新的对话：
{formatted_msgs}
请更新摘要，简洁地总结关键信息："""
res = self.llm.invoke(summary_prompt)
self.summary = res.content
self.recent_messages = self.recent_messages[-4:]
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_deepseek import ChatDeepSeek
from model.summary_memory import SummaryMemory
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
store = {}
def get_session_history(session_id: str):
if session_id not in store:
store[session_id] = SummaryMemory(
llm=deepseek_llm,
max_messages=2
)
print("=====================start=======================")
print(store[session_id])
print("======================end========================")
return store[session_id]
def main():
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个AI助手"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
chain1 = prompt | deepseek_llm | StrOutputParser()
chain2 = RunnableWithMessageHistory(
chain1,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
res = chain2.invoke({"input": "我是眼眸，职位是高级测试开发工程师"},
config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我最喜欢的水果是苹果"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我最喜欢的颜色是青色"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我最喜欢的颜色是青色"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我的岗位是什么？"}, config={"configurable": {"session_id": "user1"}})
print(res)
if __name__ == "__main__":
main()
8.5、持久化存储
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_deepseek import ChatDeepSeek
from model.summary_memory import SummaryMemory
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
store = {}
def get_session_history(session_id: str):
return RedisChatMessageHistory(
session_id=session_id,
url=f"redis://:xxxxxxxx@124.222.145.150:6379",
)
def main():
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个AI助手"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
chain1 = prompt | deepseek_llm | StrOutputParser()
chain2 = RunnableWithMessageHistory(
chain1,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
res = chain2.invoke({"input": "我是眼眸，职位是高级测试开发工程师"},
config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我最喜欢的水果是苹果"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我最喜欢的颜色是青色"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我最喜欢的颜色是青色"}, config={"configurable": {"session_id": "user1"}})
print(res)
res = chain2.invoke({"input": "我的岗位是什么？"}, config={"configurable": {"session_id": "user1"}})
print(res)
if __name__ == "__main__":
main()
9、工具
工具（Tool） 是连接 LLM 与外部世界的桥梁，LLM 本身只能进行推理，无法直接执行数据库查询、API 调用或文件操作。工具将普通的 Python 函数转化为 LLM 可以理解、选择和调用的格式
用户问题 → HumanMessage
↓
LLM 决定调用工具 → AIMessage (包含 tool_calls)
↓
手动执行工具 → ToolMessage (工具返回结果)
↓
将所有消息传回 LLM → AIMessage (最终答案)
9.1、基本使用
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek
from model.summary_memory import SummaryMemory
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
store = {}
@tool
def get_weather(city: str):
"""根据城市名查询天气。如果用户询问特定城市的天气，请使用此工具。"""
return f"{city} 的天气晴朗，气温 25°C。"
def get_session_history(session_id: str):
return RedisChatMessageHistory(
session_id=session_id,
url=f"redis://:xxxxxxxx@124.222.145.150:6379",
)
def main():
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个AI助手"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
tools = [get_weather]
llm_bind_tools = deepseek_llm.bind_tools(tools)
chain1 = prompt | llm_bind_tools
chain2 = RunnableWithMessageHistory(
chain1,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
res = chain2.invoke({"input": "今天上海的天气怎么样？"}, config={"configurable": {"session_id": "user1"}})
if res.tool_calls:
print("LLM 想调用工具:")
print(res.tool_calls)
else:
print(res.content)
if __name__ == "__main__":
main()
9.2、处理工具调用
单个工具调用
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.messages import ToolMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek
from model.summary_memory import SummaryMemory
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
store = {}
@tool
def get_weather(city: str):
"""根据城市名查询天气。如果用户询问特定城市的天气，请使用此工具。"""
return f"{city} 的天气晴朗，气温 25°C。"
def get_session_history(session_id: str):
return RedisChatMessageHistory(
session_id=session_id,
url=f"redis://:xxxxxxxx@124.222.145.150:6379",
)
def main():
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个AI助手"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
tools = [get_weather]
llm_bind_tools = deepseek_llm.bind_tools(tools)
chain1 = prompt | llm_bind_tools
chain2 = RunnableWithMessageHistory(
chain1,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
res = chain2.invoke({"input": "今天上海的天气怎么样？"}, config={"configurable": {"session_id": "user1"}})
print("LLM思考结果: ", res)
if res.tool_calls:
print("LLM 想调用工具", res.tool_calls)
history = get_session_history("user1")
for tool_call in res.tool_calls:
if tool_call["name"] == "get_weather":
result = get_weather.invoke(tool_call["args"])
print("工具调用结果: ", result)
tool_message = ToolMessage(
content=str(result),
tool_call_id=tool_call["id"],
)
history.add_message(tool_message)
res = chain2.invoke({"input": ""}, config={"configurable": {"session_id": "user1"}})
print("LLM思考结果: ", res)
else:
print(res.content)
if __name__ == "__main__":
main()
多个工具调用
一定要将所有的工具调用完才能调用模型
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.messages import ToolMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek
from model.summary_memory import SummaryMemory
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
store = {}
@tool
def get_weather(city: str):
"""根据城市名查询天气。如果用户询问特定城市的天气，请使用此工具。"""
return f"{city} 的天气晴朗，气温 25°C。"
@tool
def get_activity(city: str):
"""根据城市名查询活动。如果用户询问特定城市的活动，请使用此工具。"""
return f"{city} 的活动有玉兰花拍摄、樱花拍摄"
def get_session_history(session_id: str):
return RedisChatMessageHistory(
session_id=session_id,
url=f"redis://:xxxxxxxx@124.222.145.150:6379",
)
def main():
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个AI助手"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
tools = [get_weather, get_activity]
llm_bind_tools = deepseek_llm.bind_tools(tools)
chain1 = prompt | llm_bind_tools
chain2 = RunnableWithMessageHistory(
chain1,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
res = chain2.invoke({"input": "今天上海的天气怎么样？顺便帮我查询一下最近的活动"},
config={"configurable": {"session_id": "user1"}})
print("LLM思考结果: ", res)
if res.tool_calls:
print("LLM 想调用工具", res.tool_calls)
history = get_session_history("user1")
for tool_call in res.tool_calls:
if tool_call["name"] == "get_weather":
result = get_weather.invoke(tool_call["args"])
print("工具调用结果: ", result)
tool_message = ToolMessage(
content=str(result),
tool_call_id=tool_call["id"],
)
history.add_message(tool_message)
if tool_call["name"] == "get_activity":
result = get_activity.invoke(tool_call["args"])
print("工具调用结果: ", result)
tool_message = ToolMessage(
content=str(result),
tool_call_id=tool_call["id"],
)
history.add_message(tool_message)
res = chain2.invoke({"input": ""}, config={"configurable": {"session_id": "user1"}})
print("LLM思考结果: ", res)
else:
print(res.content)
if __name__ == "__main__":
main()
10、Agent
Agent 是一个使用大语言模型（LLM）作为推理引擎的系统，它能够自主决定采取哪些行动、执行顺序以及如何解释结果来达成目标，并且具有自动重试和纠错能力，当工具调用失败时，Agent 会自动分析错误原因并重试
1. 输入（Input）：用户提供查询或指令
2. 思考（Thought）：LLM 分析输入，查看可用工具，决定行动计划或选择特定工具
3. 行动（Action）：Agent 执行选定的工具并传入参数
4. 观察（Observation）：系统获取工具输出并反馈给 LLM
5. 迭代（Iteration）：LLM 使用观察结果更新上下文，然后决定：
- 执行另一个行动（返回步骤 2）
- 提供最终答案（目标已达成）
10.1、基本使用
单论对话
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_deepseek import ChatDeepSeek
from tool.activity_tool import get_activity
from tool.weather_tool import get_weather
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
def main():
# 定义工具
tools = [get_weather, get_activity]
# 创建 agent（最简单的方式）
agent = create_agent(model=deepseek_llm, tools=tools)
# 执行 agent - 使用正确的消息格式
result = agent.invoke({"messages": [HumanMessage(content="北京的天气怎么样？")]})
# 打印结果
for message in result["messages"]:
if isinstance(message, SystemMessage):
print(f"😊 系统: {message.content}")
elif isinstance(message, HumanMessage):
print(f"👤 用户: {message.content}")
elif isinstance(message, AIMessage):
print(f"🤖 AI: {message.content}")
elif isinstance(message, ToolMessage):
print(f"🔧 工具: {message.content}")
if __name__ == "__main__":
main()
多轮对话
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import MemorySaver
from tool.activity_tool import get_activity
from tool.weather_tool import get_weather
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
def main():
# 定义工具
tools = [get_weather, get_activity]
# 创建内存保存器
memory = MemorySaver()
# 创建 agent，添加 checkpointer 以保存对话历史
agent = create_agent(model=deepseek_llm, tools=tools, checkpointer=memory)
# 执行 agent - 使用正确的消息格式
result = agent.invoke({"messages": [HumanMessage(content="北京的天气怎么样？")]},
config={"configurable": {"thread_id": "user1"}})
for message in result["messages"]:
print(message)
print("=================================")
result = agent.invoke({"messages": [HumanMessage(content="我刚才问了什么？")]},
config={"configurable": {"thread_id": "user1"}})
for message in result["messages"]:
print(message)
if __name__ == "__main__":
main()
结构化输出
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import MemorySaver
from model.person import Person
from tool.activity_tool import get_activity
from tool.weather_tool import get_weather
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
def main():
# 定义工具
tools = [get_weather, get_activity]
# 创建内存保存器
memory = MemorySaver()
# 创建 agent，添加 checkpointer 以保存对话历史
agent = create_agent(model=deepseek_llm, tools=tools, checkpointer=memory, response_format=Person)
# 执行 agent - 使用正确的消息格式
result = agent.invoke({"messages": [HumanMessage(
content="我是眼眸，2000年出生，目前就职于上海哔哩哔哩科技有限公司，职位是高级测试开发工程师，邮箱是xumeng03@bilibili.com")]},
config={"configurable": {"thread_id": "user1"}})
# 打印结果
person = result["structured_response"]
print(person.model_dump_json())
if __name__ == "__main__":
main()
10.2、Tools
所有的工具都由三部分组成：工具名称、工具描述、工具函数
使用 @tool 装饰器
from langchain_core.tools import tool
@tool
def get_weather(city: str):
"""根据城市名查询天气。如果用户询问特定城市的天气，请使用此工具。"""
return f"{city} 的天气晴朗，气温 25°C。"
使用 Tool 类
from langchain_core.tools import tool, Tool
def get_activity(city: str):
return f"{city} 的活动有玉兰花拍摄、樱花拍摄"
activity = Tool(
name="get_activity",
description="根据城市名查询活动。如果用户询问特定城市的活动，请使用此工具。",
func=get_activity
)
继承 BaseTool
from langchain_core.tools import BaseTool
class ActivityTool(BaseTool):
name = "get_activity",
description = "根据城市名查询活动。如果用户询问特定城市的活动，请使用此工具。",
def _run(self, city: str) -> str:
return f"{city} 的活动有玉兰花拍摄、樱花拍摄"
10.3、Agent Prompts
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import MemorySaver
from tool.activity_tool import get_activity
from tool.weather_tool import get_weather
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
def main():
# 定义工具
tools = [get_weather, get_activity]
# 创建内存保存器
memory = MemorySaver()
# 创建 agent，添加 checkpointer 以保存对话历史
agent = create_agent(
tools=tools,
model=deepseek_llm,
system_prompt="""你是一个专业的助手。
工具使用规则：
- 涉及天气查询，必须调用get_weather工具，不要猜测
- 涉及活动推荐，必须调用get_activity工具
- 如果工具返回错误，明确告知用户查询失败
- 不要编造数据
回答要求：
- 基于工具返回的真实数据回答
- 如果没有相关工具，明确说明无法查询
""",
checkpointer=memory,
)
# 执行 agent - 使用正确的消息格式
result = agent.invoke({"messages": [HumanMessage("今天上海的天气怎么样？")]},
config={"configurable": {"thread_id": "user1"}, "recursion_limit": 5})
# 打印结果
for message in result["messages"]:
if isinstance(message, SystemMessage):
print(f"😊 系统: {message.content}")
elif isinstance(message, HumanMessage):
print(f"👤 用户: {message.content}")
elif isinstance(message, AIMessage):
print(f"🤖 AI: {message.content}")
elif isinstance(message, ToolMessage):
print(f"🔧 工具: {message.content}")
if __name__ == "__main__":
main()
10.4、Output Parsers
Output Parser 是 Agent 系统的"翻译官"，负责将 LLM 的文本输出转换为程序可以理解和执行的结构化数据
import datetime
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_core.output_parsers import PydanticOutputParser
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import MemorySaver
from model.person import Person
from tool.activity_tool import get_activity
from tool.weather_tool import get_weather
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
def main():
# 定义工具
tools = [get_weather, get_activity]
# 创建内存保存器
memory = MemorySaver()
# OutputParser
parser = PydanticOutputParser(pydantic_object=Person)
# 创建 agent，添加 checkpointer 以保存对话历史
agent = create_agent(
tools=tools,
model=deepseek_llm,
system_prompt=f"""当前时间是{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}, 你是一个专业的助手。
工具使用规则：
- 涉及天气查询，必须调用get_weather工具，不要猜测
- 涉及活动推荐，必须调用get_activity工具
- 如果工具返回错误，明确告知用户查询失败
- 不要编造数据
回答要求：
- 基于工具返回的真实数据回答
- 如果没有相关工具，明确说明无法查询
输出要求：
- {parser.get_format_instructions()}
""",
checkpointer=memory,
)
# 执行 agent - 使用正确的消息格式
result = agent.invoke({"messages": [HumanMessage(
"我是眼眸，2000年出生，目前就职于上海哔哩哔哩科技有限公司，职位是高级测试开发工程师，邮箱是xumeng03@bilibili.com")]},
config={"configurable": {"thread_id": "user1"}, "recursion_limit": 5})
# 打印结果
for message in result["messages"]:
if isinstance(message, SystemMessage):
print(f"😊 系统: {message.content}")
elif isinstance(message, HumanMessage):
print(f"👤 用户: {message.content}")
elif isinstance(message, AIMessage):
print(f"🤖 AI: {message.content}")
elif isinstance(message, ToolMessage):
print(f"🔧 工具: {message.content}")
if __name__ == "__main__":
main()
10.5、Agent 模式
新版不需要选择Agent类，但需要根据场景设计prompt
ReAct模式
默认就是ReAct模式，特点是推理（Reasoning）→ 行动（Acting）→ 观察（Observation）循环
import datetime
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_core.output_parsers import PydanticOutputParser
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import MemorySaver
from model.person import Person
from tool.activity_tool import get_activity
from tool.weather_tool import get_weather
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
def main():
# 定义工具
tools = [get_weather, get_activity]
# 创建内存保存器
memory = MemorySaver()
# OutputParser
parser = PydanticOutputParser(pydantic_object=Person)
# 创建 agent，添加 checkpointer 以保存对话历史
agent = create_agent(
tools=tools,
model=deepseek_llm,
system_prompt=f"""当前时间是{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}, 你是一个专业的助手。
工具使用规则：
- 涉及天气查询，必须调用get_weather工具，不要猜测
- 涉及活动推荐，必须调用get_activity工具
- 如果工具返回错误，明确告知用户查询失败
- 不要编造数据
回答要求：
- 基于工具返回的真实数据回答
- 如果没有相关工具，明确说明无法查询
输出要求：
- {parser.get_format_instructions()}
""",
checkpointer=memory,
)
# 执行 agent - 使用正确的消息格式
result = agent.invoke({"messages": [HumanMessage(
"我是眼眸，2000年出生，目前就职于上海哔哩哔哩科技有限公司，职位是高级测试开发工程师，邮箱是xumeng03@bilibili.com")]},
config={"configurable": {"thread_id": "user1"}, "recursion_limit": 5})
# 打印结果
for message in result["messages"]:
if isinstance(message, SystemMessage):
print(f"😊 系统: {message.content}")
elif isinstance(message, HumanMessage):
print(f"👤 用户: {message.content}")
elif isinstance(message, AIMessage):
print(f"🤖 AI: {message.content}")
elif isinstance(message, ToolMessage):
print(f"🔧 工具: {message.content}")
if __name__ == "__main__":
main()
Plan-and-Execute模式
Plan-and-Execute模式特点是先制定完整计划，再逐步执行
import datetime
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_core.output_parsers import PydanticOutputParser
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import MemorySaver
from model.person import Person
from tool.activity_tool import get_activity
from tool.weather_tool import get_weather
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
def main():
# 定义工具
tools = [get_weather, get_activity]
# 创建内存保存器
memory = MemorySaver()
# 创建 agent，添加 checkpointer 以保存对话历史
agent = create_agent(
tools=tools,
model=deepseek_llm,
system_prompt=f"""当前时间是{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}, 你是一个规划型助手。
工作流程：
1. 首先制定完整的执行计划
2. 列出需要调用的工具和顺序
3. 按计划逐步执行
4. 汇总结果
示例：
用户问："上海天气如何，适合什么活动？"
计划：
- 步骤1：调用get_weather获取上海天气
- 步骤2：根据天气调用get_activity获取活动建议
- 步骤3：整合信息回答用户
""",
checkpointer=memory,
debug=True,
)
# 执行 agent - 使用正确的消息格式
result = agent.invoke({"messages": [HumanMessage("上海天气如何，适合什么活动？")]},
config={"configurable": {"thread_id": "user1"}, "recursion_limit": 25})
# 打印结果
for message in result["messages"]:
if isinstance(message, SystemMessage):
print(f"😊 系统: {message.content}")
elif isinstance(message, HumanMessage):
print(f"👤 用户: {message.content}")
elif isinstance(message, AIMessage):
print(f"🤖 AI: {message.content}")
elif isinstance(message, ToolMessage):
print(f"🔧 工具: {message.content}")
if __name__ == "__main__":
main()
对话模式
对话模式特点是使用memory保持上下文，多轮对话
import datetime
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_core.output_parsers import PydanticOutputParser
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import MemorySaver
from model.person import Person
from tool.activity_tool import get_activity
from tool.weather_tool import get_weather
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
def main():
# 定义工具
tools = [get_weather, get_activity]
# 创建内存保存器
memory = MemorySaver()
# 创建 agent，添加 checkpointer 以保存对话历史
agent = create_agent(
tools=tools,
model=deepseek_llm,
system_prompt=f"""当前时间是{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}, 你是一个AI助手。
工具使用规则：
- 涉及天气查询，必须调用get_weather工具，不要猜测
- 涉及活动推荐，必须调用get_activity工具
- 如果工具返回错误，明确告知用户查询失败
- 不要编造数据
回答要求：
- 基于工具返回的真实数据回答
- 如果没有相关工具，明确说明无法查询
""",
checkpointer=memory,
debug=True,
)
# 执行 agent - 使用正确的消息格式
result = agent.invoke({"messages": [HumanMessage("上海天气如何，适合什么活动？")]},
config={"configurable": {"thread_id": "user1"}, "recursion_limit": 25})
result = agent.invoke({"messages": [HumanMessage("我的时间有限，只能参加一个活动，帮我推荐一个")]},
config={"configurable": {"thread_id": "user1"}, "recursion_limit": 25})
# 打印结果
for message in result["messages"]:
if isinstance(message, SystemMessage):
print(f"😊 系统: {message.content}")
elif isinstance(message, HumanMessage):
print(f"👤 用户: {message.content}")
elif isinstance(message, AIMessage):
print(f"🤖 AI: {message.content}")
elif isinstance(message, ToolMessage):
print(f"🔧 工具: {message.content}")
if __name__ == "__main__":
main()
工具优先模式
工具优先模式的特点是强制使用工具，避免幻觉
import datetime
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_core.output_parsers import PydanticOutputParser
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import MemorySaver
from model.person import Person
from tool.activity_tool import get_activity
from tool.weather_tool import get_weather
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
def main():
# 定义工具
tools = [get_weather, get_activity]
# 创建内存保存器
memory = MemorySaver()
# 创建 agent，添加 checkpointer 以保存对话历史
agent = create_agent(
tools=tools,
model=deepseek_llm,
system_prompt=f"""当前时间是{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}, 你是一个严格的助手。
铁律：
- 任何事实性问题必须使用工具查询
- 禁止猜测或编造信息
- 如果没有合适的工具，明确说"无法查询"
- 不要基于常识回答，必须基于工具返回的数据
示例：
❌ 错误："上海今天应该是晴天"（猜测）
✅ 正确：调用get_weather("上海")，然后基于返回结果回答
""",
checkpointer=memory,
debug=True,
)
# 执行 agent - 使用正确的消息格式
result = agent.invoke({"messages": [HumanMessage("上海天气如何，适合什么活动？")]},
config={"configurable": {"thread_id": "user1"}, "recursion_limit": 25})
result = agent.invoke({"messages": [HumanMessage("我的时间有限，只能参加一个活动，帮我推荐一个")]},
config={"configurable": {"thread_id": "user1"}, "recursion_limit": 25})
# 打印结果
for message in result["messages"]:
if isinstance(message, SystemMessage):
print(f"😊 系统: {message.content}")
elif isinstance(message, HumanMessage):
print(f"👤 用户: {message.content}")
elif isinstance(message, AIMessage):
print(f"🤖 AI: {message.content}")
elif isinstance(message, ToolMessage):
print(f"🔧 工具: {message.content}")
if __name__ == "__main__":
main()
10.6、Agent Memory
旧版Memory对象存储对话历史，新版Checkpointer保存整个Agent状态（包括消息、中间推理步骤、工具执行历史、自定义变量等）
MemorySaver
import datetime
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_core.output_parsers import PydanticOutputParser
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import MemorySaver
from model.person import Person
from tool.activity_tool import get_activity
from tool.weather_tool import get_weather
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url
)
def main():
# 定义工具
tools = [get_weather, get_activity]
# 创建内存保存器
memory = MemorySaver()
# 创建 agent，添加 checkpointer 以保存对话历史
agent = create_agent(
tools=tools,
model=deepseek_llm,
system_prompt=f"""当前时间是{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}, 你是一个严格的助手。
铁律：
- 任何事实性问题必须使用工具查询
- 禁止猜测或编造信息
- 如果没有合适的工具，明确说"无法查询"
- 不要基于常识回答，必须基于工具返回的数据
示例：
❌ 错误："上海今天应该是晴天"（猜测）
✅ 正确：调用get_weather("上海")，然后基于返回结果回答
""",
checkpointer=memory,
debug=True,
)
# 执行 agent - 使用正确的消息格式
result = agent.invoke({"messages": [HumanMessage("上海天气如何，适合什么活动？")]},
config={"configurable": {"thread_id": "user1"}, "recursion_limit": 25})
result = agent.invoke({"messages": [HumanMessage("我的时间有限，只能参加一个活动，帮我推荐一个")]},
config={"configurable": {"thread_id": "user1"}, "recursion_limit": 25})
# 打印结果
for message in result["messages"]:
if isinstance(message, SystemMessage):
print(f"😊 系统: {message.content}")
elif isinstance(message, HumanMessage):
print(f"👤 用户: {message.content}")
elif isinstance(message, AIMessage):
print(f"🤖 AI: {message.content}")
elif isinstance(message, ToolMessage):
print(f"🔧 工具: {message.content}")
if __name__ == "__main__":
main()
PostgresSaver
docker部署postgres
docker run -d -p 35432:5432 --name postgres \
-e POSTGRES_DB=agent \
-e POSTGRES_USER=root \
-e POSTGRES_PASSWORD=xxxxxxxx \
-v postgres-data:/var/lib/postgresql/data \
postgres:16
安装依赖
uv add "psycopg[binary]"
uv add "langgraph-checkpoint-postgres[binary]"
PostgresSaver使用
import datetime
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver
from tool.activity_tool import get_activity
from tool.weather_tool import get_weather
from util.env_util import deepseek_api_key, deepseek_base_url
deepseek_llm = ChatOpenAI(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url,
timeout=30,
max_retries=2
)
def main():
# 定义工具
tools = [get_weather, get_activity]
# 创建内存保存器
conn_string = "postgresql://root:xxxxxxxx@124.222.145.150:5432/agent"
# 使用 with 的好处是自动处理资源清理，不需要手动调用 close() 或 aclose()
with PostgresSaver.from_conn_string(conn_string) as checkpointer:
checkpointer.setup()
# 创建 agent，添加 checkpointer 以保存对话历史
agent = create_agent(
tools=tools,
model=deepseek_llm,
system_prompt=f"""当前时间是{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}, 你是一个AI助手。
工具使用规则：
- 涉及天气查询，必须调用get_weather工具，不要猜测
- 涉及活动推荐，必须调用get_activity工具
- 如果工具返回错误，明确告知用户查询失败
- 不要编造数据
回答要求：
- 基于工具返回的真实数据回答
- 如果没有相关工具，明确说明无法查询
""",
checkpointer=checkpointer,
debug=True,
)
# 执行 agent - 使用正确的消息格式
result = agent.invoke({"messages": [HumanMessage("上海天气如何，适合什么活动？")]},
config={"configurable": {"thread_id": "user1"}})
print(result)
result = agent.invoke({"messages": [HumanMessage("我的时间有限，只能参加一个活动，帮我推荐一个")]},
config={"configurable": {"thread_id": "user1"}, "recursion_limit": 25})
print(result)
# 打印结果
for message in result["messages"]:
if isinstance(message, SystemMessage):
print(f"😊 系统: {message.content}")
elif isinstance(message, HumanMessage):
print(f"👤 用户: {message.content}")
elif isinstance(message, AIMessage):
print(f"🤖 AI: {message.content}")
elif isinstance(message, ToolMessage):
print(f"🔧 工具: {message.content}")
if __name__ == "__main__":
main()
PostgresSaver存储详解
基本概念了解
Channel：通道，存储不同类型数据的"容器"
- messages 通道：存放对话消息
- agent_state 通道：存放 agent 的内部状态
- __start__ 通道：存放起始输入
Node：节点，执行流程中的各个步骤
- model 节点：调用 LLM 生成回复
- tools 节点：执行工具调用
- __start__ 节点：开始节点
执行流程分析
select * from checkpoints where thread_id = 'user1' order by checkpoint_id asc;
第一次调用："上海天气如何，适合什么活动？"
这个问题需要调用两个工具：get_weather 和 get_activity
Checkpoint 1: 初始状态
├─ 触发时机: agent.invoke 开始
├─ 状态内容: {"messages": [HumanMessage("上海天气如何，适合什么活动？")]}
├─ next: "agent"
└─ parent: null
↓ [agent 节点执行：LLM 分析问题]
Checkpoint 2: Agent 决策完成
├─ 触发时机: LLM 决定调用 get_weather 工具
├─ 状态内容:
│ - messages: [用户消息, AIMessage(tool_calls=[get_weather])]
├─ next: "tools"
└─ parent: checkpoint_1
↓ [tools 节点执行：调用 get_weather("上海")]
Checkpoint 3: 第一个工具执行完成
├─ 触发时机: get_weather 返回结果
├─ 状态内容:
│ - messages: [..., ToolMessage(name="get_weather", content="上海晴天...")]
├─ next: "agent"
└─ parent: checkpoint_2
↓ [agent 节点执行：LLM 看到天气结果]
Checkpoint 4: Agent 再次决策
├─ 触发时机: LLM 决定调用 get_activity 工具
├─ 状态内容:
│ - messages: [..., AIMessage(tool_calls=[get_activity])]
├─ next: "tools"
└─ parent: checkpoint_3
↓ [tools 节点执行：调用 get_activity("上海", "晴天")]
Checkpoint 5: 第二个工具执行完成
├─ 触发时机: get_activity 返回结果
├─ 状态内容:
│ - messages: [..., ToolMessage(name="get_activity", content="推荐活动...")]
├─ next: "agent"
└─ parent: checkpoint_4
↓ [agent 节点执行：LLM 生成最终回复]
Checkpoint 6: 第一轮对话完成
├─ 触发时机: LLM 生成最终回复，不再调用工具
├─ 状态内容:
│ - messages: [..., AIMessage(content="上海今天晴天，适合户外活动...")]
├─ next: null (END)
└─ parent: checkpoint_5
第二次调用："我的时间有限，只能参加一个活动，帮我推荐一个"
这个问题基于之前的上下文，可能不需要再调用工具
Checkpoint 7: 新一轮对话开始
├─ 触发时机: 第二次 agent.invoke 开始
├─ 状态内容:
│ - messages: [checkpoint_6 的所有历史消息 + 新的 HumanMessage]
├─ next: "agent"
└─ parent: checkpoint_6
↓ [agent 节点执行：LLM 基于历史上下文分析]
Checkpoint 8: 第二轮对话完成
├─ 触发时机: LLM 直接生成回复（无需调用工具）
├─ 状态内容:
│ - messages: [..., AIMessage(content="基于之前的活动推荐，我建议...")]
├─ next: null (END)
└─ parent: checkpoint_7
checkpoints表结构介绍
-- checkpoints 表
-- 每条记录是一个"状态快照"，不是执行图中的节点
CREATE TABLE IF NOT EXISTS checkpoints
(
thread_id TEXT NOT NULL COMMENT '会话线程ID，用于区分不同的对话上下文',
checkpoint_ns TEXT NOT NULL DEFAULT '' COMMENT '检查点命名空间，用于在复杂图结构中隔离不同子图的状态',
checkpoint_id TEXT NOT NULL COMMENT '检查点唯一标识符，通常为UUID',
parent_checkpoint_id TEXT COMMENT '父检查点ID，指向上一个检查点形成状态演进链，支持回溯和分支',
type TEXT COMMENT '检查点类型标识',
checkpoint JSONB NOT NULL COMMENT '检查点完整状态数据（JSONB格式），包含消息历史、中间状态等',
metadata JSONB NOT NULL DEFAULT '{}' COMMENT '检查点元数据（JSONB格式），存储时间戳、版本号、用户信息等附加信息',
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
) COMMENT='对话状态检查点主表，存储会话的完整状态快照';
checkpoints.checkpoint字段结构介绍
{
// LangGraph 的 schema 版本
"v": 4,
// checkpoint 的唯一标识符（UUID），对应表中的 checkpoint_id
"id": "1f12f75d-3ee6-6cad-800e-673d15950963",
// checkpoint 创建的时间戳（ISO 8601 格式，带时区）
"ts": "2026-04-03T15:57:33.849293+00:00",
// 节点数据版本（每个节点处理数据时，需要知道自己看到的是哪个版本的数据）
"versions_seen": {
"model": {
"branch:to:model": "00000000000000000000000000000015.0.32419480173508886"
},
"tools": {},
"__input__": {},
"__start__": {
"__start__": "00000000000000000000000000000014.0.8675745171161198"
}
},
// 通道数据，存储各个通道的实际数据值，数据在 checkpoint_blobs 表中
"channel_values": {},
// 通道版本号，
"channel_versions": {
// 消息通道版本（或者说容器被修改的次数）
"messages": "00000000000000000000000000000016.0.25846931507483395",
// 起始通道版本（或者说容器被修改的次数）
"__start__": "00000000000000000000000000000015.0.32419480173508886",
// LangGraph 内部任务队列版本（或者说容器被修改的次数）
"__pregel_tasks": "00000000000000000000000000000012.0.34774614538873905",
// 分支到模型版本（或者说容器被修改的次数）
"branch:to:model": "00000000000000000000000000000016.0.25846931507483395"
},
// 本次更新的通道
"updated_channels": [
"messages"
]
}
checkpoints.metadata字段结构介绍
{
// 执行步骤编号
"step": 8,
// 产生阶段
"source": "loop",
// 父节点信息
"parents": {},
// 创建方式标识
"ls_integration": "langchain_create_agent"
}
执行步骤分析
select * from checkpoint_writes where thread_id = 'user1' order by checkpoint_id asc, idx asc;
第一轮对话："上海天气如何，适合什么活动？"
记录1 → 用户输入
checkpoint_id: 1f12f7ca-e37f-6369-bfff-610650e70739
task_id: e0f2243c-3d0a-e98d-1b93-65c50d265057
channel: messages
内容: "上海天气如何，适合什么活动？"
记录2 → 路由到 model 节点
channel: branch:to:model
作用: 将消息路由给 AI 模型处理
记录3 → AI 决定调用 get_weather 工具
checkpoint_id: 1f12f7ca-e383-647e-8000-6b6432dd7fd4
channel: messages
内容: AI 消息 + tool_call (get_weather, city="上海")
记录4 → 任务调度
channel: __pregel_tasks
作用: LangGraph 内部任务调度，准备执行工具
记录5 → get_weather 工具返回结果
checkpoint_id: 1f12f7ca-fe17-65b4-8001-c0e1a430e44f
channel: messages
内容: "上海 的天气晴朗，气温 25°C。"
记录6 → 再次路由到 model
channel: branch:to:model
作用: 将工具结果返回给 AI 继续处理
记录7 → AI 决定调用 get_activity 工具
checkpoint_id: 1f12f7ca-fe1d-6b66-8002-4d40112fc355
channel: messages
内容: AI 消息 + tool_call (get_activity, city="上海")
记录8 → 任务调度
channel: __pregel_tasks
作用: 准备执行 get_activity 工具
记录9 → get_activity 工具返回结果
checkpoint_id: 1f12f7cb-1501-6331-8003-5fa698212b4e
channel: messages
内容: "上海 的活动有玫兰芳拉扯、樱芳拉扯"
记录10 → 路由到 model
channel: branch:to:model
记录11 → AI 生成最终回复
checkpoint_id: 1f12f7cb-1505-6bdc-8004-38ff86c9bedb
channel: messages
内容: 综合天气和活动信息的完整回答
第二轮对话："我的时间有限，只能参加一个活动，帮我推荐一个"
记录12 → 用户第二轮输入
checkpoint_id: 1f12f7cb-3b67-67cc-8006-d8daa405a148
channel: messages
内容: "我的时间有限，只能参加一个活动，帮我推荐一个"
记录13 → 路由到 model
channel: branch:to:model
记录14 → AI 基于上下文直接回复
checkpoint_id: 1f12f7cb-3b6a-6c08-8007-68be15499541
channel: messages
内容: 推荐樱芳拉扯（基于之前的天气和活动信息）
checkpoint_writes 表解构
-- checkpoint_writes 表
-- 每条记录是一个步骤
CREATE TABLE IF NOT EXISTS checkpoint_writes
(
thread_id TEXT NOT NULL COMMENT '关联的会话线程ID，外键关联到checkpoints表',
checkpoint_ns TEXT NOT NULL DEFAULT '' COMMENT '关联的命名空间，外键关联到checkpoints表',
checkpoint_id TEXT NOT NULL COMMENT '关联的检查点ID，外键关联到checkpoints表',
task_id TEXT NOT NULL COMMENT '写入任务唯一标识，用于标识一个具体的写入操作',
idx INTEGER NOT NULL COMMENT '同一任务中多个写入操作的顺序索引',
channel TEXT NOT NULL COMMENT '状态通道名称，如messages、context等，标识写入的目标状态键',
type TEXT COMMENT '写入操作类型标识',
value JSONB COMMENT '写入的具体数据（JSONB格式）',
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
) COMMENT='检查点写入操作记录表，追踪状态变更的细粒度操作';
checkpoint_writes.channel枚举
1. messages
用途: 存储对话消息流
内容: HumanMessage、AIMessage、ToolMessage 等
示例: 用户输入、AI回复、工具调用结果
2. branch:to:model
用途: 路由控制信号
内容: 通常为 null 或空
作用: 指示消息流应该路由到 model 节点继续处理
3. __pregel_tasks
用途: LangGraph 内部任务调度
内容: 序列化的任务对象（包含工具调用信息、状态等）
作用: 协调多个节点的执行顺序和数据传递
4. start
用途: 节点启动标记
内容: 通常为空对象 {}
作用: 标记某个节点开始执行
5. end
用途: 节点结束标记
内容: 通常为空对象 {}
作用: 标记某个节点执行完成或整个流程结束
6. state
用途: 自定义状态存储
内容: 用户定义的状态数据
示例: {"current_step": "processing", "retry_count": 0}
7. context
用途: 上下文信息存储
内容: 会话级别的上下文数据
示例: 用户偏好、会话配置等
8. metadata
用途: 元数据存储
内容: 时间戳、token 使用量、执行耗时等
示例: {"timestamp": "...", "tokens": 150}
9. tools
用途: 工具相关信息
内容: 工具定义、工具状态等
10. 自定义 channel
用途: 用户自定义的状态通道
命名: 任意字符串（如 "user_profile", "session_data"）
消息存储
checkpoint_blobs 表
-- checkpoint_blobs 表
--
CREATE TABLE IF NOT EXISTS checkpoint_blobs
(
thread_id TEXT NOT NULL COMMENT '关联的会话线程ID',
checkpoint_ns TEXT NOT NULL DEFAULT '' COMMENT '关联的命名空间',
channel TEXT NOT NULL COMMENT '数据通道名称，标识大对象的类型或用途',
version TEXT NOT NULL COMMENT '版本标识，通常对应checkpoint_id，用于关联到具体检查点',
type TEXT NOT NULL COMMENT '大对象数据类型标识，如image/png、application/pdf等',
blob BYTEA COMMENT '二进制大对象数据（BYTEA格式），存储文件、图片、文档等大型数据',
PRIMARY KEY (thread_id, checkpoint_ns, channel, version)
) COMMENT='大对象存储表，用于存储不适合JSONB格式的大型二进制数据';
checkpoint_blobs.type枚举
1. bytes: 原始字节对象
2. bytearray: 字节数组对象
3. msgpack: MessagePack 序列化（大多数对象的默认方式）
4. pickle: pickle 序列化（仅当启用 pickle_fallback 时）
5. empty: 标记未设置的 channel 值
6. null: None/空值
checkpoint_migrations 表
-- checkpoint_migrations 表
CREATE TABLE IF NOT EXISTS checkpoint_migrations
(
version INTEGER NOT NULL PRIMARY KEY COMMENT '迁移版本号，递增的整数标识',
applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '迁移应用时间戳，记录schema更新的时间'
) COMMENT='数据库迁移版本管理表，跟踪schema变更历史';
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
10
0- 0
已为社区贡献2条内容
所有评论(0)