使用 LangGraph 构建状态化 Agent Harness

使用 LangGraph 构建状态化 Agent Harness标题选项从零到一：使用 LangGraph 构建高度可控的状态化 Agent 系统LangGraph 实战指南：构建具有记忆和推理能力的智能 Agent Harness告别简单链：使用 LangGraph 构建复杂状态化 Agent 的完整教程掌握 LangGraph：构建企业级状态化 Agent Harness 的最佳实践从 LangChain 到 LangGraph：构建下一代状态化智能 Agent引言痛点引入你是否曾经尝试过使用 LangChain 构建一个复杂的 AI Agent，却发现简单的链式结构无法满足你的需求？当你需要处理多轮对话、保持上下文状态、实现条件分支或循环逻辑时，传统的链式结构显得力不从心。你可能发现自己在不断地 hack 代码，试图让系统按照你的意愿工作，但结果往往是代码变得难以维护和扩展。在当今的 AI 应用开发中，我们需要的不仅仅是简单的“提示词-响应”模式，而是能够模拟人类思维过程、具有状态记忆、能够做出复杂决策的智能 Agent。这些 Agent 需要能够：记住之前的交互和决策根据当前状态选择不同的行动路径处理错误和异常情况在多个工具和数据源之间协调实现复杂的推理和规划流程文章内容概述本文将带你深入了解 LangGraph——一个由 LangChain 团队开发的、专门用于构建状态化 Agent 的库。我们将从基础概念开始，逐步深入到实际应用，最终构建一个功能完整的状态化 Agent Harness。具体来说，我们将涵盖以下内容：LangGraph 的核心概念和设计理念如何设置开发环境构建基础的状态图和节点实现条件边和循环逻辑集成工具和外部 API处理多轮对话和状态管理错误处理和重试机制最佳实践和性能优化读者收益读完本文后，你将能够：理解 LangGraph 与传统 LangChain 链的区别和优势独立设计和构建复杂的状态化 Agent 系统实现具有记忆、推理和决策能力的智能应用掌握 LangGraph 的高级特性和最佳实践为你的 AI 应用添加更强大的交互和控制能力无论你是刚刚开始探索 AI Agent 开发，还是已经有一定 LangChain 使用经验但希望构建更复杂系统的开发者，本文都将为你提供有价值的 insights 和实用的代码示例。准备工作在开始我们的 LangGraph 之旅之前，让我们先确保你具备必要的知识和环境。技术栈/知识为了充分理解本文的内容，你应该具备以下基础知识：Python 编程：熟悉 Python 3.8+ 的语法和特性，包括异步编程、类型提示等。LangChain 基础：了解 LangChain 的核心概念，如 LLMs、Prompts、Chains、Tools 等。虽然不是必需的，但这将帮助你更快地理解 LangGraph。状态机概念：对有限状态机（FSM）有基本了解将有助于理解 LangGraph 的设计理念。API 调用：熟悉如何使用 Python 调用外部 API，处理 JSON 数据等。异步编程：了解 async/await 模式，因为 LangGraph 充分利用了异步编程来提高性能。环境/工具在开始编写代码之前，你需要设置好以下环境和工具：Python 环境：确保你安装了 Python 3.8 或更高版本。我推荐使用 conda 或 venv 来创建虚拟环境，以避免依赖冲突。# 使用 venv 创建虚拟环境python-mvenv langgraph-env# 激活虚拟环境# Windowslanggraph-env\Scripts\activate# macOS/Linuxsourcelanggraph-env/bin/activate必要的库：我们需要安装以下核心库：langgraph：我们的主角，用于构建状态图langchain：提供 LLM 集成、工具等langchain-openai：OpenAI 模型的集成python-dotenv：用于管理环境变量pipinstalllanggraph langchain langchain-openai python-dotenvAPI 密钥：你需要一个 OpenAI API 密钥（或其他支持的 LLM 提供商的密钥）。你可以从 OpenAI 官网获取。代码编辑器：选择你喜欢的代码编辑器，如 VS Code、PyCharm 等。可选工具：Jupyter Notebook：用于实验和调试Git：用于版本控制一旦你准备好了这些，我们就可以开始探索 LangGraph 的世界了！核心概念：理解 LangGraph在开始编写代码之前，让我们先深入理解 LangGraph 的核心概念和设计理念。这将帮助你更好地理解后续的代码示例和最佳实践。问题背景传统的 LangChain 链（Chains）虽然强大，但它们本质上是线性的。它们按照预定义的顺序执行一系列步骤，每个步骤的输出作为下一个步骤的输入。这种线性结构对于简单的任务来说足够了，但当我们需要构建更复杂的应用时，就会遇到以下限制：缺乏状态管理：传统链在执行过程中很难维护和更新复杂的状态。难以实现条件逻辑：虽然可以使用RouteChain等实现简单的条件分支，但复杂的条件逻辑仍然难以处理。不支持循环：传统链很难实现循环执行某个步骤直到满足特定条件的逻辑。错误处理有限：当链中的某个步骤失败时，很难实现灵活的错误恢复策略。可观察性差：很难跟踪和调试链的执行过程。为了解决这些问题，LangChain 团队开发了 LangGraph——一个基于状态机概念的库，专门用于构建复杂的、状态化的 AI Agent。核心概念LangGraph 的核心概念非常简单，但却非常强大。让我们逐一介绍：1. 状态（State）状态是 LangGraph 的核心概念之一。它代表了 Agent 在执行过程中所维护的所有信息。状态可以包括：用户的输入和历史对话Agent 的中间思考和推理从工具或 API 获取的结果任何其他需要在执行过程中保持的数据在 LangGraph 中，状态通常定义为一个 TypedDict 或 Pydantic 模型，这样可以提供类型安全和更好的开发体验。fromtypingimportTypedDict,Annotated,Sequencefromlangchain_core.messagesimportBaseMessageimportoperatorclassAgentState(TypedDict):# 消息历史messages:Annotated[Sequence[BaseMessage],operator.add]# 当前任务current_task:str# 任务结果task_result:str# 错误信息error:str注意这里使用了Annotated和operator.add，这告诉 LangGraph 当更新messages字段时，应该将新消息添加到现有列表中，而不是替换整个列表。2. 节点（Nodes）节点是 LangGraph 中的执行单元。每个节点代表 Agent 可以执行的一个动作或函数。节点接收当前状态作为输入，并返回一个更新后的状态（或状态的一部分）。节点可以是：调用 LLM 进行推理执行某个工具或 API处理和转换数据做出决策下面是一个简单的节点示例：fromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage llm=ChatOpenAI(model="gpt-4")defcall_model(state:AgentState)-AgentState:"""调用 LLM 生成响应"""messages=state["messages"]response=llm.invoke(messages)# 返回更新后的状态return{"messages":[response]}3. 边（Edges）边定义了节点之间的连接关系，决定了状态流转的路径。边可以是：普通边：从一个节点无条件地流向另一个节点条件边：根据当前状态的某些条件决定下一个节点入口点：定义图的起始节点结束点：定义图的结束节点条件边是 LangGraph 最强大的特性之一，它允许我们根据当前状态动态地决定执行路径。fromtypingimportLiteraldefshould_continue(state:AgentState)-Literal["tool","end"]:"""决定是否需要继续执行工具或结束"""messages=state["messages"]last_message=messages[-1]# 如果有工具调用，则继续执行工具iflast_message.tool_calls:return"tool"# 否则结束return"end"4. 图（Graph）图是 LangGraph 的顶层容器，它将节点和边组合在一起，形成一个完整的执行流程。LangGraph 提供了StateGraph类来构建状态图。fromlanggraph.graphimportStateGraph,END# 创建图workflow=StateGraph(AgentState)# 添加节点workflow.add_node("agent",call_model)workflow.add_node("tool",execute_tools)# 设置入口点workflow.set_entry_point("agent")# 添加条件边workflow.add_conditional_edges("agent",should_continue,{"tool":"tool","end":END,},)# 添加普通边workflow.add_edge("tool","agent")# 编译图app=workflow.compile()概念之间的关系为了更好地理解这些核心概念之间的关系，让我们使用一个 ER 图来表示：containscontainsmanagesreadsupdatesconnectsmay_haveGRAPHNODEEDGESTATECONDITION这个图展示了：一个图（Graph）包含多个节点（Node）和边（Edge）图管理着一个状态（State）节点读取和更新状态边连接节点，并可能有条件（Condition）状态流转过程让我们再用一个流程图来展示 LangGraph 中状态的流转过程：