LangChain从入门到精通（2）

上篇回顾：LangChain从入门到精通（上）：基础实战篇

目录

1. LangGraph工作流编排

在上篇中，我们通过 create_agent 快速构建了具备检索和工具调用能力的Agent。你可能已经注意到，Agent内部的执行路径并非线性的——它可能多次调用工具、反复调用模型，甚至陷入循环。这就是为什么从LangChain 1.x开始，Agent底层已全面迁移至LangGraph。

LangGraph的核心价值：让开发者精确控制Agent的每一步执行，在保持自主性的同时大幅提升稳定性。

1.1 LangGraph三要素

LangGraph基于图（Graph）来编排AI工作流，其核心抽象仅三个：

基础示例——简单的笑话生成器：

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class JokeState(TypedDict):
topic: str
joke: str

def generate_joke(state: JokeState):
msg = llm.invoke(f"写一个关于 {state['topic']} 的笑话")
return {"joke": msg.content}

graph = (
StateGraph(JokeState)
.add_node("generate", generate_joke)
.add_edge(START, "generate")
.add_edge("generate", END)
.compile()
)

1.2 七种核心编排模式

LangGraph将常见的AI工作流归纳为七种模式，覆盖了从简单到复杂的全部场景：

模式一：顺序执行（Prompt Chaining）

前一个LLM的输出作为后一个LLM的输入，类似流水线。可以在节点间加入验证门控——质量不达标时退回重做。

典型场景：写作流程（大纲→初稿→润色）、翻译+校对

模式二：并行执行（Parallelization）

多个节点同时执行，有两种实现方式：

• 任务拆分：将一个大任务拆为多个子任务并行处理，最后汇总
• 冗余投票：多个LLM运行相同任务，合并优化结果（提高准确率）

# 三个节点从START并行出发
builder.add_edge(START, "joke")
builder.add_edge(START, "story")
builder.add_edge(START, "poem")
# 汇聚到聚合节点
builder.add_edge("joke", "aggregator")
builder.add_edge("story", "aggregator")
builder.add_edge("poem", "aggregator")

模式三：路由（Routing）

根据输入内容将请求分发到不同的处理节点，是实现Agent决策的核心机制。

典型场景：智能客服（售前/退款/退货路由）、按领域分类的知识问答

模式四：编排器-工作者（Orchestrator-Worker）

编排器（Orchestrator）动态将复杂任务拆解为多个子任务，通过 Send API 分发给工作节点（Worker）并行执行，最后合成器汇总结果。

与普通并行的本质区别：子任务数量是动态的、不确定的。

from langgraph.types import Send

def orchestrator(state: State):
"""为每个章节创建一个Worker"""
return [Send("worker", {"section": section}) for section in state["sections"]]

典型场景：深度研究报告生成、批量文档处理

模式五：评估-优化循环（Evaluator-Optimizer）

生成器产出内容，评估器打分给出反馈，不达标则带着反馈重返生成器。形成"生成-评估-反馈-优化"的闭环。

与普通循环的区别：自带终止条件，仅在质量不达标时重复。

典型场景：文案优化、代码审查、翻译质量提升

模式六：子图嵌套（Subgraphs）

将复杂图拆分为多层子图，降低单个图的复杂度：

parent_graph.add_node("analysis_subgraph", compiled_subgraph)

典型场景：大规模工作流的模块化拆分、团队协作开发

模式选择指南

1.3 Think In LangGraph：客户支持Agent完整案例

上篇我们学会了用 create_agent 快速构建Agent，但面对分支路径、人工审核、多步操作等复杂场景时，我们需要LangGraph的精细控制。本节从零构建一个自动处理客服邮件的Agent，展示完整的LangGraph思维框架。

需求分析

假设我们需要处理的邮件场景：

核心观察：不同的问题走不同的处理路径——这正是需要**图（Graph）**而非链（Chain）的原因。

第一阶段：设计工作流

首先梳理业务流程，确定节点：

第二阶段：设计State

State是所有节点共享的工作笔记本。判断字段是否入State的原则：如果下游节点需要用它做决策或处理，就入State；如果能从其他字段推导出来，就不存。

from typing import TypedDict, Literal

class EmailClassification(TypedDict):
intent: Literal["question", "bug", "billing", "feature", "complex"]
urgency: Literal["low", "medium", "high", "critical"]
topic: str
summary: str

class EmailAgentState(TypedDict):
email_content: str
sender_email: str
email_id: str
classification: EmailClassification | None
search_results: list[str] | None
customer_history: dict | None
draft_response: str | None

第三阶段：构建Node

在LangGraph中，Node内部通过 Command 声明路由，避免了复杂的外部条件边：

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, RetryPolicy

# 意图识别节点——内部决策路由
def classify_intent(state: EmailAgentState) –> Command[Literal["search_documentation", "bug_tracking", "draft_response"]]:
classification = classifier_llm.invoke(f"""
分析这封客户邮件并分类：
邮件: {state['email_content']}
发件人: {state['sender_email']}
提供 intent（question/bug/billing/feature/complex）、urgency、topic、summary。
""")

if classification['intent'] in ['question', 'feature']:
goto = "search_documentation"
elif classification['intent'] == 'bug':
goto = "bug_tracking"
else: # billing / complex → 生成草稿（后续判断是否人工审核）
goto = "draft_response"

return Command(update={"classification": classification}, goto=goto)

# 生成草稿节点——判断是否需要人工审核
def draft_response(state: EmailAgentState) –> Command[Literal["human_review", "send_reply"]]:
# … 调用LLM生成草稿 …
needs_review = (
classification.get('urgency') in ['high', 'critical'] or
classification.get('intent') in ['complex', 'billing']
)
goto = "human_review" if needs_review else "send_reply"
return Command(update={"draft_response": response.content}, goto=goto)

异常处理策略——不同节点类型有不同的容错方案：

第四阶段：构建图

由于节点内部已通过 Command 声明路由，图的边只需定义固定路径：

workflow = StateGraph(EmailAgentState)
workflow.add_node("read_email", read_email)
workflow.add_node("classify_intent", classify_intent)
workflow.add_node("search_documentation", search_documentation,
retry=RetryPolicy(max_attempts=3))
workflow.add_node("bug_tracking", bug_tracking)
workflow.add_node("draft_response", draft_response)
workflow.add_node("human_review", human_review)
workflow.add_node("send_reply", send_reply)

# 最小边原则——只定义固定路径
workflow.add_edge(START, "read_email")
workflow.add_edge("read_email", "classify_intent")
workflow.add_edge("human_review", "send_reply")
workflow.add_edge("send_reply", END)

app = workflow.compile(checkpointer=InMemorySaver())

设计哲学小结：LangGraph的"最小边原则"——尽可能在Node内部用Command做路由决策，Graph层面只保留固定的串行边。这使图的拓扑结构极度简洁，将复杂的路由逻辑留给LLM去处理。

2. 多智能体协作（Multi-Agent）

在上篇中，我们构建的单Agent配合适量工具和精心设计的提示词，足以解决大部分问题。但当工具数量膨胀到10+、涉及多个领域时，单Agent的上下文窗口将不堪重负。此时就需要引入多Agent协作。

2.1 何时使用多Agent

当出现以下信号时，可考虑引入多Agent：

2.2 常见协作模式

这些模式并非互斥，实际开发中可灵活组合。

2.3 实战案例：婚礼策划多Agent

以婚礼策划为例，展示Subagents模式的具体实现。

需求梳理

婚礼策划Agent需完成三项核心任务：机票搜索（Travel）、场地搜索（Venue）、歌单策划（Playlist）。

选择多Agent而非单Agent的决策依据：

• 工具过多：航班+场地+音乐数据库，上下文难以承受
• 可并行：三个任务彼此独立，可同时执行提高效率
• 便于扩展：未来新增"餐饮""摄影"等Agent不影响现有逻辑

实现方案

Step 1：定义三个子Agent

# Travel Agent（调用MCP工具查询航班）
travel_agent = create_agent(
model="deepseek-chat",
tools=[flights_tool, time_tool],
system_prompt="你是旅行规划专家…"
)

# Venue Agent（基于Tavily搜索场地）
venue_agent = create_agent(
model="deepseek-chat",
tools=[web_search],
system_prompt="你是婚礼场地专家…"
)

# Playlist Agent（查询SQLite数据库）
playlist_agent = create_agent(
model="deepseek-chat",
tools=[query_playlist_db],
system_prompt="你是音乐策划专家…"
)

Step 2：定义主Agent（Coordinator）

主Agent通过Tool调用子Agent，实现协调和汇总：

from langchain.agents import AgentState

class WeddingState(AgentState):
origin: str
destination: str
guest_count: str
genre: str

# 将子Agent封装为Tool
@tool
async def search_flights(runtime: ToolRuntime) –> str:
"""搜索前往目的地的航班"""
origin = runtime.state["origin"]
destination = runtime.state["destination"]
response = await travel_agent.ainvoke({
"messages": [HumanMessage(content=f"从{origin}到{destination}的航班")]
})
return response['messages'][–1].content

@tool
def search_venues(runtime: ToolRuntime) –> str:
"""搜索婚礼场地"""
query = f"在{ runtime.state['destination'] }适合{ runtime.state['guest_count'] }人的婚礼场地"
response = venue_agent.invoke({"messages": [HumanMessage(content=query)]})
return response['messages'][–1].content

# 主Agent——婚礼策划协调者
coordinator = create_agent(
model="deepseek-chat",
tools=[search_flights, search_venues, suggest_playlist, update_state],
state_schema=WeddingState,
system_prompt="你是婚礼策划协调员…"
)

Step 3：执行验证

运行后，主Agent先通过对话收集婚礼需求，更新State，然后并行调用三个子Agent：

用户：我来自伦敦，想在巴黎举办100人的婚礼，爵士风格

1. update_state → origin=伦敦, destination=巴黎, guest_count=100, genre=爵士
2. 并行调用：
→ search_flights（查询伦敦→巴黎航班）
→ search_venues（搜索巴黎婚礼场地）
→ suggest_playlist（策划爵士歌单）
3. 汇总输出完整的婚礼策划方案

三个子Agent互不依赖，可并行执行，这是Subagents模式相较于串行处理的核心效率优势。

3. MCP模型上下文协议

多Agent协同需要接入多种外部服务——航班查询、时间获取、数据库操作等。在传统模式下，每接入一个服务都需要手动编写Tool代码，不仅工作量大，而且难以复用。MCP（Model Context Protocol）正是为解决这一问题而生。

3.1 什么是MCP

MCP是由Anthropic推出的开放标准，可理解为 “AI世界的USB接口协议”。

在没有MCP的时代，每次接入外部服务都需要手动定义Tool，存在两个根本问题：

• 复用性差：不同Agent有相同的工具需求，每次都重复定义
• 协议不统一：千家服务接口各异，集成成本极高

MCP的解决思路是标准化：

• 外部服务遵循MCP协议提供接口（MCP Server）
• AI应用（MCP Host）通过MCP Client统一对接，无需手动定义Tool

核心概念：

通信方式：

3.2 连接外部MCP服务

LangChain通过 MultiServerMCPClient 统一管理多个MCP连接：

Time MCP（stdio方式——本地进程通信）：

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({
"time": {
"transport": "stdio",
"command": "uvx",
"args": ["mcp-server-time", "–local-timezone=Asia/Shanghai"]
}
})
tools = await client.get_tools()
# tools 包含: get_current_time, convert_time

agent = create_agent("deepseek-chat", tools)
response = await agent.ainvoke({
"messages": [HumanMessage("现在几点了？")]
})

Kiwi MCP（SSE方式——远程HTTP通信）：

client = MultiServerMCPClient({
"kiwi-com-flight-search": {
"transport": "http",
"url": "https://mcp.kiwi.com"
}
})
tools = await client.get_tools()

agent = create_agent(
model="deepseek-chat",
tools=tools,
system_prompt="You are a travel agent. If the user uses Chinese, use zh-cn as locale."
)

说明：Kiwi擅长国际航班，国内航班可选用 VariFlight（飞友AI） 等MCP服务。

3.3 自定义MCP服务

团队内部也可以将自己的服务封装为MCP Server。使用FastMCP框架可以快速创建：

from fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: float, b: float) –> float:
"""Add two numbers"""
return a + b

@mcp.tool()
def multiply(a: float, b: float) –> float:
"""Multiply two numbers"""
return a * b

if __name__ == "__main__":
mcp.run(transport="stdio")

LangChain侧连接自定义MCP：

client = MultiServerMCPClient({
"math": {
"transport": "stdio",
"command": "python",
"args": ["math_mcp_server.py"]
}
})
tools = await client.get_tools()

agent = create_agent("deepseek-chat", tools,
system_prompt="使用数学工具回答问题。")
response = await agent.ainvoke({
"messages": [HumanMessage("467和529的平方根之和是多少?")]
})

MCP使用流程小结：

配置MultiServerMCPClient → 调用get_tools()获取工具 → 创建Agent并调用

自定义MCP Server可提供三类资源：

4. 实战案例：AI私厨

现在，让我们通过一个完整的实战案例，将上、下篇的所有知识串联起来。

4.1 项目概述

AI私厨是一个基于LangChain多模态AI的食谱推荐应用：

• 用户上传冰箱/食材照片
• AI自动识别图片中的食材
• 根据食材搜索相关食谱
• 多维度评估排序（营养、难度、口味）
• 输出结构化的食谱推荐报告

4.2 技术选型

4.3 核心代码

from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch
from langchain.agents import create_agent
from langgraph.checkpoint.sqlite import SqliteSaver

# 1. 初始化搜索工具
web_search = TavilySearch(max_results=5, topic="general")

# 2. 初始化多模态模型
model = init_chat_model(
model="qwen3.5-plus",
model_provider="openai",
base_url=os.getenv("DASHSCOPE_BASE_URL"),
api_key=os.getenv("DASHSCOPE_API_KEY")
)

# 3. 初始化记忆
checkpointer = SqliteSaver(
sqlite3.connect("db/personal_chief.db", check_same_thread=False)
)
checkpointer.setup()

# 4. 系统提示词
system_prompt = """你是一名私人厨师。收到食材照片或清单后：
1. 识别和评估食材
2. 调用web_search搜索对应食谱
3. 从营养和难度两个维度打分排序
4. 输出结构化的食谱推荐报告（含步骤、配料、预估时间）"""

# 5. 创建Agent
agent = create_agent(
model=model,
tools=[web_search],
checkpointer=checkpointer,
system_prompt=system_prompt
)

4.4 多模态消息处理

from langchain.messages import HumanMessage

# 推荐使用OSS URL而非base64，避免请求体过大
message = HumanMessage(content=[
{"type": "image", "url": oss_image_url},
{"type": "text", "text": "帮我看看这些食材能做些什么？"}
])

response = agent.invoke(
{"messages": [message]},
config={"configurable": {"thread_id": "chef-001"}}
)

4.5 部署方案

方案一——LangGraph CLI部署：

创建 langgraph.json 配置文件，执行 langgraph dev 一键启动，自带RESTful API和可视化调试界面。

方案二——自定义FastAPI服务：

提供更灵活的控制，包含对话接口、会话管理接口和文件上传接口，适合与现有系统集成。

5. 总结

5.1 技术栈全景图

LangChain生态全景

├── LangChain Core（核心框架）
│ ├── 模型调用（ChatModels + with_structured_output）
│ ├── 工具系统（Tools + MCP）
│ ├── 消息系统（Messages + 多模态）
│ ├── 记忆管理（Checkpointer）
│ └── 中间件（Middleware + HITL）
│
├── RAG（检索增强生成）
│ ├── 知识库构建（加载 → 切分 → 向量化 → 存储）
│ ├── 检索优化（HyDE / Multi-Hop / 多路召回 + RRF）
│ ├── 重排序（Cross-Encoder / MMR）
│ └── 评估体系（RAGAS：Faithfulness / Precision / Recall）
│
├── LangGraph（工作流编排）
│ ├── 三要素（State / Node / Edge）
│ ├── 七种编排模式（Chaining / Parallel / Routing 等）
│ └── 最小边原则（Command内部路由）
│
├── Multi-Agent（多智能体协作）
│ ├── Subagents / Handoffs / Skills / Router
│ └── 实战：婚礼策划协调者模式
│
└── 工具链
├── LangSmith（调试 / 监控 / 部署）
├── MCP协议（Time / Kiwi / 自定义Math Server）
├── FastAPI（后端服务）
└── LangGraph CLI（一键部署）

5.2 实践

LangChain生态仍在快速演进，以下方向值得持续关注：

• Agentic RAG的深化：从"检索→生成"的固定模式，演进为Agent自主决定检索时机、检索策略和结果利用方式，使RAG系统真正具备"思考能力"。

• MCP协议的生态化：随着越来越多服务商提供MCP接口，"AI应用连接一切"的开发成本将大幅降低，标准化的Tool市场正在形成。

• Multi-Agent编排的成熟：从手动编写协调逻辑，演进为框架级的多Agent编排方案，支持Agent间更灵活的消息传递和任务协调。

• 生产级可观测性：LangSmith等平台将持续深化，从Trace级别演进为融合Metrics、Logging、Alerting的完整可观测性体系。

上篇回顾：LangChain从入门到精通（上）：基础实战篇