LangGraph实战编码：核心概念、开发范式与高级特性

目录状态定义Agent的“记忆中枢”节点开发纯函数与异常捕获路由设计条件边与循环保护高级特性工具调用与人工中断实战编码的核心价值在上一篇教程中我们已经从零开始搭建好了支持可视化调试的LangGraph开发环境。很多人反馈“环境搭好了但代码到底该怎么写才规范为什么我写的Agent总是状态混乱、循环失控甚至直接报错”其实LangGraph的编码核心就三件事定义好状态、写好纯函数节点、连对路由边。只要掌握这三个核心范式再结合工具调用、人工中断等高级特性你就能写出稳定、可维护的Agent。本文将以“自动周报Agent”为贯穿案例手把手带你掌握LangGraph的实战编码规范所有代码均可直接运行。状态定义Agent的“记忆中枢”State是LangGraph的灵魂它决定了节点之间传递什么数据。很多新手容易犯的错误是把State当成普通字典随便写导致节点之间数据丢失或覆盖。规范的写法是使用TypedDict或Pydantic模型定义State并用Annotated指定Reducer。Reducer决定了状态字段的更新方式比如消息列表要用add_messages追加而不是直接覆盖。以下是一个规范的周报Agent状态定义直接复制即可使用from typing import TypedDict, Annotated from langgraph.graph import add_messages class WeeklyReportState(TypedDict): raw_data: str # 原始工作数据如Git日志、任务列表 draft: str # 周报初稿 final_report: str # 最终通过的周报 retry_count: int # 重试次数防止无限循环 feedback: str # 审核节点的反馈如“缺少量化数据” messages: Annotated[list, add_messages] # 对话历史自动追加避坑提示不要直接修改传入的state对象节点函数必须返回一个字典LangGraph会自动用Reducer合并更新。比如想更新retry_count只需返回{retry_count: state[retry_count] 1}而不是state[retry_count] 1。直接修改state会导致状态更新丢失Agent无法感知到重试次数的变化最终陷入死循环。节点开发纯函数与异常捕获节点是Agent的执行单元每个节点只负责一件事。规范的节点函数必须是纯函数接收state作为输入返回更新后的state字典不修改外部变量不产生副作用。同时必须在节点中做好异常捕获避免单个节点失败导致整个工作流崩溃。以下是一个带异常捕获的周报生成节点可直接运行from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage import os from dotenv import load_dotenv load_dotenv() llm ChatOpenAI(modelgpt-4o-mini, temperature0.3) def report_generator_node(state: WeeklyReportState): 内容生成节点根据原始数据生成周报初稿 try: prompt f 你是一位资深技术主管请根据以下原始工作数据生成周报。 要求使用【本周重点】、【量化成果】、【下周计划】三个模块必须包含具体数字。 原始数据{state[raw_data]} response llm.invoke([HumanMessage(contentprompt)]) print(f[生成节点] 已生成周报初稿字数{len(response.content)}) return {draft: response.content} except Exception as e: print(f[生成节点] 执行失败{str(e)}) return {feedback: f生成失败{str(e)}, retry_count: state[retry_count] 1}避坑提示节点返回值必须是字典即使只更新一个字段也要返回{field: value}而不是直接返回value。否则LangGraph无法识别更新内容导致状态丢失。同时异常捕获后必须返回包含feedback和retry_count的字典否则Agent无法感知到生成失败也不会触发重试逻辑。路由设计条件边与循环保护边决定了数据在节点之间的流动方向。普通边用于固定流程条件边用于动态路由。很多新手在写条件边时容易犯两个错误一是路由函数返回值与映射字典不匹配二是没有设置循环保护导致Agent陷入死循环。以下是一个带循环保护的条件路由函数可直接运行from langgraph.graph import StateGraph, END def should_continue(state: WeeklyReportState): 条件路由函数决定下一步走向 # 检查是否有反馈不合格且未超过最大重试次数 if state[feedback] and state[retry_count] 3: return regenerate else: return end # 构建状态图 workflow StateGraph(WeeklyReportState) workflow.add_node(generator, report_generator_node) workflow.add_node(checker, quality_checker_node) workflow.add_edge(generator, checker) workflow.add_conditional_edges( checker, should_continue, { regenerate: generator, # 不合格且未超重试次数 → 退回生成器 end: END # 合格或超重试次数 → 结束 } )避坑提示条件边的映射字典key必须与路由函数返回值完全一致比如路由函数返回regenerate映射字典就不能写retry否则会直接报错KeyError。同时一定要在路由函数中检查retry_count避免死循环。建议将最大重试次数设置为3既给Agent足够的修正机会又避免无效消耗。高级特性工具调用与人工中断当Agent需要调用外部工具如查询数据库、调用API时可以使用ToolNode自动接管工具调用。同时在关键节点插入人工中断让Agent在重要决策前等待人类确认。以下是一个带工具调用和人工中断的完整示例可直接运行from langgraph.prebuilt import ToolNode from langgraph.graph import interrupt from langchain_core.tools import tool # 定义工具 tool def query_git_log(branch: str) - str: 查询指定分支的Git提交记录 return f分支{branch}本周提交记录修复登录Bug、完成单元测试、参与需求评审 tools [query_git_log] tool_node ToolNode(tools) # 人工审批节点 def human_approval_node(state: WeeklyReportState): 人工审批节点等待用户确认周报内容 print(f[人工审批] 当前周报草稿\n{state[draft]}) # 暂停工作流等待用户输入 user_input interrupt(请确认周报内容是否合格输入yes或no) if user_input.lower() yes: return {final_report: state[draft]} else: return {feedback: 用户审核未通过请重新生成, retry_count: state[retry_count] 1}避坑提示interrupt()会暂停整个工作流直到用户输入才会继续。在生产环境中建议将中断结果持久化到数据库避免服务重启后丢失审批状态。同时工具节点的返回值必须与ToolMessage格式一致否则LLM无法正确解析工具结果。ToolNode会自动处理工具调用和结果返回无需手动构造ToolMessage。实战编码的核心价值掌握状态定义、节点开发、路由设计这三个核心范式你就已经具备了编写规范LangGraph代码的能力。高级特性则是在此基础上的延伸帮助你构建更复杂、更稳定的Agent。本文的完整代码已整理为单文件可直接在上一篇文章搭建的环境中运行。如果你在编码过程中遇到任何问题欢迎在评论区留言。下一篇《LangGraph生产维护》将带你深入稳定性、可观测性与运维实战让Agent真正敢上线、能落地。