从零到1读懂LangChain和 LangGraph

醉舞经阁半卷书1

362人浏览 · 2026-05-04 15:38:35

醉舞经阁半卷书1 · 2026-05-04 15:38:35 发布

前言

LangChain和LangGraph的核心区别在于LangChain是面向线性任务流的组件化框架，而LangGraph是面向复杂状态管理的图结构工作流引擎。两者并非替代关系，而是LangChain提供基础组件能力，LangGraph负责复杂流程编排

LangChain：全栈 LLM 应用开发的 "基础底盘"，提供连接大模型与外部世界的标准化组件库
LangGraph：聚焦 "非线性、循环、多角色协作" 的 "智能工作流引擎"，专门解决 LangChain 在复杂 Agent 场景下的局限性
LangSmith：全生命周期可观测性与调试平台
LangServe：API 部署与服务化框架

形象类比：如果把构建 AI 应用比作造车，LangChain 是轮胎、发动机、方向盘等零部件供应商；LangGraph 是整车的底盘和控制系统，负责将这些零部件组装成能自主决策、灵活应对复杂路况的智能汽车；LangSmith 是 4S 店的诊断和维护系统

一.核心架构

1.1 设计范式

LangChain 采用链式（Chain）架构：
数据按固定顺序单向流动（A → B → C），适合输入→处理→输出的线性流程。其核心是模块化组件（如提示词模板、工具调用）的串联，通过LCEL（LangChain Expression Language）声明式组合，开发效率高但缺乏动态路径调整能力。
LangGraph 采用图结构（Graph）架构：
将工作流定义为节点（Node）和边（Edge）的有向图，支持条件分支、循环回溯和并行执行。其本质是状态驱动的状态机，能处理“思考→行动→观察→再思考”的闭环逻辑，例如代码生成失败后自动跳回修正节点

1.2 状态管理机

LangChain 的状态传递依赖上下文或Memory模块：
仅支持短期对话历史管理（如ConversationBufferMemory），状态随函数调用传递，难以跨多步骤持久化。适合单轮任务或简单多轮对话，但无法可靠处理长周期任务中的状态回溯。
LangGraph 显式维护中央状态对象（State）：
所有节点共享同一份状态数据，支持状态快照、故障恢复和人工干预。例如在审批流程中，若某步骤被驳回，可直接回滚到上一状态重新处理，而无需从头开始。

二关键能力对比

2.1 控制流支持

LangChain 仅支持有限条件分支：
通过ConditionalPromptSelector等工具可实现简单跳转，但循环逻辑需手动用while封装，代码易碎片化。例如实现“代码生成→测试→失败则重试”需外部循环控制。
LangGraph 原生支持复杂控制流：
- 条件边（Conditional Edge）：根据状态动态选择下一节点（如置信度<75%时触发重新检索）。
- 循环机制：允许节点跳回前置步骤（如论文评审中“检查不通过→反馈修改→重新检查”闭环）。
- 人工介入（Human-in-the-loop）：关键步骤可暂停等待人工确认，再恢复执行。

2.2 适用场景边界

适合 LangChain 的场景：

标准化的一次性任务：文本翻译、摘要生成、基础RAG问答（检索→生成单向流程）。
快速原型验证：利用丰富组件库（文档加载器、向量库接口）5分钟内搭建MVP。
无状态工具链：邮件自动回复、数据提取等无需跨步骤共享状态的轻量任务。

适合 LangGraph 的场景：

自主Agent系统：需反复“感知→决策→行动”的闭环（如代码生成器自动修复报错）。
多Agent协作：研究员Agent生成报告→编辑Agent润色→审核Agent终审的角色分工流程。
长周期任务管理：需人工审批的贷款流程、需多次迭代的科研实验设计，依赖状态持久化和回溯。

2.3 二者对比

对比维度	LangChain 1.0	LangGraph 1.0	技术影响
执行模型	线性链与简单分支	有向图 (支持循环)	LangGraph 能实现 "自我反思直到满足条件"，这是智能的核心特征
状态管理	轻量 Memory 组件，独立维护	原生内置 StateGraph，全局共享状态	LangGraph 避免了长链路中的信息丢失，支持增量更新和 Reducer 合并
循环支持	需手动实现 (递归或外部 while)	原生支持，通过边指向自身或前序节点	LangGraph 天然适合 ReAct、Reflexion 等需要迭代的 Agent 模式
条件分支	通过 RunnableBranch 实现，嵌套后可读性差	原生支持，节点函数直接返回下一节点名	LangGraph 的分支逻辑更清晰，易于调试和维护
并发处理	有限，依赖外部 asyncio 机制	原生支持并行节点执行	官方压测显示 LangGraph 并发吞吐量比 LangChain 高 3 倍
持久化能力	基础支持，需手动实现	内置检查点 (Checkpoint) 机制	LangGraph 支持断点续跑、跨会话记忆和故障恢复
人机协作	有限支持	原生支持 Human-in-the-Loop	LangGraph 可在执行中任意节点暂停，等待人类审批后继续
多智能体	需要手写大量协调逻辑	内置 Supervisor/Swarm 模式	LangGraph 大幅降低了多智能体协作系统的开发复杂度
可观测性	有限的日志输出	与 LangSmith 深度集成的图可视化	LangGraph 能清晰展示每一步的执行路径和状态变化

三应用场景

3.1 LangChain 的黄金应用场景

LangChain 在以下场景中表现最佳，能够以最少的代码实现强大的功能：

3.1.1. 标准 RAG (检索增强生成) 应用

文档问答系统
企业知识库搜索
产品手册智能助手
法律合同查询

3.1.2. 单轮次文本处理任务

文本摘要与翻译
数据提取与结构化
内容分类与情感分析
文案生成与润色

3.1.3. 简单工具调用应用

天气查询机器人
计算器助手
邮件发送与日程安排
简单的 API 调用代理

3.1.4. 快速原型验证

产品概念验证
技术可行性研究
演示系统开发

3.2 LangGraph 的不可替代场景

当你的应用遇到以下任何一种情况时，LangGraph 将成为必然选择：

3.2.1. 需要循环与反思的自主 Agent

代码生成与调试 (生成 - 测试 - 修复循环)
数学问题求解 (多步推理与验证)
研究助手 (搜索 - 分析 - 再搜索)
写作助手 (草稿 - 修改 - 润色迭代)

3.2.2. 多智能体协作系统

软件开发团队 (产品经理 + 程序员 + 测试员)
内容创作团队 (研究员 + 作者 + 编辑 + 设计师)
金融分析团队 (数据分析师 + 风险评估师 + 投资顾问)
客服团队 (自动客服 + 人工客服 + 专家支持)

3.2.3. 长时运行与复杂业务流程

客户服务工单处理系统
IT 运维自动化 (故障检测 - 诊断 - 修复)
供应链管理与优化
医疗问诊与分诊系统

3.2.4. 生产级企业应用

需要可审计性和可追溯性的系统
需要高可靠性和故障恢复能力的系统
需要人工介入审批的敏感操作
需要长期记忆和用户个性化的系统

四代码示例对比

4.1 LangChain 实现写代码

from deepseek_test import deep_llm as llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import subprocess


# 定义提示词模板
generate_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个Python专家，请根据用户需求生成代码。只返回代码，不要其他解释。"),
    ("human", "{query}")
])

verify_prompt = ChatPromptTemplate.from_messages([
    ("system", "请分析以下代码执行结果，判断是否正确。如果有错误，请指出问题并提供修正建议。"),
    ("human", "代码：{code}\n执行结果：{result}")
])

# 构建链
generate_chain = generate_prompt | llm | StrOutputParser()
verify_chain = verify_prompt | llm | StrOutputParser()


# 手动实现循环
def generate_and_verify(query, max_retries=3):
    code = generate_chain.invoke({"query": query})

    for i in range(max_retries):
        try:
            # 执行代码
            result = subprocess.run(
                ["python", "-c", code],
                capture_output=True,
                text=True,
                timeout=5
            )

            if result.returncode == 0:
                return f"代码执行成功！\n代码：\n{code}\n输出：\n{result.stdout}"

            # 验证并修正
            feedback = verify_chain.invoke({"code": code, "result": result.stderr})
            code = generate_chain.invoke({"query": f"{query}\n\n之前的代码有错误：{feedback}\n请重新生成正确的代码。"})

        except Exception as e:
            feedback = f"执行异常：{str(e)}"
            code = generate_chain.invoke({"query": f"{query}\n\n之前的代码有错误：{feedback}\n请重新生成正确的代码。"})

    return f"经过{max_retries}次尝试，未能生成正确的代码。\n最后一次尝试的代码：\n{code}"


# 使用
print(generate_and_verify("写一个函数计算斐波那契数列的第n项"))

输出结果：

4.2 LangGraph 实现方式

from deepseek_test import deep_llm as llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
import subprocess


# 定义状态
class CodeState(TypedDict):
    query: str
    code: str
    result: str
    feedback: str
    retry_count: Annotated[int, operator.add]
    max_retries: int


# 定义节点函数
def generate_code(state: CodeState):
    """生成代码"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个Python专家，请根据用户需求生成代码。只返回代码，不要其他解释。"),
        ("human", "{query}\n\n{feedback}")
    ])

    chain = prompt | llm | StrOutputParser()
    code = chain.invoke({
        "query": state["query"],
        "feedback": f"之前的代码有错误：{state.get('feedback')}" if state.get('feedback') else ""
    })

    return {"code": code, "retry_count": 1}


def execute_code(state: CodeState):
    """执行代码"""
    try:
        result = subprocess.run(
            ["python", "-c", state["code"]],
            capture_output=True,
            text=True,
            timeout=5
        )

        if result.returncode == 0:
            return {"result": f"执行成功！输出：\n{result.stdout}"}
        else:
            return {"result": f"执行失败！错误：\n{result.stderr}"}

    except Exception as e:
        return {"result": f"执行异常：{str(e)}"}


def verify_code(state: CodeState):
    """验证代码执行结果"""
    if "执行成功" in state["result"]:
        return {"feedback": ""}

    prompt = ChatPromptTemplate.from_messages([
        ("system", "请分析以下代码执行结果，判断错误原因并提供修正建议。"),
        ("human", "代码：{code}\n执行结果：{result}")
    ])

    chain = prompt | llm | StrOutputParser()
    feedback = chain.invoke({"code": state["code"], "result": state["result"]})

    return {"feedback": feedback}


# 定义条件边
def should_continue(state: CodeState):
    """决定下一步操作"""
    if state["retry_count"] >= state["max_retries"]:
        return "end"
    elif state["feedback"] == "":
        return "end"
    else:
        return "generate"


# 构建图
workflow = StateGraph(CodeState)

# 添加节点
workflow.add_node("generate", generate_code)
workflow.add_node("execute", execute_code)
workflow.add_node("verify", verify_code)

# 添加边
workflow.set_entry_point("generate")
workflow.add_edge("generate", "execute")
workflow.add_edge("execute", "verify")

# 添加条件边
workflow.add_conditional_edges(
    "verify",
    should_continue,
    {
        "generate": "generate",
        "end": END
    }
)

# 编译图
app = workflow.compile()

# 使用
result = app.invoke({
    "query": "写一个函数计算斐波那契数列的第n项",
    "max_retries": 3,
    "retry_count": 0
})

print(f"最终结果：\n{result['result']}\n\n代码：\n{result['code']}\n\n尝试次数：{result['retry_count']}")

执行结果：