摘要：还在用 AgentExecutor？还在写臃肿的 Python 链式调用？醒醒吧，LangChain 的天已经变了！本文基于 LangChain 最新架构（v0.3+），深入剖析 LangGraph、LCEL、Pydantic v2 三大核心变革。不论你是刚入门的 AI 小白，还是被旧版本折磨的资深开发者，这一篇足以让你从“Demo 侠”进化为“架构师”。含泪整理，建议先收藏再阅读！

关键词：LangChain v0.3, LangGraph, LCEL, RAG, Agent, LLM Ops, 企业级开发

目录 (Table of Contents)

  1. 前言：告别草台班子，拥抱工业标准

  2. 版本大迁徙：从 0.1 到 0.3，LangChain 到底经历了什么？

  3. 核心革命一：LCEL (LangChain Expression Language) —— 优雅的声明式编程

  4. 核心革命二：LangGraph —— Agent 的终局是图（Graph），不是链（Chain）

  5. 架构深潜：Pydantic v2 与 langchain-core 的解耦哲学

  6. 实战演练：构建一个带有记忆与搜索能力的 ReAct Agent（基于 LangGraph）

  7. RAG 2.0：结合 GraphRAG 与多模态检索

  8. LangSmith：在大模型黑盒中点亮一盏灯

1. 前言：告别草台班子，拥抱工业标准 

兄弟们，如果你还在翻看半年前的 LangChain 教程，你会发现代码全是红色的 Warning。为什么？因为 LLM 应用开发正在经历从“手工作坊”到“工业流水线”的剧变。

LangChain 曾被诟病“过度封装”、“调试困难”。官方听到了！从 v0.1 的稳定API，到 v0.2 的彻底解耦，再到如今 v0.3 全面拥抱 Pydantic v2 和 LangGraph，LangChain 已经不再是一个简单的工具箱，而是一套完整的 LLM OS（大模型操作系统） 级别的开发框架。

今天，我们就把旧时代的 LLMChain、ConversationalRetrievalChain 全部扔进垃圾桶，用最新的理念重塑你的 AI 世界观！

2. 版本大迁徙：从 0.1 到 0.3，LangChain 到底经历了什么？

在开始写代码前，你必须理解架构的变化，否则你永远在解决 ImportError。

2.1 模块化解耦（The Great Decoupling）

以前我们 pip install langchain 就能搞定一切，现在不行了。为了减小包体积和依赖冲突，架构被拆分成了：

  • langchain-core：核心中的核心。定义了 Runnable 协议、消息类型（HumanMessage/AIMessage）、LCEL 基础。这是你需要最先安装的。

  • langchain-community：社区贡献的集成。比如 Chroma、FAISS、Wikipedia 工具等。

  • langchain：构建认知架构的高级库（Chains, Agents, Retrieval strategies）。

  • langchain-experimental：实验性功能。

  • Partners Packages：如 langchain-openai, langchain-anthropic。现在的最佳实践是单独安装特定厂商的包，而不是混在一起。

2.2 为什么 v0.3 是里程碑？

v0.3 最大的改变是内部全面迁移至 Pydantic v2。这听起来像个底层细节，但实际上它带来了：

  1. 性能起飞：序列化和验证速度提升数倍。

  2. 流式输出（Streaming）：更稳定的 Token 级流式传输。

  3. 类型安全：IDE 的自动补全终于能用了！

3. 核心革命一：LCEL —— 优雅的声明式编程 

如果你还在写 Python 的 class 继承来定制 Chain，那你就 Out 了。LCEL (LangChain Expression Language) 是 LangChain 的灵魂。

它利用了 Python 的管道操作符 |，让代码像 Unix 管道一样清晰。

3.1 传统写法 vs LCEL 写法

 传统写法 (Deprecated style):

from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

prompt = PromptTemplate.from_template("讲一个关于{topic}的笑话")
chain = LLMChain(llm=OpenAI(), prompt=prompt)
result = chain.run("程序员")

 LCEL 写法 (Modern style):

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 1. 定义组件
model = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_template("讲一个关于 {topic} 的笑话")
output_parser = StrOutputParser()

# 2. 组装流水线
chain = prompt | model | output_parser

# 3. 调用 (支持 invoke, ainvoke, stream, batch)
result = chain.invoke({"topic": "程序员"})
print(result)

3.2 LCEL 的黑魔法：Runnable协议

为什么 LCEL 强？因为链中的每个环节都实现了 Runnable 协议。这意味着你写好的 Chain 自动拥有了以下能力，无需额外代码：

  • .stream(): 实时打字机效果。

  • .batch(): 并发处理请求，极大提升吞吐量。

  • .astream_log(): 获取中间步骤的日志（对调试 RAG 非常有用）。

4. 核心革命二：LangGraph —— Agent 的终局是图 

这是本文最重要的部分。

在 v0.2 之前，我们使用 AgentExecutor。它是一个黑盒，你不知道 LLM 在里面循环了多少次，也无法精准控制它的“思考路径”。

LangGraph 的出现，标志着 LangChain 进入了多智能体（Multi-Agent）时代。LangGraph 将 Agent 的执行流程建模为 图（Graph），包含 Nodes（节点） 和 Edges（边）。

4.1 为什么抛弃 AgentExecutor？

  1. 循环依赖：真实的业务往往需要循环（Loop），比如：生成 -> 检查 -> 报错 -> 重新生成。

  2. 状态管理：LangGraph 引入了 State（状态机）的概念，让上下文管理变得显式且可控。

  3. 人机协同（Human-in-the-loop）：你可以在图的执行过程中暂停，等待人类批准后再继续。

4.2 LangGraph 核心概念图解

  • State: 一个 TypedDict，存储对话历史、工具输出等。

  • Nodes: 执行具体任务的函数（如调用 LLM，执行 Python 代码）。

  • Edges: 逻辑跳转（如：如果 LLM 决定调用工具，跳转到工具节点；否则结束）。

5. 实战演练：构建一个企业级 ReAct Agent 

Talk is cheap, show me the code. 我们用 LangGraph 写一个具备“联网搜索”能力的 Agent。

5.1 环境准备

pip install langchain langchain-openai langgraph langchain-community duckduckgo-search

5.2 定义状态与工具

import operator
from typing import Annotated, List, TypedDict, Union

from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END

# 1. 定义状态 (State)
# `add_messages` reducer 确保新消息是被追加而不是覆盖
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]

# 2. 准备工具
tool = DuckDuckGoSearchRun()
tools = [tool]

# 3. 绑定工具到 LLM
model = ChatOpenAI(model="gpt-4o", temperature=0)
model_with_tools = model.bind_tools(tools)

5.3 构建图节点 (Nodes)

我们需要两个节点：

  1. Agent 节点：负责思考和生成回复。

  2. Tool 节点：负责执行工具调用。

from langgraph.prebuilt import ToolNode

# 节点1: 调用模型
def call_model(state: AgentState):
    messages = state['messages']
    response = model_with_tools.invoke(messages)
    return {"messages": [response]}

# 节点2: 工具执行节点 (LangGraph 提供了预构建的 ToolNode)
tool_node = ToolNode(tools)

5.4 定义边 (Edges) 与 条件逻辑

这是最关键的一步。我们需要决定何时调用工具，何时结束。

from typing import Literal

# 条件边逻辑
def should_continue(state: AgentState) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]
    
    # 如果 LLM 返回了 tool_calls，则跳转到工具节点
    if last_message.tool_calls:
        return "tools"
    # 否则结束
    return END

5.5 组装图并编译

workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# 设置入口
workflow.set_entry_point("agent")

# 添加边
workflow.add_conditional_edges(
    "agent",
    should_continue,
)
workflow.add_edge("tools", "agent") # 工具执行完，必须回传给 Agent 再次思考

# 编译应用
app = workflow.compile()

5.6 运行起来！

inputs = {"messages": [HumanMessage(content="目前 LangChain 最新的版本是多少？LangGraph 有什么新特性？")]}

# stream_mode="values" 可以看到每一步的状态变化
for event in app.stream(inputs, stream_mode="values"):
    message = event["messages"][-1]
    if isinstance(message, tuple):
        print(message)
    else:
        message.pretty_print()

代码解析：
这段代码展示了 LangGraph 的强大之处。你可以清晰地看到 Agent 是如何进入 Loop，执行搜索，拿到结果，然后返回给 LLM 进行总结的。这比黑盒的 AgentExecutor 更加透明、可调试。

6. RAG 2.0：从 Naive RAG 到 Agentic RAG 

在 LangChain 1.0 (v0.3) 时代，简单的“切片->向量化->检索”已经不够看了。我们需要 Agentic RAG（代理式检索）。

6.1 自适应检索 (Self-RAG)

利用 LangGraph，我们可以构建一个具备“自我反思”能力的 RAG 系统：

  1. Retrieval: 检索文档。

  2. Grade: LLM 给检索到的文档打分（相关/不相关）。

  3. Rewrite: 如果相关性低，LLM 自动重写搜索 Query，重新检索。

  4. Generate: 生成答案。

  5. Hallucination Check: 检查生成的答案是否出现幻觉。

这种Loop（循环）机制，正是 LangGraph 最擅长的领域，也是企业级知识库准确率突破 90% 的关键。

7. LangSmith：全链路监控与评估 

如果你在生产环境跑 LangChain 却不用 LangSmith，那你就像是在黑夜里开车不灯。

在 v0.3 中，LangSmith 的集成更加无感。只需要设置环境变量：

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=your_api_key

运行任何 LCEL 链或 LangGraph 应用，你就可以在 LangSmith 后台看到：

  • Token 消耗与成本分析。

  • Latency 瀑布图：哪一步卡住了？

  • 数据集评估：当你修改了 Prompt，准确率是升了还是降了？

8. 常见坑点与最佳实践 (FAQ) 

Q: 以前的 ConversationalRetrievalChain 还能用吗？
A: 能用，但不推荐。它很难定制。现在的最佳实践是用 LangGraph 手动构建 retrieval 流程，代码虽然多了点，但掌控力是 100%。

Q: create_pandas_dataframe_agent 报错怎么办？
A: 检查 langchain-experimental 是否安装。另外，注意 Pydantic v2 的兼容性，尽量使用 pip install -U langchain 更新所有相关包。

Q: 如何处理超长上下文？
A: 使用 LongContextReorder 或者结合 LangGraph 实现“分治总结（Map-Reduce）”模式。

进阶篇：从 API 调用侠到架构师的必修课

在掌握了 LCEL 和 LangGraph 的基础后，我们需要深入到底层细节。这一部分将解决你在实际开发中一定会遇到的“卡脖子”问题：如何处理结构化输出？如何做大规模并发？如何持久化 Agent 记忆？

1. LCEL 接口全家桶：不止是 invoke 

很多同学只知道 invoke，但在生产环境中，流式传输和并发才是常态。Runnable 协议统一了所有组件的接口，以下是这“五大金刚”的详细对比：

1.1 接口详解对照表

接口方法 (Method) 同步/异步 核心用途 返回类型 典型场景
invoke 同步 标准单次调用 Output (Str/Dict/Obj) 简单的问答、无需流式展示的后台任务。
ainvoke 异步 invoke 的异步版本 Coroutine[Output] Web 服务（FastAPI）中避免阻塞主线程。
stream 同步 流式输出 Iterator[Chunk] 类似 ChatGPT 的打字机效果，逐个 Token 返回。
astream 异步 stream 的异步版本 AsyncIterator[Chunk] 高并发 Web 服务的流式响应。
batch 同步 批量并发处理 List[Output] 同时处理 100 个文档摘要，底层自动使用线程池。
astream_events 异步 全链路事件流 (v0.2+新增) AsyncIterator[Event] 最强接口。不仅返回最终 Token，还能返回中间步骤（Retriever 查到了什么、Tool 调用了什么）。

1.2 astream_events 实战：获取 RAG 的中间步骤

在 RAG 应用中，用户不仅想看到答案，还想看到“正在搜索知识库...”、“正在思考...”。astream_events 是实现这一功能的完美方案。

# 假设 chain 是一个 RAG 链
# version="v2" 是必须的，标准化了事件输出
async for event in chain.astream_events("如何学习 LangChain?", version="v2"):
    kind = event["event"]
    
    # 1. 监听 Retriever 开始工作
    if kind == "on_retriever_start":
        print("🔍 开始检索知识库...")
    
    # 2. 监听 Retriever 结束，获取查到的文档片段
    elif kind == "on_retriever_end":
        docs = event["data"]["output"]
        print(f"📚 检索到 {len(docs)} 篇相关文档")
        
    # 3. 监听 LLM 的流式输出
    elif kind == "on_chat_model_stream":
        content = event["data"]["chunk"].content
        if content:
            print(content, end="", flush=True)

2. Prompt Engineering 进阶：动态与样本的力量 

简单的 f-string 拼接已经过时了。在 LangChain 中，Prompt 是结构化的对象。

2.1 MessagesPlaceholder：搞定记忆注入

在 Chat 模型中，历史记录（Memory）是动态变化的，你无法预知有多少条消息。这时就需要 MessagesPlaceholder 占位符。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个暴躁的程序员，回答问题要带有偏见。"),
    # 👇 这里是关键！它会自动替换为传入的 'history' 列表中的所有消息
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

# 调用时
input_data = {
    "history": [
        ("human", "你好"), 
        ("ai", "有什么事快说，我代码还没写完。")
    ],
    "input": "Python 慢吗？"
}
# 渲染后的 Prompt 会完整包含上述对话历史

2.2 FewShotPromptTemplate：让模型举一反三

为了让模型按照特定格式输出（例如 SQL 语句），最好的办法是给它几个示例（Few-Shot）。LangChain 支持基于语义相似度动态选择示例（ExampleSelector）。

from langchain_core.prompts import FewShotChatMessagePromptTemplate, SemanticSimilarityExampleSelector
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 1. 准备示例库
examples = [
    {"input": "给我查下用户表", "output": "SELECT * FROM users;"},
    {"input": "我要看最近的订单", "output": "SELECT * FROM orders ORDER BY date DESC LIMIT 5;"},
    # ... 假设有 100 个示例
]

# 2. 使用向量库构建选择器（自动选出最相似的 k 个例子）
example_selector = SemanticSimilarityExampleSelector.from_examples(
    examples,
    OpenAIEmbeddings(),
    Chroma,
    k=1
)

# 3. 定义 FewShot 模板
few_shot_prompt = FewShotChatMessagePromptTemplate(
    input_variables=["input"],
    example_selector=example_selector, # 动态选择
    # example_prompt=... (定义单条示例的格式)
)

# 这样，当用户问 "查下库存" 时，系统会自动找出和库存相关的 SQL 示例塞入 Prompt

3. OutputParsers 全家桶与自动修复 (Retry) 

LLM 最大的痛点是“胡言乱语”。我们需要它稳定输出 JSON。

3.1 PydanticOutputParser vs JsonOutputFunctionsParser

  • PydanticOutputParser: 通用性强，适用于所有 LLM。它通过在 Prompt 中注入一段复杂的指令（Format Instructions）来告诉模型如何生成 JSON。

  • JsonOutputToolsParser / PydanticToolsParser: 推荐使用。利用 OpenAI/Anthropic 原生的 Tool Calling (Function Calling) 能力。更精准，Token 消耗更少。

3.2 炸裂实战：自动修复解析错误 (RetryOutputParser)

模型偶尔会输出格式错误的 JSON（比如少个大括号）。LangChain 提供了一个“自动纠错循环”：

from langchain.output_parsers import PydanticOutputParser, RetryOutputParser
from pydantic import BaseModel, Field

# 1. 定义数据结构
class UserInfo(BaseModel):
    name: str = Field(description="用户姓名")
    age: int = Field(description="用户年龄")

parser = PydanticOutputParser(pydantic_object=UserInfo)
retry_parser = RetryOutputParser.from_llm(parser=parser, llm=ChatOpenAI())

# 2. 构建带有纠错能力的链
# 这里的逻辑是：如果 parser 解析失败，retry_parser 会把 原始错误输出 + 报错信息 
# 重新发给 LLM，让它“再试一次”
main_chain = prompt | model
final_chain = main_chain | retry_parser

# 3. 测试（假设模型第一次返回了坏数据）
# 系统会自动进行第二次调用，修正格式
result = final_chain.invoke({"query": "Tom今年18岁"})
print(result) # UserInfo(name='Tom', age=18)

4. VectorStore 对比：四国争霸 

选哪种向量数据库？这取决于你的业务场景。以下是 LangChain 中主流库的实战对比：

特性 FAISS Chroma Pinecone Milvus
部署方式 本地库 (In-memory/Disk) 本地 / Docker SaaS 云服务 Docker / K8s 分布式
适用场景 个人项目、离线分析、POC 开发环境、中小型应用 快速上线、不想维护基础设施 企业级、亿级数据、高并发
优点 极快，无依赖，Meta 出品 API 友好，支持元数据过滤 Zero Ops，全托管，即开即用 真正的工业级，功能最全，稳定
缺点 掉电丢失（需手动 save/load），不支持复杂过滤 性能在数据量大时下降明显 贵，数据在海外（合规问题） 部署维护重，学习曲线陡峭

选型建议：

  • 新手/Demo：无脑选 Chroma 或 FAISS。

  • 生产环境(自托管)：数据量 < 100万选 PgVector (Postgres 插件)；数据量 > 1000万选 Milvus。

  • 生产环境(SaaS)：Pinecone 或 Zilliz Cloud。

5. LangGraph 高级模式：子图与持久化 

5.1 Subgraph (子图)：复杂系统的解药

当你的 Agent 既要写代码，又要画图，还要查网页，一张大图会变成“意大利面条”。
Subgraph 允许你将图嵌套。

  • 创建一个 CodingGraph 专门处理编程。

  • 创建一个 WritingGraph 专门处理写作。

  • 在 MainGraph 中，将它们视为普通的 Node 调用。

5.2 Checkpointer：给 Agent 装上“存档点”

这是 LangGraph 最具杀伤力的功能。它允许你实现：

  1. 超长对话记忆：重启服务后记忆不丢失。

  2. 人机回环 (Human-in-the-loop)：Agent 申请转账 -> 暂停 -> 人类批准 -> 恢复执行。

Postgres 持久化实战代码：

from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

# 1. 建立 DB 连接池
pool = ConnectionPool(conninfo="postgresql://user:pass@localhost:5432/db")

# 2. 初始化 Checkpointer
checkpointer = PostgresSaver(pool)

# 3. 编译图时传入 checkpointer
app = workflow.compile(checkpointer=checkpointer)

# 4. 调用时指定 thread_id
# 只要 thread_id 相同，LangGraph 就会自动从数据库加载之前的 State
config = {"configurable": {"thread_id": "user_123_session_1"}}
response = app.invoke(inputs, config=config)

6. 部署环节：LangServe 一键 API 化 

写好了代码怎么给前端调用？别去手写 Flask/FastAPI 路由了，用 LangServe。

它是 LangChain 官方推出的部署工具，基于 FastAPI，但提供了更多：

  • 自动生成 /invoke, /stream, /batch 等所有端点。

  • 自带可视化的 Playground (类似 ChatGPT 界面) 用于调试。

  • 自动生成 Swagger 文档。

6.1 server.py 标准模板

#!/usr/bin/env python
from fastapi import FastAPI
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langserve import add_routes

# 1. 定义你的链
model = ChatOpenAI()
prompt = ChatPromptTemplate.from_template("讲一个关于 {topic} 的笑话")
chain = prompt | model

# 2. 初始化 FastAPI
app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="A simple API server using LangServe",
)

# 3. 添加路由 (这就是魔法所在)
add_routes(
    app,
    chain,
    path="/joke",  # 访问路径
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

运行后，你将获得：

  • API 文档: http://localhost:8000/docs

  • 调试界面: http://localhost:8000/joke/playground (直接在这里输入 topic 测试，无需写前端代码！)

Logo
腾讯云开发者社区

腾讯云面向开发者汇聚海量精品云计算使用和开发经验，营造开放的云计算技术生态圈。

更多推荐

终极指南：Flink SQL连接器版本管理从混乱到有序的升级之路

Apache Flink作为流处理领域的佼佼者，其SQL连接器的版本管理一直是开发者面临的核心挑战。本文将系统讲解Flink SQL连接器版本管理的最佳实践，帮助你轻松应对版本兼容性问题，实现从混乱到有序的升级之旅。## 连接器版本管理的常见痛点 😫在Flink应用开发中，连接器版本管理常常让开发者头疼不已。不同版本的连接器可能导致各种兼容性问题，例如API变更、功能差异甚至运行时错误。

avatar 腾讯云开发者社区

Elasticsearch复杂数据类型终极指南：从入门到精通

Elasticsearch作为功能强大的搜索引擎，支持多种复杂数据类型，让开发者能够灵活处理各种结构化和非结构化数据。本文将带你全面了解Elasticsearch中的复杂数据类型，从基础概念到实际应用，助你轻松掌握数据建模的核心技巧。## 内部对象：构建层级化数据结构在Elasticsearch中，对象类型（Object）是最基础的复杂数据类型之一，用于表示具有嵌套关系的数据。例如，我们可

avatar 腾讯云开发者社区

如何快速搭建Neon无服务器PostgreSQL：面向初学者的完整指南

Neon是一款革命性的无服务器PostgreSQL解决方案，它通过分离存储和计算层，实现了自动扩缩容、类代码式数据库分支以及零级扩展能力。本指南将帮助你从零开始搭建Neon开发环境，体验这款创新数据库的强大功能。## 准备工作：环境要求与依赖项在开始搭建Neon环境前，请确保你的系统满足以下要求：- Linux操作系统（推荐Ubuntu 20.04+或Debian 11+）- Git

avatar 腾讯云开发者社区