一、概览与提问(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
LangChainAgent 框架(核心抽象层)自定义 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 的一举一动

常见误解澄清

  1. 误解:LangChain 是一个 LLM。事实:LangChain 是一个编排框架,它不包含 LLM,而是提供统一接口调用各家 LLM。
  2. 误解:LangGraph 只能配合 LangChain 使用。事实:LangGraph 可以独立使用,不依赖 LangChain。
  3. 误解:DeepSeek 是 LangChain 的竞争对手。事实:DeepSeek 是 LLM Provider,与 LangChain 是互补关系——LangChain 提供框架,DeepSeek 提供模型。
  4. 误解: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——用三个元素构建图:

  1. State:共享状态,定义图的"记忆"
  2. Node:节点函数,执行具体逻辑
  3. 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 包含链式思维推理过程
  • 不支持 temperaturetop_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)":

  1. 所有节点初始状态为 inactive
  2. 节点收到新消息(状态更新)时变为 active
  3. Active 节点执行函数,发送更新
  4. 没有新消息的节点投票 halt,变为 inactive
  5. 所有节点 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_agentLangChain 核心 API,接收 model + tools + prompt + middleware,返回可调用的 Agent
StateGraphLangGraph 的主图类,参数化一个用户定义的 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 ModeDeepSeek 的思考模式,输出 CoT 推理过程(reasoning_content
MiddlewareLangChain v1 引入的行为拦截机制,每个中间件处理一个关注点
LangSmith可观测性平台:追踪、评估、提示词工程、部署

核心 API 速查

API / 类用途示例
create_agent(model, tools, system_prompt)创建 LangChain Agentagent = 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 最后两步:复述核心要点,通过实践巩固理解。

核心要点回顾

  1. LangChain 是 Agent 框架create_agent 是核心 API,通过 model + tools + prompt + middleware 组合构建 Agent
  2. LangGraph 是编排运行时,用 StateGraph(State + Node + Edge)建模复杂工作流,支持持久化、中断、流式输出
  3. DeepSeek 通过 OpenAI 兼容接口集成,只需更改 base_url,支持 Thinking Mode(CoT 推理)
  4. LangSmith 提供可观测性:追踪、评估、提示词工程、一键部署
  5. 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}")

常见陷阱

  1. 忘记 compile:StateGraph 必须调用 .compile() 才能执行,直接 invoke 会报错
  2. State 更新策略搞错:默认 Reducer 是覆盖,如果需要追加列表要用 Annotated[list, add]
  3. DeepSeek 思考模式忽略 reasoning_content 回传:有工具调用时必须将 reasoning_content 回传 API,否则会返回 400 错误
  4. 递归限制溢出:LangGraph 默认递归限制为 1000 步,复杂 Agent 可能超出。使用 RemainingSteps 提前检测
  5. 混淆 LangChain 和 LangGraph 的职责:LangChain 管"Agent 是什么",LangGraph 管"Agent 怎么跑"
  6. Node 中产生不可逆副作用:LangGraph 在 checkpoint 后恢复时会重跑节点。确保节点逻辑幂等

延伸阅读