前言

在大模型应用开发(LLM App Development)的早期,LangChain 几乎是开发者的首选,它像胶水一样把模型、提示词和数据连接在一起。但随着业务越来越复杂,我们开始需要构建 Agent(智能体)——也就是那些能自主思考、循环执行、甚至自我纠错的系统。

这时候,线性的“链(Chain)”就不够用了。我们需要“图(Graph)”和“循环(Loop)”。LangGraph 因此应运而生。

本文将带你快速厘清 LangChain 与 LangGraph 的区别,并通过一个经典的**“开发者-测试者(Coder-Tester)”多智能体协作案例**,展示如何用 LangGraph 构建一个能自动修复 Bug 的编程助手。


一、 核心概念:LangChain vs LangGraph

在开始写代码之前,我们需要先理解为什么要从 Chain 升级到 Graph。

1. LangChain:线性的流水线

LangChain 的核心是 DAG(有向无环图)。任务通常是线性的:

输入 -> 步骤A -> 步骤B -> 步骤C -> 输出

它非常适合 RAG(检索增强生成)或简单的问答机器人,因为流程是确定的,只有向前走,没有回头路。

2. LangGraph:有“记忆”的循环

LangGraph 是建立在 LangChain 之上的扩展。它的核心特性是 State(状态)Cycle(循环)

输入 -> 思考 -> 执行 -> 观察结果 -> (如果不满意) -> 回到思考 -> ...

它允许我们将流程定义为图中的节点(Nodes)和边(Edges),并且可以根据条件(Conditional Edges)决定是继续下一步,还是回退重试。这正是构建能够自我纠错的 Agent 所需的基础。

特性

LangChain (LCEL)

LangGraph

结构

线性链 (DAG)

图结构 (支持循环 Cycle)

状态管理

较弱,主要靠传参

强,全局状态 (State) 共享

适用场景

简单问答、数据处理流水线

复杂 Agent、多步推理、自我纠错


二、 实战案例:构建“自动纠错”编程助手

为了演示 LangGraph 的强大,我们来构建一个多智能体协作系统。

场景描述

我们需要两个 AI 角色配合完成任务:

  1. Coder(开发者):负责编写代码。
  2. Tester(测试者):负责运行代码并反馈结果。

工作流逻辑:
Coder 写代码 -> Tester 运行 -> 如果报错,把错误信息扔回给 Coder 重写 -> 直到测试通过或达到最大重试次数。


三、 代码实现 (Python)

1. 环境准备

首先,你需要安装必要的库:



pip install langgraph langchain langchain_openai
2. 定义“共享状态” (The State)

这是多智能体协作的“黑板”,所有节点都能读写这里的信息。

from typing import TypedDict, List, Annotated
import operator

class AgentState(TypedDict):
    messages: List[str]      # 记录用户的原始需求
    code: str                # 存放当前生成的代码
    test_output: str         # 存放测试运行结果或报错信息
    iterations: int          # 记录重试次数,防止死循环
3. 定义节点 (The Nodes)

我们定义两个函数,分别代表 Coder 和 Tester 的逻辑。

Coder 节点:负责写代码和改代码

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o") # 或者其他你喜欢的模型

def coder_node(state: AgentState):
    task = state["messages"][-1]
    error = state.get("test_output", "")
    current_iter = state.get("iterations", 0)
    
    # 构建 Prompt:如果有报错,要求根据报错修复
    if error and error != "Success":
        prompt = f"""
        任务:{task}
        
        上次运行代码报错如下:
        {error}
        
        请根据报错信息,重写修复后的 Python 代码。只返回代码,不要其他废话。
        """
    else:
        prompt = f"任务:{task}\n\n请编写 Python 代码解决该问题。只返回代码。"
    
    # 调用模型 (此处简化了代码提取逻辑)
    response = llm.invoke(prompt)
    code = response.content.strip().replace("```python", "").replace("```", "")
    
    print(f"\n[Coder] 第 {current_iter + 1} 次编写代码...")
    
    return {
        "code": code,
        "iterations": current_iter + 1
    }

Tester 节点:负责“质检”

def tester_node(state: AgentState):
    code = state["code"]
    print("[Tester] 正在运行测试...")
    
    try:
        # 警告:生产环境请使用沙箱执行代码 (如 e2b 或 docker)
        # 这里为了演示,使用 exec 模拟运行
        exec_globals = {}
        exec(code, exec_globals)
        return {"test_output": "Success"}
    except Exception as e:
        return {"test_output": f"Error: {str(e)}"}
4. 定义控制流 (The Router)

这是 LangGraph 的核心,决定图的走向。

from langgraph.graph import END

def router(state: AgentState):
    result = state["test_output"]
    iterations = state["iterations"]
    
    # 1. 成功则结束
    if result == "Success":
        print(">>> 测试通过,任务完成!")
        return "end"
    
    # 2. 超过3次重试强制结束
    if iterations >= 3:
        print(">>> 重试次数过多,任务失败。")
        return "end"
    
    # 3. 失败则回退给 Coder
    print(f">>> 检测到错误,路由回 Coder 进行修复...")
    return "retry"
5. 组装图 (Build Graph)

最后,把积木搭起来。

from langgraph.graph import StateGraph

# 初始化图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("coder", coder_node)
workflow.add_node("tester", tester_node)

# 设置起点
workflow.set_entry_point("coder")

# 添加普通边:Coder 写完必然去 Tester
workflow.add_edge("coder", "tester")

# 添加条件边:Tester 之后,由 router 决定去向
workflow.add_conditional_edges(
    "tester",           # 上一个节点
    router,             # 路由函数
    {                   # 映射关系
        "end": END,
        "retry": "coder"
    }
)

# 编译应用
app = workflow.compile()
6. 运行测试
# 这是一个会故意出错的任务(斐波那契数列,容易出现递归深度错误或语法问题)
initial_state = {
    "messages": ["写一个计算斐波那契数列第100项的函数,确保打印结果"],
    "iterations": 0,
    "test_output": ""
}

# 启动图
for s in app.stream(initial_state):
    pass 
    # stream 会逐步输出每个节点的状态,这里略过打印细节

四、 运行过程模拟

当你运行上述代码时,可能会看到这样的日志流:

  1. [Coder] 第一次写代码,使用了简单的递归 fib(n-1) + fib(n-2)
  2. [Tester] 运行代码,因为计算第100项,导致计算超时或极其缓慢(或者如果模型写错语法,直接报错)。
  3. [Router] 捕获到问题,决定 retry
  4. [Coder] 收到反馈:“运行超时/报错”。思考后改为动态规划或迭代法。
  5. [Tester] 再次运行。
  6. [Router] 发现结果 "Success",路由至 END

总结

通过这个案例,我们可以看到 LangGraph 相比 LangChain 的质变:

  • 容错性:错误不再是程序的终点,而是下一次迭代的输入。
  • 状态流转:多智能体之间通过共享状态(State)实现了无缝的上下文传递。

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐