← 返回文章列表
LangChain 全指南:从入门到精通
基于费曼学习法、西蒙学习法、SQ3R 阅读法和康奈尔笔记法,系统掌握 LangChain 生态开源 Agent 框架的核心概念与实战技巧。
一、概览与提问(SQ3R · Survey & Question)
SQ3R 第一步:快速浏览全貌,提出关键问题。
什么是 LangChain / LangGraph?
LangChain 是一个开源的 Agent 工程平台,提供了构建 AI Agent 所需的核心抽象:统一的模型接口(Model)、工具(Tool)、提示词(Prompt)和中间件(Middleware)。它通过 create_agent 提供了一个极简但高度可配置的 Agent 驱动器(harness),你可以从 model + tools + prompt + middleware 自由组合出满足需求的 Agent。
LangGraph 是 LangChain 生态中的底层编排框架和运行时,用于构建、管理和部署长时间运行的有状态 Agent。它将 Agent 工作流建模为图(Graph)——由节点(Node)、边(Edge)和状态(State)组成,提供了持久化执行、Human-in-the-Loop、流式输出等生产级能力。LangGraph 灵感来自 Google 的 Pregel 系统和 Apache Beam,被 Klarna、Uber、J.P. Morgan 等公司信任使用。
简单理解两者的关系:
| 产品 | 定位 | 适用场景 |
|---|---|---|
| Deep Agents | 高层 Agent 驱动器("开箱即用") | 快速构建带规划、子 Agent、文件系统的复杂 Agent |
| LangChain | Agent 框架(核心抽象层) | 自定义 Agent 驱动器,灵活组合 model/tools/middleware |
| LangGraph | 编排运行时(底层控制) | 需要精细控制执行流程、持久化、Human-in-the-Loop 的场景 |
| LangSmith | 可观测性平台 | 追踪、调试、评估、部署 Agent |
DeepSeek 在 AI 生态中的定位
DeepSeek 是一家中国 AI 公司,提供高性能、极低成本的大语言模型 API 服务。截至 2026 年,其核心模型包括:
- DeepSeek-V4-Pro / V4-Flash:旗舰通用模型,性能对标 GPT-5.5 和 Claude Opus 4.7,但成本仅为其 1/10
- DeepSeek-R1:推理模型,支持链式思维(Chain-of-Thought)的"思考模式",在复杂推理任务上表现卓越
- DeepSeek Coder:专注于编程的模型
DeepSeek API 的核心优势:
- OpenAI SDK 兼容:只需更改
base_url即可切换 - Anthropic 接口支持:同时兼容 Anthropic API 格式
- 极低定价:缓存命中时低至 $0.0028/百万 token
- 思考模式:内置 CoT 推理,可控制推理力度(
reasoning_effort)
核心问题
- LangChain 和 LangGraph 的关系是什么?——LangChain 提供高层 Agent 抽象,LangGraph 提供底层图编排运行时;LangChain 的 Agent 底层构建在 LangGraph 之上
- 什么时候用 LangChain,什么时候直接用 LangGraph?——简单 Agent 用 LangChain 的
create_agent;需要精细控制工作流、状态持久化、Human-in-the-Loop 时用 LangGraph - DeepSeek 如何集成到 LangChain 生态中?——通过 OpenAI 兼容接口,设置
base_url指向 DeepSeek API 即可 - StateGraph 的核心编程模型是什么?——定义 State → 添加 Node → 添加 Edge → Compile → Invoke
技术全景图
LangChain 生态
├── 核心基础
│ ├── Models(模型)——统一的 LLM 接口
│ ├── Messages(消息)——标准化的消息格式
│ ├── Tools(工具)——Agent 可调用的函数
│ ├── Agents(代理)——create_agent 驱动器
│ └── Middleware(中间件)——行为拦截与增强
├── LangGraph 编排
│ ├── StateGraph ——状态图定义
│ ├── Node(节点)——执行逻辑的函数
│ ├── Edge(边)——控制流转的连接
│ ├── Conditional Edges ——条件路由
│ ├── Persistence(持久化)——Checkpoint 管理
│ ├── Interrupts(中断)——Human-in-the-Loop
│ └── Streaming(流式输出)
├── DeepSeek 集成
│ ├── OpenAI 兼容接口
│ ├── 思考模式(Thinking Mode)
│ ├── 推理力度控制
│ └── 流式与非流式调用
└── 生产级能力
├── LangSmith 可观测性
├── 部署与扩展
├── 评估与测试
└── 多 Agent 协作
二、用最简单的话说清楚(费曼学习法)
费曼学习法:如果你不能用简单的话解释一个概念,说明你还没有真正理解它。
核心概念讲解
LLM(大语言模型)
LLM 就像一个读过万卷书的超级大脑。你给它一段文字(prompt),它预测接下来应该说什么。DeepSeek、GPT、Claude 都是 LLM。
from openai import OpenAI
# 使用 DeepSeek API(OpenAI 兼容格式)
client = OpenAI(
api_key="your-deepseek-api-key",
base_url="https://api.deepseek.com"
)
response = client.chat.completions.create(
model="deepseek-v4-flash",
messages=[
{"role": "system", "content": "你是一个有帮助的助手。"},
{"role": "user", "content": "什么是 Agent?"}
]
)
print(response.choices[0].message.content)Agent(智能体)
Agent = LLM + 工具 + 循环。想象一个能使用工具的助手:它不仅会回答问题,还会主动决定"我需要搜索一下"、"我需要计算一下",然后调用工具获取信息,最终给出答案。
from langchain.agents import create_agent
def search_web(query: str) -> str:
"""搜索网页获取信息。"""
return f"搜索结果:{query} 的相关信息..."
def calculate(expression: str) -> str:
"""计算数学表达式。"""
return str(eval(expression))
agent = create_agent(
model="openai:deepseek-v4-flash", # 通过 LangChain 标准 model 参数使用
tools=[search_web, calculate],
system_prompt="你是一个有用的助手,可以搜索和计算。"
)
result = agent.invoke({
"messages": [{"role": "user", "content": "帮我算一下 3^10 等于多少"}]
})Chain(链)
Chain 就是把多个步骤串联起来。比如:先提取用户意图 → 再调用搜索引擎 → 最后用 LLM 总结结果。在 LangChain v1 中,create_agent 已经取代了旧的 Chain 概念,成为构建 Agent 的主要方式。
Tool(工具)
Tool 就是 Agent 可以调用的函数。任何 Python 函数都可以变成 Tool,LangChain 会自动根据函数签名和 docstring 生成工具描述,让 LLM 知道何时该调用它。
def get_weather(city: str) -> str:
"""获取指定城市的天气信息。"""
# 实际场景中这里会调用天气 API
return f"{city}今天晴,温度 22°C"
# 直接传入 create_agent 即可
agent = create_agent(
model="openai:deepseek-v4-flash",
tools=[get_weather],
system_prompt="你是一个天气助手。"
)Memory(记忆)
Memory 让 Agent 记住之前的对话。就像人一样,如果每次对话都从零开始,就无法处理多轮对话中的上下文引用。LangGraph 通过 State 和 Checkpointer 实现了短期记忆和长期记忆。
State(状态)
State 是 LangGraph 的核心数据结构。它记录了图在执行过程中的所有信息——对话历史、中间结果、用户输入等。每个 Node 读取 State、执行逻辑、返回更新。
from typing import Annotated
from typing_extensions import TypedDict
from operator import add
class AgentState(TypedDict):
messages: Annotated[list, add] # 消息列表,新消息会被追加
current_step: str # 当前步骤Graph(图)
Graph 是 LangGraph 的核心编程模型。你把工作流画成一张"流程图"——每个节点是一个处理步骤,每条边决定下一步走向哪里。
from langgraph.graph import StateGraph, MessagesState, START, END
def chatbot(state: MessagesState):
"""处理用户消息的节点。"""
return {"messages": [{"role": "ai", "content": "你好!我是你的助手。"}]}
graph = StateGraph(MessagesState)
graph.add_node(chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
graph = graph.compile()
result = graph.invoke({"messages": [{"role": "user", "content": "你好!"}]})类比与比喻
| 概念 | 比喻 |
|---|---|
| LLM | 一个知识渊博但不会动手的顾问 |
| Agent | 一个既能思考又能动手的全能助手 |
| Tool | 助手手中的工具箱(计算器、搜索引擎、数据库...) |
| Chain | 工作流水线——第一步做什么、第二步做什么 |
| State | 助手的笔记本——记录所有中间信息和对话历史 |
| Graph | 整个工作流程图——什么条件下走哪条路 |
| Middleware | 工作流程中的"质检站"——拦截、检查、修改每一步 |
| LangSmith | 监控摄像头——记录 Agent 的一举一动 |
常见误解澄清
- 误解:LangChain 是一个 LLM。事实:LangChain 是一个编排框架,它不包含 LLM,而是提供统一接口调用各家 LLM。
- 误解:LangGraph 只能配合 LangChain 使用。事实:LangGraph 可以独立使用,不依赖 LangChain。
- 误解:DeepSeek 是 LangChain 的竞争对手。事实:DeepSeek 是 LLM Provider,与 LangChain 是互补关系——LangChain 提供框架,DeepSeek 提供模型。
- 误解:Agent 一定要很复杂。事实:LangChain 的
create_agent几行代码就能创建一个可用的 Agent。
三、锥形深入(西蒙学习法)
西蒙学习法:锥形突击,从基础到高级层层递进。
第一层:核心基础
LangChain 核心组件
Models(模型)
LangChain 提供了统一的模型接口,让你无缝切换不同 Provider 而无需修改业务代码。DeepSeek 通过 OpenAI 兼容接口集成:
# pip install -qU langchain "langchain[openai]"
from langchain.agents import create_agent
# 使用 DeepSeek 模型
agent = create_agent(
model="openai:deepseek-v4-flash",
system_prompt="你是一个有帮助的助手。"
)LangChain 标准化了不同 Provider 的 API 差异,支持 OpenAI、Anthropic、Google、DeepSeek 等。
Tools(工具)
工具是 Agent 与外部世界交互的桥梁。任何 Python 函数都可以成为 Tool:
def lookup_database(query: str) -> str:
"""查询数据库获取信息。
Args:
query: SQL 查询语句
"""
# 实际场景中连接真实数据库
return f"查询结果:共找到 3 条匹配记录"
def send_email(to: str, subject: str, body: str) -> str:
"""发送邮件。
Args:
to: 收件人邮箱
subject: 邮件主题
body: 邮件正文
"""
return f"邮件已发送至 {to}"函数的 docstring 和类型注解会被 LangChain 自动解析,告诉 LLM 这个工具做什么、需要什么参数。
Middleware(中间件)
中间件是 LangChain v1 引入的重要概念,用于在 Agent 执行过程中拦截和修改行为。每个中间件处理一个关注点,自由组合:
from langchain.agents import create_agent
agent = create_agent(
model="openai:deepseek-v4-flash",
tools=[get_weather],
system_prompt="你是一个助手。",
# 可以添加预构建的或自定义的 middleware
)LangGraph 基础:StateGraph
LangGraph 的核心编程模型是 StateGraph——用三个元素构建图:
- State:共享状态,定义图的"记忆"
- Node:节点函数,执行具体逻辑
- Edge:边,决定下一步去哪
from langgraph.graph import StateGraph, MessagesState, START, END
# 1. 定义节点函数
def greet(state: MessagesState):
"""问候节点。"""
return {"messages": [{"role": "ai", "content": "你好!有什么可以帮你?"}]}
# 2. 构建图
graph = StateGraph(MessagesState)
graph.add_node(greet)
graph.add_edge(START, "greet") # 从 START 到 greet 节点
graph.add_edge("greet", END) # 从 greet 到 END
# 3. 编译并执行
app = graph.compile()
result = app.invoke({"messages": [{"role": "user", "content": "你好"}]})DeepSeek API 基础调用
DeepSeek API 完全兼容 OpenAI 格式,切换只需修改 base_url:
from openai import OpenAI
client = OpenAI(
api_key="your-api-key",
base_url="https://api.deepseek.com"
)
# 基础对话
response = client.chat.completions.create(
model="deepseek-v4-flash",
messages=[
{"role": "system", "content": "你是一个有帮助的助手。"},
{"role": "user", "content": "解释一下量子计算"}
],
stream=False
)
print(response.choices[0].message.content)
# 流式输出
stream = client.chat.completions.create(
model="deepseek-v4-flash",
messages=[{"role": "user", "content": "写一首关于春天的诗"}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="")第二层:进阶用法
自定义 Agent 与 Tool 集成
from langchain.agents import create_agent
def search_knowledge_base(query: str) -> str:
"""在知识库中搜索相关文档。
Args:
query: 搜索关键词
"""
# 实际中连接向量数据库
return f"找到与 '{query}' 相关的文档:LangGraph 是一个..."
def create_summary(text: str, max_words: int = 100) -> str:
"""为给定文本生成摘要。
Args:
text: 原始文本
max_words: 摘要最大词数
"""
return f"摘要({max_words}词以内):{text[:50]}..."
agent = create_agent(
model="openai:deepseek-v4-flash",
tools=[search_knowledge_base, create_summary],
system_prompt="你是一个研究助手。先搜索知识库,再总结信息。"
)
result = agent.invoke({
"messages": [{"role": "user", "content": "帮我查一下 LangGraph 的核心特性并总结"}]
})RAG 管道(检索增强生成)
from langchain.agents import create_agent
def retrieve_documents(query: str) -> str:
"""从向量数据库中检索相关文档。
Args:
query: 用户查询
"""
# 实际中使用 LangChain 的 retriever
return f"检索到 3 篇相关文档:[文档1...], [文档2...], [文档3...]"
rag_agent = create_agent(
model="openai:deepseek-v4-flash",
tools=[retrieve_documents],
system_prompt="""你是一个 RAG 助手。
1. 先使用 retrieve_documents 工具检索相关文档
2. 基于检索到的文档内容回答用户问题
3. 如果文档中没有相关信息,如实告知用户"""
)LangGraph 条件边与循环
条件边让图具有分支决策能力,这是实现复杂 Agent 逻辑的关键:
from typing import Literal
from langgraph.graph import StateGraph, MessagesState, START, END
def agent_node(state: MessagesState):
"""Agent 思考节点。"""
return {"messages": [{"role": "ai", "content": "我需要使用工具"}]}
def tool_node(state: MessagesState):
"""工具执行节点。"""
return {"messages": [{"role": "tool", "content": "工具执行结果"}]}
def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
"""路由函数:决定是继续使用工具还是结束。"""
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "__end__"
# 构建带循环的 Agent 图
builder = StateGraph(MessagesState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent") # 工具执行后回到 agent —— 形成循环
app = builder.compile()这就是经典的 ReAct 模式:Agent 思考 → 调用工具 → 观察结果 → 继续思考 → ... 直到得出最终答案。
Human-in-the-Loop 中断
LangGraph 的 interrupt 机制让你可以在关键节点暂停执行,等待人类审批:
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.types import Command, interrupt
def human_review(state: MessagesState):
"""暂停执行,等待人类审批。"""
answer = interrupt("请审核 Agent 的操作,是否继续?")
return {"messages": [{"role": "user", "content": answer}]}
builder = StateGraph(MessagesState)
builder.add_node("review", human_review)
builder.add_edge(START, "review")
builder.add_edge("review", END)
checkpointer = InMemorySaver()
app = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "review-1"}}
# 第一次调用 —— 在 interrupt 处暂停
stream = app.invoke({"messages": []}, config)
# 人工审核后恢复执行
result = app.invoke(Command(resume="继续执行"), config)DeepSeek 推理模型(思考模式)
DeepSeek 的"思考模式"在输出最终答案前先输出链式思维推理过程,显著提升复杂任务的准确性:
from openai import OpenAI
client = OpenAI(
api_key="your-api-key",
base_url="https://api.deepseek.com"
)
# 启用思考模式
response = client.chat.completions.create(
model="deepseek-v4-pro",
messages=[
{"role": "user", "content": "证明:根号2是无理数"}
],
reasoning_effort="high", # 控制推理力度:high 或 max
extra_body={"thinking": {"type": "enabled"}},
)
# 访问推理过程和最终答案
reasoning = response.choices[0].message.reasoning_content # CoT 推理过程
answer = response.choices[0].message.content # 最终答案
print("=== 推理过程 ===")
print(reasoning)
print("=== 最终答案 ===")
print(answer)思考模式的关键点:
reasoning_content包含链式思维推理过程- 不支持
temperature、top_p等参数(设置不会报错但也无效) - 多轮对话中,无工具调用时的
reasoning_content可省略;有工具调用时必须回传
多 Agent 协作
LangGraph 支持通过子图(Subgraph)实现多 Agent 协作:
from langgraph.graph import StateGraph, MessagesState, START, END
def research_agent(state: MessagesState):
"""研究 Agent:负责信息收集。"""
return {"messages": [{"role": "ai", "content": "研究完成,以下是收集到的信息..."}]}
def writer_agent(state: MessagesState):
"""写作 Agent:负责内容生成。"""
return {"messages": [{"role": "ai", "content": "基于研究信息,生成文章..."}]}
def reviewer_agent(state: MessagesState):
"""审核 Agent:负责质量检查。"""
return {"messages": [{"role": "ai", "content": "审核通过。"}]}
# 构建协作流程
builder = StateGraph(MessagesState)
builder.add_node("researcher", research_agent)
builder.add_node("writer", writer_agent)
builder.add_node("reviewer", reviewer_agent)
builder.add_edge(START, "researcher")
builder.add_edge("researcher", "writer")
builder.add_edge("writer", "reviewer")
builder.add_edge("reviewer", END)
app = builder.compile()第三层:深度解析
LangGraph 的图执行引擎原理
LangGraph 的底层算法基于消息传递机制,灵感来自 Google 的 Pregel 系统。执行过程分为离散的"超级步(super-step)":
- 所有节点初始状态为
inactive - 节点收到新消息(状态更新)时变为
active - Active 节点执行函数,发送更新
- 没有新消息的节点投票
halt,变为inactive - 所有节点
inactive且无消息在途时,执行终止
并行节点属于同一个超级步,顺序节点属于不同超级步。这种模型既支持并行执行,又保证了确定性。
持久化状态管理
LangGraph 通过 Checkpointer 实现状态持久化。每次超级步结束时会保存检查点(checkpoint),这意味着:
- Agent 可以在故障后恢复,从中断的地方继续
- 支持时间旅行:回到任意历史状态,分支探索不同路径
- 长时间运行的任务可以在多次会话中延续
from langgraph.checkpoint.memory import InMemorySaver
# 使用内存 Checkpointer(开发环境)
checkpointer = InMemorySaver()
# 生产环境使用 PostgreSQL、MongoDB 等持久化存储
app = builder.compile(checkpointer=checkpointer)
# 每次 invoke 使用不同的 thread_id 隔离状态
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [...]}, config)Agent 架构设计模式
ReAct 模式(Reasoning + Acting)
最经典的 Agent 模式:思考 → 行动 → 观察 → 循环。LangGraph 的条件边 + 循环边天然支持这一模式。
Plan-and-Execute 模式
先制定完整计划,再逐步执行。适合复杂的多步骤任务:
from typing import Literal
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class PlanState(TypedDict):
goal: str
plan: list[str]
current_step: int
results: list[str]
def planner(state: PlanState):
"""制定执行计划。"""
plan = ["步骤1:收集信息", "步骤2:分析数据", "步骤3:生成报告"]
return {"plan": plan, "current_step": 0, "results": []}
def executor(state: PlanState):
"""执行当前步骤。"""
step = state["plan"][state["current_step"]]
return {
"results": state["results"] + [f"完成:{step}"],
"current_step": state["current_step"] + 1
}
def should_continue(state: PlanState) -> Literal["executor", "__end__"]:
"""是否还有步骤要执行?"""
if state["current_step"] < len(state["plan"]):
return "executor"
return "__end__"
builder = StateGraph(PlanState)
builder.add_node("planner", planner)
builder.add_node("executor", executor)
builder.add_edge(START, "planner")
builder.add_edge("planner", "executor")
builder.add_conditional_edges("executor", should_continue)
app = builder.compile()Multi-Agent 模式
多个专业 Agent 协作,通过路由器分配任务。LangGraph 通过子图(Subgraph)和 Command 实现跨图导航。
生产环境部署与可观测性
LangSmith 是 LangChain 生态的可观测性平台,提供:
- 追踪(Tracing):可视化 Agent 的每一步执行路径、状态转换和运行时指标
- 评估(Evaluation):在生产数据或离线数据集上测试和评分 Agent 行为
- 提示词工程:版本控制、协作优化、A/B 测试
- 一键部署:通过 LangSmith Deployment 部署 Agent
- LangSmith Engine:自动检测追踪中的问题并建议修复
# 启用 LangSmith 追踪(只需设置环境变量)
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-langsmith-key"
# 此后所有 Agent 执行都会自动记录到 LangSmith四、要点笔记(康奈尔笔记法)
康奈尔笔记法:左侧线索/关键词,右侧详细笔记,底部总结。
关键概念速查表
| 线索/关键词 | 详细笔记 |
|---|---|
create_agent | LangChain 核心 API,接收 model + tools + prompt + middleware,返回可调用的 Agent |
| StateGraph | LangGraph 的主图类,参数化一个用户定义的 State 对象 |
| State | 图的共享数据结构,通常用 TypedDict 定义,支持 Reducer 函数控制更新方式 |
| Node | 图中的节点函数,接收 state + config + runtime,返回 state 更新 |
| Edge | 连接节点的边:普通边(固定路径)、条件边(动态路由) |
| Reducer | 定义 State 更新策略。默认是覆盖;用 Annotated[list, add] 实现追加 |
| MessagesState | 预置的 State,包含 messages 字段和 add_messages Reducer |
| Checkpointer | 状态持久化组件,支持 InMemorySaver、PostgreSQL、MongoDB |
| interrupt | 暂停图执行,等待外部输入(Human-in-the-Loop) |
| Command | 多功能原语:update + goto + resume,统一状态更新和控制流 |
| Thinking Mode | DeepSeek 的思考模式,输出 CoT 推理过程(reasoning_content) |
| Middleware | LangChain v1 引入的行为拦截机制,每个中间件处理一个关注点 |
| LangSmith | 可观测性平台:追踪、评估、提示词工程、部署 |
核心 API 速查
| API / 类 | 用途 | 示例 |
|---|---|---|
create_agent(model, tools, system_prompt) | 创建 LangChain Agent | agent = create_agent(model="openai:gpt-5.4", tools=[fn]) |
StateGraph(state_schema) | 创建 LangGraph 状态图 | graph = StateGraph(MessagesState) |
graph.add_node(name, fn) | 添加节点 | graph.add_node("agent", agent_fn) |
graph.add_edge(from, to) | 添加普通边 | graph.add_edge(START, "agent") |
graph.add_conditional_edges(node, fn) | 添加条件边 | graph.add_conditional_edges("agent", route_fn) |
graph.compile(checkpointer=...) | 编译图 | app = graph.compile(checkpointer=InMemorySaver()) |
app.invoke(input, config) | 执行图 | result = app.invoke({"messages": [...]}, config) |
interrupt(question) | 暂停等待人类输入 | answer = interrupt("确认继续?") |
Command(update=..., goto=..., resume=...) | 状态更新+控制流 | Command(resume="yes") |
Send(node, state) | 动态发送状态到节点 | Send("worker", {"item": x}) |
| DeepSeek API(OpenAI 格式) | 调用 DeepSeek 模型 | client.chat.completions.create(model="deepseek-v4-flash", ...) |
| DeepSeek Thinking Mode | 启用推理模式 | extra_body={"thinking": {"type": "enabled"}} |
本节总结
LangChain 生态围绕 Agent 工程构建了完整的工具链:LangChain 提供 create_agent 高层抽象,LangGraph 提供 StateGraph 底层编排,DeepSeek 提供高性价比的 LLM 能力。三者可以独立使用,也可以无缝组合。核心编程模型是图:State 定义数据、Node 定义逻辑、Edge 定义流转,compile 后 invoke 执行。
五、复习与实践(SQ3R · Recite & Review)
SQ3R 最后两步:复述核心要点,通过实践巩固理解。
核心要点回顾
- LangChain 是 Agent 框架,
create_agent是核心 API,通过 model + tools + prompt + middleware 组合构建 Agent - LangGraph 是编排运行时,用 StateGraph(State + Node + Edge)建模复杂工作流,支持持久化、中断、流式输出
- DeepSeek 通过 OpenAI 兼容接口集成,只需更改
base_url,支持 Thinking Mode(CoT 推理) - LangSmith 提供可观测性:追踪、评估、提示词工程、一键部署
- LangChain 的 Agent 底层构建在 LangGraph 之上,两者不是竞争关系而是分层关系
动手练习
练习 1:创建一个简单的 DeepSeek Agent
from langchain.agents import create_agent
def get_time() -> str:
"""获取当前时间。"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
agent = create_agent(
model="openai:deepseek-v4-flash",
tools=[get_time],
system_prompt="你是一个时间助手。"
)
result = agent.invoke({
"messages": [{"role": "user", "content": "现在几点了?"}]
})练习 2:构建一个带条件路由的 LangGraph
from typing import Literal
from langgraph.graph import StateGraph, MessagesState, START, END
def classifier(state: MessagesState):
"""分类用户意图。"""
return {"messages": [{"role": "ai", "content": "分类结果:天气查询"}]}
def weather_handler(state: MessagesState):
"""处理天气查询。"""
return {"messages": [{"role": "ai", "content": "今天晴天,25°C"}]}
def general_handler(state: MessagesState):
"""处理一般问题。"""
return {"messages": [{"role": "ai", "content": "我是你的助手,有什么可以帮你?"}]}
def route(state: MessagesState) -> Literal["weather", "general"]:
last = state["messages"][-1].content if hasattr(state["messages"][-1], "content") else ""
if "天气" in str(last):
return "weather"
return "general"
builder = StateGraph(MessagesState)
builder.add_node("classifier", classifier)
builder.add_node("weather", weather_handler)
builder.add_node("general", general_handler)
builder.add_edge(START, "classifier")
builder.add_conditional_edges("classifier", route, {
"weather": "weather",
"general": "general"
})
builder.add_edge("weather", END)
builder.add_edge("general", END)
app = builder.compile()练习 3:实现 DeepSeek 思考模式 + 工具调用
from openai import OpenAI
import json
client = OpenAI(
api_key="your-api-key",
base_url="https://api.deepseek.com"
)
tools = [
{
"type": "function",
"function": {
"name": "get_population",
"description": "获取城市人口数据",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
}
}
]
def get_population(city: str) -> str:
populations = {"北京": "2189万", "上海": "2487万", "深圳": "1768万"}
return populations.get(city, "数据不可用")
# 带思考模式的工具调用
response = client.chat.completions.create(
model="deepseek-v4-pro",
messages=[{"role": "user", "content": "北京和上海哪个城市人口更多?"}],
tools=tools,
reasoning_effort="high",
extra_body={"thinking": {"type": "enabled"}},
)
# 处理工具调用
if response.choices[0].message.tool_calls:
for tool_call in response.choices[0].message.tool_calls:
args = json.loads(tool_call.function.arguments)
result = get_population(**args)
print(f"工具调用 {tool_call.function.name}({args}) = {result}")常见陷阱
- 忘记 compile:StateGraph 必须调用
.compile()才能执行,直接 invoke 会报错 - State 更新策略搞错:默认 Reducer 是覆盖,如果需要追加列表要用
Annotated[list, add] - DeepSeek 思考模式忽略 reasoning_content 回传:有工具调用时必须将
reasoning_content回传 API,否则会返回 400 错误 - 递归限制溢出:LangGraph 默认递归限制为 1000 步,复杂 Agent 可能超出。使用
RemainingSteps提前检测 - 混淆 LangChain 和 LangGraph 的职责:LangChain 管"Agent 是什么",LangGraph 管"Agent 怎么跑"
- Node 中产生不可逆副作用:LangGraph 在 checkpoint 后恢复时会重跑节点。确保节点逻辑幂等
延伸阅读
- LangChain 官方文档 — LangChain Agent 框架完整指南
- LangGraph 官方文档 — LangGraph 编排运行时详解
- LangGraph Graph API 概览 — StateGraph 核心概念
- DeepSeek API 文档 — DeepSeek 模型使用与定价
- DeepSeek 思考模式指南 — Thinking Mode 详细说明
- LangSmith 可观测性 — 追踪、调试 Agent
- LangChain Academy — 官方学习课程