LangGraph v0.1 正式发布:构建自定义认知架构的新工具
LangChain 近日正式发布了 LangGraph v0.1 的稳定版本。这一版本为开发者提供了强大的工具,能够设计和运行自定义的认知架构,将 AI 智能体(Agent)的构建推向了一个新的阶段。LangGraph 的核心设计理念是帮助开发者在 Agent 工作流中增加更好的精度和控制,以适用于现实世界系统的复杂性。
什么是 LangGraph?
LangGraph 是一个用于构建和管理复杂 Agent 的框架,允许开发者通过直观的编程接口控制 Agent 的行为。它基于图结构(Graph)来管理状态和流程,使得开发者可以定义节点(Nodes)和边(Edges),从而精确地控制代码流、提示词以及大型语言模型(LLM)的调用。
与传统的链式(Chain)处理不同,LangGraph 支持循环、条件分支以及持久化状态,这对于需要多步决策、记忆保持或人工干预的复杂任务至关重要。
核心特性详解
1. 可控性:超强的 Flow 控制流
LangGraph 的灵活 API 允许开发者设计自定义的认知架构。这意味着开发者可以完全掌控代码流、提示词以及 LLM 调用,这些调用接受用户输入并执行操作或生成响应。
- 条件分支与循环:LangGraph 支持条件分支和循环,使用户能够构建具有分层或顺序决策模式的单代理或多代理设置,从而实现更复杂的逻辑和工作流程。例如,可以根据 LLM 的输出决定是继续执行下一步还是回退重试。
- 自定义提示词工程:开发者可以根据特定领域或任务需求,对 LLM 提示词进行精细化设计,以引导模型生成更准确、更符合预期的输出。每个节点都可以拥有独立的 System Prompt。
2. 可靠性:完备的审核与中断
LangGraph 简化了在代理工作流中添加审核和质量检查的流程,确保代理在继续任务之前满足特定条件。这对于生产环境中的稳定性尤为重要。
- 质量控制关卡:开发者可以设置质量控制关卡,例如检查 LLM 输出的格式、内容或情感倾向,确保代理的行动符合预期标准。如果不符合,可以自动触发修正逻辑。
- 错误处理机制:LangGraph 允许开发者定义错误处理机制,例如在代理遇到问题时触发警报、记录日志或请求人工干预,从而提高系统的稳定性和可靠性。
- 状态持久化:通过内置的 Checkpointer 机制,LangGraph 可以在任何时间点保存和恢复状态,防止因服务重启或网络波动导致的工作流中断。
3. 灵活性:开箱即用的人机协作
LangGraph 通过内置的持久层实现了人机协作,使得人类能够参与到代理的自动化流程中。这是构建高信任度企业级应用的关键。
- 人工审核与批准:对于敏感任务,LangGraph 允许开发者设计代理在执行任务之前明确等待人类批准,例如在金融交易或医疗诊断等场景中。系统会暂停执行直到收到确认信号。
- 实时干预与调整:人类专家可以实时监控代理的行动,并在必要时进行干预或调整,例如修改代理的目标、策略或参数,以应对突发情况或优化性能。
技术实现原理
StateGraph 架构
LangGraph 的核心是 StateGraph。它由以下组件构成:
- State Schema:定义了工作流中共享的状态数据结构,通常是一个 TypedDict 或 Pydantic 模型。
- Nodes:代表图中的计算单元,可以是函数或类方法。它们接收当前状态,处理后返回更新后的状态。
- Edges:定义了节点之间的连接关系,包括普通边(无条件跳转)和条件边(根据状态判断跳转路径)。
- Checkpointer:负责状态的序列化存储,支持内存、SQL 等多种后端,确保工作流的断点续传能力。
代码示例
以下是一个简单的 Python 代码示例,展示如何定义一个包含条件分支的简单 Agent 工作流:
from typing import TypedDict, Annotated,
langgraph.graph StateGraph, END
langchain_core.messages BaseMessage
():
messages: Annotated[[BaseMessage], x, y: x + y]
should_continue:
():
{: []}
():
{: []}
workflow = StateGraph(AgentState)
workflow.add_node(, model_node)
workflow.add_node(, human_node)
workflow.set_entry_point()
workflow.add_conditional_edges(
,
state: state END,
{: , END: END}
)
workflow.add_edge(, )
app = workflow.()


