LangChain 2026 深度解析：从 Chain 到 Agent，LLM 应用开发的事实标准全面进化

LangChain 在 2026 年完成了一次架构级重构：Agent 取代 Chain 成为核心抽象、LangGraph 状态机成为编排引擎、Deep Agents 支持多智能体协作、MCP Adapters 打通工具生态、LangSmith 提供全链路可观测性。本文深度解析 LangChain 2026 的架构变革、核心 API 演进、实战代码对比（2024 vs 2026），以及从旧版迁移的完整指南。

一、LangChain 2026 架构全景

LangChain 2026 生态图谱：

                    ┌─────────────────────────┐
                    │      LangSmith          │
                    │  全链路可观测性平台      │
                    │  Tracing / Eval / Hub   │
                    └───────────┬─────────────┘
                                │
        ┌───────────────────────┼───────────────────────┐
        │                       │                       │
┌───────┴───────┐     ┌────────┴────────┐     ┌────────┴────────┐
│  LangGraph    │     │  Deep Agents    │     │  LangChain Core │
│  状态机编排    │     │  多智能体协作    │     │  组件+集成      │
│               │     │                 │     │                 │
│ · StateGraph  │     │ · Plan & Execute│     │ · Chat Models   │
│ · State       │     │ · SubAgents     │     │ · Embeddings    │
│ · Checkpointer│     │ · File System   │     │ · Vector Stores │
│ · Channels    │     │ · Human-in-loop │     │ · Tools         │
└───────┬───────┘     └────────┬────────┘     └────────┬────────┘
        │                       │                       │
        └───────────────────────┼───────────────────────┘
                                │
                    ┌───────────┴─────────────┐
                    │    MCP Adapters         │
                    │  框架无关的工具协议      │
                    └─────────────────────────┘

二、核心变革：Chain → Agent

2.1 2024 vs 2026 API 对比

# ====== 2024 年：Chain 模式（已废弃） ======
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

prompt = PromptTemplate(
    template="请根据以下问题回答：{question}",
    input_variables=["question"]
)
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run(question="什么是微服务？")

# 多步 Chain 串联
from langchain.chains import SimpleSequentialChain
chain1 = LLMChain(llm=llm1, prompt=prompt1)
chain2 = LLMChain(llm=llm2, prompt=prompt2)
combined = SimpleSequentialChain(chains=[chain1, chain2])

# ❌ 问题：
# 1. Chain 是线性管道，不支持条件分支
# 2. 状态管理靠手动传递
# 3. 无法实现循环（如"反思-修正"循环）
# 4. 错误处理薄弱

# ====== 2026 年：Agent 模式（推荐） ======
from langchain.agents import create_agent
from langchain.tools import tool

# 定义工具
@tool
def search_docs(query: str) -> str:
    """搜索技术文档"""
    return vector_store.similarity_search(query)

@tool
def write_code(code: str, filename: str) -> str:
    """写入代码文件"""
    with open(filename, 'w') as f:
        f.write(code)
    return f"已写入 {filename}"

@tool
def run_tests(filename: str) -> str:
    """运行测试"""
    import subprocess
    result = subprocess.run(
        ["pytest", filename], capture_output=True, text=True
    )
    return result.stdout

# 创建 Agent
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[search_docs, write_code, run_tests],
    system_prompt="你是一个全栈开发工程师。请根据需求搜索文档、编写代码并测试。",
)

# 执行（Agent 自动规划步骤、选择工具、处理错误）
result = agent.invoke({
    "messages": [{"role": "user", "content": "帮我写一个用户认证模块，包含 JWT 登录和注册功能"}]
})

# ✅ Agent 模式优势：
# 1. 自动规划（Plan → Execute → Reflect）
# 2. 工具自主选择
# 3. 支持条件分支和循环
# 4. 内置状态管理（LangGraph）

2.2 统一 create_agent API

# 2026：所有 Agent 类型统一用一个 API

from langchain.agents import create_agent, AgentType

# ReAct 模式（最常用）
react_agent = create_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[search_tool, calculator_tool],
    agent_type=AgentType.REACT,
)

# Plan-and-Execute 模式（复杂任务）
plan_agent = create_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[search_tool, code_tool, test_tool],
    agent_type=AgentType.PLAN_EXECUTE,
    max_iterations=10,
    early_stopping=True,
)

# 结构化输出模式（JSON Schema）
structured_agent = create_agent(
    model=ChatOpenAI(model="gpt-4o"),
    output_schema={
        "type": "object",
        "properties": {
            "analysis": {"type": "string"},
            "recommendation": {"type": "string"},
            "confidence": {"type": "number"},
        }
    }
)

# 2024 旧版（多种初始化方式，已废弃）：
# initialize_agent → ❌
# AgentExecutor.from_agent_and_tools → ❌
# create_openai_functions_agent → ❌

三、LangGraph：状态机编排引擎

3.1 为什么需要状态机？

Agent 的核心难点不是"调用工具"，而是"控制流程"：

用户请求
    │
    ├─→ 需要搜索？ → 搜索 → 分析结果
    │                      │
    │              结果不够？───┘
    │
    ├─→ 需要写代码？ → 写代码 → 跑测试
    │                          │
    │                   测试失败？─── 修正代码
    │
    └─→ 需要人工确认？ → 暂停 → 等待输入 → 继续

这不是线性管道（Chain），而是有向图（Graph）——这正是 LangGraph 的核心价值。

3.2 StateGraph 基础

from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import MessagesState
from langgraph.checkpoint.memory import MemorySaver

# 1. 定义状态
class AgentState(MessagesState):
    """继承 MessagesState，自带 messages 字段"""
    documents: list[str]      # 搜索到的文档
    code: str                 # 生成的代码
    test_result: str          # 测试结果
    iteration: int            # 当前迭代次数
    max_iterations: int = 3   # 最大迭代次数

# 2. 定义节点
def search_node(state: AgentState) -> AgentState:
    """搜索文档"""
    last_msg = state["messages"][-1].content
    docs = search_tool.invoke(last_msg)
    return {**state, "documents": docs}

def code_node(state: AgentState) -> AgentState:
    """生成代码"""
    docs_context = "\n".join(state["documents"])
    prompt = f"根据以下文档编写代码：\n{docs_context}\n\n需求：{state['messages'][0].content}"
    code = llm.invoke(prompt).content
    return {**state, "code": code}

def test_node(state: AgentState) -> AgentState:
    """运行测试"""
    result = subprocess.run(["pytest", "-v"], capture_output=True, text=True)
    passed = result.returncode == 0
    return {**state, "test_result": result.stdout, "iteration": state["iteration"] + 1}

def router(state: AgentState) -> str:
    """路由决策"""
    if state["iteration"] >= state["max_iterations"]:
        return "end"
    if "FAILED" in state.get("test_result", ""):
        return "code"  # 测试失败，回到代码节点
    return "end"       # 测试通过，结束

# 3. 构建图
graph = StateGraph(AgentState)

graph.add_node("search", search_node)
graph.add_node("code", code_node)
graph.add_node("test", test_node)

graph.add_edge(START, "search")
graph.add_edge("search", "code")
graph.add_edge("code", "test")
graph.add_conditional_edges("test", router, {
    "code": "code",  # 测试失败 → 修改代码
    "end": END,      # 完成
})

# 4. 编译（带持久化）
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# 5. 执行
config = {"configurable": {"thread_id": "session-001"}}
result = app.invoke(
    {"messages": [{"role": "user", "content": "实现一个 LRUCache"}]},
    config=config
)

3.3 Human-in-the-Loop

from langgraph.graph import StateGraph, END, START
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver

def review_node(state: AgentState) -> AgentState:
    """人工审核节点：暂停等待用户确认"""
    # interrupt() 会暂停图执行，等待人类输入
    feedback = interrupt({
        "question": "请审核以下代码，确认是否通过：",
        "code": state["code"],
        "options": ["通过", "需要修改", "拒绝"]
    })
    
    return {**state, "feedback": feedback}

def router(state: AgentState) -> str:
    if state.get("feedback") == "通过":
        return "deploy"
    elif state.get("feedback") == "需要修改":
        return "code"  # 回去修改
    else:
        return "end"

# 使用：
result = app.invoke(
    {"messages": [{"role": "user", "content": "实现排序算法"}]},
    config=config
)
# 图在 review_node 处暂停，返回 interrupt 数据

# 用户提供反馈后继续：
result = app.invoke(
    Command(resume="需要修改：请用快排替代冒泡"),
    config=config
)
# 图从 review_node 继续，路由到 code 节点

四、Deep Agents：多智能体协作

4.1 SubAgent 模式

from langchain.agents import create_agent
from langchain_deep_agents import DeepAgent, SubAgent

# 创建专家子 Agent
code_agent = SubAgent(
    name="coder",
    model=ChatOpenAI(model="gpt-4o"),
    tools=[write_code, run_tests, search_docs],
    system_prompt="你是一个高级后端开发工程师，擅长 Python 和 Go。",
)

review_agent = SubAgent(
    name="reviewer",
    model=ChatOpenAI(model="gpt-4o"),
    tools=[read_code, lint_check],
    system_prompt="你是一个代码审查专家，注重代码质量和安全性。",
)

research_agent = SubAgent(
    name="researcher",
    model=ChatOpenAI(model="gpt-4o"),
    tools=[search_web, search_docs, summarize],
    system_prompt="你是一个技术调研专家，擅长收集和分析技术方案。",
)

# 组装多智能体团队
team = DeepAgent(
    agents=[research_agent, code_agent, review_agent],
    coordinator_model=ChatOpenAI(model="gpt-4o"),
    mode="sequential",  # sequential | parallel | hierarchical
)

# 执行复杂任务
result = team.invoke(
    "调研并实现一个基于 Redis 的分布式锁方案，要求："
    "1. 支持 TTL 自动过期"
    "2. 防止死锁"
    "3. 包含单元测试"
    "4. 代码审查通过"
)

# Deep Agent 执行流程：
# 1. Coordinator 分析任务，分配给 researcher
# 2. Researcher 搜索 Redis 分布式锁的最佳实践
# 3. Researcher 输出调研报告
# 4. Coordinator 将调研结果 + 需求交给 coder
# 5. Coder 编写实现代码和测试
# 6. Coordinator 将代码交给 reviewer
# 7. Reviewer 审查代码，给出反馈
# 8. 如果审查不通过，Coordinator 安排 coder 修改
# 9. 所有子 Agent 通过后，输出最终结果

4.2 文件系统集成

from langchain_deep_agents import DeepAgent, SubAgent

# 2026 新特性：Agent 可以直接操作文件系统
agent = DeepAgent(
    agents=[coder_agent],
    workspace="/path/to/project",  # 沙箱目录
    allowed_paths=["/path/to/project/src"],  # 允许访问的路径
)

# Agent 可以：
# - 读取项目文件（理解现有代码）
# - 创建/修改文件（写代码）
# - 运行命令（测试、构建）
# - 查看目录结构（导航项目）

result = agent.invoke(
    "在 src/auth/ 目录下实现 JWT 认证模块，"
    "参考 src/models/ 下已有的 User 模型，"
    "并编写对应的单元测试。"
)

五、MCP Adapters：框架无关的工具协议

5.1 什么是 MCP？

MCP（Model Context Protocol）是 Anthropic 提出的开放协议，让 AI 工具可以跨框架互操作。LangChain 2026 内置 MCP 适配器。

from langchain_mcp_adapters import MCPToolkit

# 连接 MCP 服务器（可以是任何实现 MCP 协议的服务）
toolkit = MCPToolkit(
    server_url="http://localhost:3000/mcp",
    # 或者本地 stdio 方式：
    # command="npx",
    # args=["-y", "@anthropic/mcp-server-filesystem", "/path/to/dir"],
)

# 自动发现工具
tools = toolkit.get_tools()
# [
#   Tool(name="read_file", description="读取文件内容"),
#   Tool(name="write_file", description="写入文件"),
#   Tool(name="list_directory", description="列出目录"),
#   Tool(name="search_code", description="搜索代码"),
# ]

# 直接用于 Agent
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=tools,
)

# MCP 的价值：
# 1. 一次实现，所有框架可用（LangChain / Claude / Cursor）
# 2. 标准化工具接口
# 3. 远程工具服务器（团队共享工具）

5.2 自定义 MCP Server

# 用 FastMCP 创建自定义 MCP 服务器（5 分钟）
from fastmcp import FastMCP

mcp = FastMCP("my-tools")

@mcp.tool()
def query_database(sql: str) -> str:
    """执行 SQL 查询"""
    import sqlite3
    conn = sqlite3.connect("app.db")
    cursor = conn.cursor()
    cursor.execute(sql)
    results = cursor.fetchall()
    conn.close()
    return str(results)

@mcp.tool()
def send_notification(message: str, channel: str = "general") -> str:
    """发送通知到 Slack"""
    # ... Slack API 调用
    return f"已发送到 {channel}"

@mcp.tool()
def deploy_service(service_name: str) -> str:
    """部署服务到 Kubernetes"""
    # ... kubectl apply
    return f"{service_name} 部署成功"

# 启动服务器（stdio 模式）
if __name__ == "__main__":
    mcp.run()

# 其他 LangChain Agent 即可通过 MCP 调用这些工具

六、中间件系统

6.1 请求/响应拦截

from langchain.agents import create_agent
from langchain.middleware import Middleware, MiddlewareContext

# 日志中间件
class LoggingMiddleware(Middleware):
    async def on_request(self, ctx: MiddlewareContext):
        print(f"[请求] 工具: {ctx.tool_name}, 输入: {ctx.input}")
    
    async def on_response(self, ctx: MiddlewareContext):
        print(f"[响应] 工具: {ctx.tool_name}, 输出长度: {len(str(ctx.output))}")
    
    async def on_error(self, ctx: MiddlewareContext, error: Exception):
        print(f"[错误] 工具: {ctx.tool_name}, 错误: {error}")

# Token 计数中间件
class TokenCounterMiddleware(Middleware):
    def __init__(self):
        self.total_tokens = 0
    
    async def on_response(self, ctx: MiddlewareContext):
        if hasattr(ctx.response, 'token_usage'):
            self.total_tokens += ctx.response.token_usage.total_tokens
            print(f"累计 Token: {self.total_tokens}")

# 重试中间件
class RetryMiddleware(Middleware):
    def __init__(self, max_retries=3, delay=1.0):
        self.max_retries = max_retries
        self.delay = delay
    
    async def on_error(self, ctx: MiddlewareContext, error: Exception):
        if ctx.retry_count < self.max_retries:
            ctx.retry_count += 1
            await asyncio.sleep(self.delay * ctx.retry_count)
            ctx.retry = True  # 告诉框架重试

# 使用中间件
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[search_tool, code_tool],
    middleware=[
        LoggingMiddleware(),
        TokenCounterMiddleware(),
        RetryMiddleware(max_retries=3),
    ]
)

七、LangSmith：全链路可观测性

7.1 Tracing

import os
os.environ["LANGSMITH_API_KEY"] = "ls__..."
os.environ["LANGSMITH_TRACING"] = "true"

# 所有 LangChain 调用自动上报 Trace
result = agent.invoke({"messages": [{"role": "user", "content": "..."}]})

# LangSmith Dashboard 显示：
# 1. 完整调用链（LLM → Tool → LLM → Tool ...）
# 2. 每次 LLM 调用的 prompt/completion
# 3. 每次 Tool 调用的输入/输出
# 4. Token 消耗和延迟
# 5. 错误堆栈

7.2 Evaluation

from langsmith.evaluation import evaluate, LangChainStringEvaluator

# 定义评估数据集
dataset = [
    {"input": "实现二分查找", "reference": "必须包含递归和迭代两种实现"},
    {"input": "实现快速排序", "reference": "必须包含单元测试"},
]

# 自动评估
def correctness_evaluator(run, example) -> dict:
    prediction = run.outputs["messages"][-1].content
    reference = example.outputs["reference"]
    # 用 LLM 判断输出是否满足要求
    score = evaluator_llm.invoke(
        f"判断以下输出是否满足要求：\n要求：{reference}\n输出：{prediction}\n评分(0-10)："
    )
    return {"score": int(score.content), "key": "correctness"}

results = evaluate(
    agent.invoke,
    data=dataset,
    evaluators=[correctness_evaluator],
)

八、从 2024 迁移到 2026

8.1 迁移清单

# ====== 迁移 1: Chain → Agent ======

# 旧版（2024）
from langchain.chains import LLMChain, RetrievalQA
chain = RetrievalQA.from_chain_type(llm=llm, retriever=retriever)

# 新版（2026）
agent = create_agent(
    model=llm,
    tools=[retriever.as_tool()],
)

# ====== 迁移 2: AgentExecutor → create_agent ======

# 旧版
from langchain.agents import initialize_agent, AgentExecutor
agent = initialize_agent(tools, llm, agent="zero-shot-react-description")
agent_executor = AgentExecutor(agent=agent, tools=tools)

# 新版
agent = create_agent(model=llm, tools=tools)

# ====== 迁移 3: ConversationChain → MessagesState ======

# 旧版
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory
chain = ConversationChain(llm=llm, memory=ConversationBufferMemory())

# 新版（LangGraph + MemorySaver）
from langgraph.graph import StateGraph
from langgraph.graph.message import MessagesState
from langgraph.checkpoint.memory import MemorySaver

graph = StateGraph(MessagesState)
# ... 定义节点和边 ...
app = graph.compile(checkpointer=MemorySaver())

# ====== 迁移 4: OutputParser → 结构化输出 ======

# 旧版
from langchain.output_parsers import PydanticOutputParser
parser = PydanticOutputParser(pydantic_object=AnalysisResult)

# 新版（模型原生支持）
agent = create_agent(
    model=llm,
    output_schema=AnalysisResult.model_json_schema(),
)

8.2 常见问题

# Q: 旧版 PromptTemplate 还能用吗？
# A: 可以，但推荐用 ChatPromptTemplate
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{role}"),
    ("human", "{question}"),
])

# Q: Memory 怎么迁移？
# A: 用 LangGraph 的 checkpointer 替代
# 旧版: ConversationBufferMemory / ConversationSummaryMemory
# 新版: MemorySaver / SqliteSaver / PostgresSaver

# Q: 工具定义方式变了？
# A: @tool 装饰器不变，但推荐加类型提示和文档字符串
@tool
def search(query: str, top_k: int = 5) -> list[dict]:
    """搜索文档，返回 top_k 个结果
    
    Args:
        query: 搜索关键词
        top_k: 返回结果数量
    """
    return vector_store.similarity_search(query, k=top_k)

九、实战：构建企业知识库 Agent

# 完整实战：企业知识库问答 Agent

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import MessagesState
from langgraph.checkpoint.memory import MemorySaver

# 1. 加载文档
loader = DirectoryLoader("./docs", glob="**/*.md", loader_cls=TextLoader)
documents = loader.load()

# 2. 切分
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(documents)

# 3. 向量化
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./chroma_db")

# 4. 定义工具
@tool
def search_knowledge_base(query: str) -> str:
    """搜索企业知识库"""
    docs = vectorstore.similarity_search(query, k=3)
    return "\n---\n".join(doc.page_content for doc in docs)

@tool
def get_related_documents(topic: str) -> str:
    """获取相关文档列表"""
    docs = vectorstore.similarity_search(topic, k=10)
    return "\n".join(f"- {doc.metadata.get('source', 'unknown')}" for doc in docs)

# 5. 创建 Agent
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o", temperature=0),
    tools=[search_knowledge_base, get_related_documents],
    system_prompt=(
        "你是企业内部知识库助手。回答问题时：\n"
        "1. 先搜索知识库获取相关文档\n"
        "2. 基于文档内容回答，标注来源\n"
        "3. 如果知识库中没有相关信息，明确告知\n"
        "4. 不要编造信息"
    ),
)

# 6. 使用
result = agent.invoke({
    "messages": [{"role": "user", "content": "公司的年假政策是什么？"}]
})

print(result["messages"][-1].content)
# 输出：根据《员工手册 v3.2》第 5 章，公司年假政策如下：
# - 入职满 1 年：5 天年假
# - 入职满 3 年：10 天年假
# - 入职满 5 年：15 天年假
# 来源：docs/hr/employee-handbook-v3.2.md

十、总结

10.1 LangChain 2026 核心变化

10.2 选型建议

✅ 推荐用 LangChain 2026 的场景：
  1. 复杂 Agent 应用（多步推理、工具调用、循环修正）
  2. 多智能体协作（代码生成 + 审查 + 测试）
  3. 需要 Human-in-the-Loop 的场景
  4. 企业级应用（需要可观测性、持久化）

⚠️ 不需要 LangChain 的场景：
  1. 简单的单轮对话（直接用 OpenAI SDK）
  2. 纯 RAG 检索（LlamaIndex 更轻量）
  3. 不需要 Agent 的文档问答

🚀 迁移步骤：
  1. pip install -U langchain langgraph
  2. Chain → Agent（create_agent）
  3. Memory → Checkpointer（MemorySaver）
  4. AgentExecutor → create_agent
  5. 接入 LangSmith（免费版够用）

一句话总结：LangChain 2026 的核心转变是从"链式调用工具"到"Agent 自主决策"。LangGraph 状态机让 Agent 可以处理复杂的控制流（分支、循环、暂停），Deep Agents 让多个专业 Agent 协作完成复杂任务，MCP 让工具生态标准化。如果你还在用 2024 版的 Chain 和 AgentExecutor，是时候迁移了。

参考资源：

• LangChain 官方文档：https://python.langchain.com/docs/
• LangGraph 文档：https://langchain-ai.github.io/langgraph/
• Deep Agents：https://github.com/langchain-ai/deep-agents
• MCP 协议：https://modelcontextprotocol.io/
• LangSmith：https://smith.langchain.com/
• 迁移指南：https://python.langchain.com/docs/versions/migrating_chains/