必须跳过LangGraph预构建图API，直接用底层图结构定义节点、边与状态更新逻辑，以实现生产级Agent的稳定性、防死循环、人工审核和跨请求状态保持。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜
要让Agent在生产环境中稳定执行多步骤任务、防止无限循环、支持人工审核介入、跨请求保持状态，必须跳过LangGraph预构建图API，直接用底层图结构定义节点、边与状态更新逻辑。
初始化项目与环境隔离
打开终端，创建专用虚拟环境并激活：
conda create -n lg-enterprise python=3.12 → conda activate lg-enterprise
安装核心依赖：
pip install langgraph langchain-openai python-dotenv redis psycopg2-binary
【必须使用Python 3.12+】 LangGraph 0.2.x 版本对异步状态持久化和Checkpointer线程安全有严格版本约束，低版本会导致跨会话状态覆盖。
定义带防死循环机制的全局状态
新建 state.py，声明可被所有节点读写的共享状态：
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
tool_calls: list
iteration_count: int
requires_human_review: bool
final_answer: str
这里 iteration_count 字段不是装饰性字段——它将被每个节点主动递增，并在分支边中作为终止条件判断依据。不设此字段，ReAct循环可能因LLM幻觉陷入无限调用工具。
构建带条件分支与人工干预出口的图结构
第一步：实例化图对象
from langgraph.graph import StateGraph
workflow = StateGraph(AgentState)
第二步：注册四个核心节点
workflow.add_node("entry", entry_node)
workflow.add_node("plan", plan_node)
workflow.add_node("execute_tool", execute_tool_node)
workflow.add_node("human_review", human_review_node)
第三步：添加带 guard 的条件边
workflow.add_conditional_edges(
"plan",
lambda state: "tool" if state["tool_calls"] else ("review" if state["requires_human_review"] else "end"),
{"tool": "execute_tool", "review": "human_review", "end": END}
)
注意：lambda 返回字符串必须与 add_node 中注册的节点名完全一致，大小写敏感。填错一个字母，图构建时不会报错，但运行时直接卡死在该节点不往下走。
第四步：设置入口与出口
workflow.set_entry_point("entry")
workflow.set_finish_point("human_review")
注入MongoDB持久化检查点
方法一：使用官方MongoDB Checkpointer（推荐）
from langgraph.checkpoint.mongodb import AsyncMongoDBSaver
import asyncio
async def init_checkpointer():
client = AsyncMongoDBSaver.from_conn_string("mongodb://localhost:27017/")
await client.setup() # 必须显式调用，否则首次运行会静默失败
return client
方法二：降级使用Redis（开发阶段快速验证）
from langgraph.checkpoint.redis import AsyncRedisSaver
saver = AsyncRedisSaver.from_url("redis://localhost:6379/0")
将检查点传入图编译：
app = workflow.compile(checkpointer=saver)
这一步完成后，每次 invoke 调用都会自动保存 state 到数据库；中断后传入相同 thread_id 即可从断点恢复，无需手写状态序列化逻辑。
配置流式输出与工具调用拦截
启用流式响应只需在 invoke 时加参数：
async for chunk in app.astream({"messages": [{"role": "user", "content": "查上季度销售额"}]}, config={"thread_id": "abc123"}):
print(chunk)
如需在工具调用前插入人工审批钩子，修改 execute_tool_node 函数，在实际调用前插入：
if state.get("requires_human_review"):
return {"requires_human_review": True} # 触发跳转到 human_review 节点
不要在工具函数内部做权限判断——状态变更必须通过节点返回字典完成，否则 Checkpointer 无法捕获变更，后续恢复时会丢失该标记。
