LangGraph 学习笔记

LangGraph 总览

LangGraph 是一个基于图（Graph）的 LLM 应用编排框架，由 LangChain 团队开发。它将 LLM 应用的执行流程建模为有向图，通过节点（Node）、边（Edge）和状态（State）的组合，实现复杂的 Agent 工作流。

属性	说明
官网	https://langchain-ai.github.io/langgraph
GitHub	https://github.com/langchain-ai/langgraph
协议	MIT
语言	Python / JavaScript
当前版本	v1.x（2025 年，已集成到 langchain 主包）
核心概念	StateGraph / Node / Edge / Conditional Edge / Checkpointer
典型场景	ReAct Agent / 多智能体编排 / 人工审批 / 复杂工作流

LangGraph 在 AI 应用中的定位

LangGraph vs 其他 Agent 框架

维度	LangGraph	LangChain AgentExecutor	AutoGen	CrewAI
执行模型	有向图（Graph）	顺序循环	对话式	角色扮演
可控性	⭐⭐⭐⭐⭐	⭐⭐	⭐⭐⭐	⭐⭐⭐
状态管理	显式 State	隐式	对话历史	任务状态
条件路由	✅ 原生支持	❌	有限	有限
人工介入	✅ 中断/审批	❌	❌	❌
持久化	✅ Checkpointer	❌	有限	有限
多 Agent	✅ Supervisor/Swarm	❌	✅	✅
调试体验	⭐⭐⭐⭐⭐	⭐⭐	⭐⭐⭐	⭐⭐⭐
学习曲线	中等	低	中等	低

选型建议：

• 需要精确控制 Agent 执行流程 → LangGraph（节点级可控 + 条件路由 + 持久化）
• 快速原型、简单 Agent → LangChain AgentExecutor / create_agent
• 多 Agent 协作、角色扮演 → CrewAI / AutoGen
• 生产级复杂工作流 → LangGraph（人工审批、错误恢复、多步骤编排）

---

一、本章概览

本模块通过五个递进式示例，覆盖 LangGraph 的核心功能：

文件	内容	状态
langgraph_test.py	Part 1~4：基础流程、ReAct、记忆、手动构建	✅ 已验证
human_approval_test.py	Part 5：Agent 智能审批 + 记忆回退	✅ 已验证

---

二、LangGraph 核心概念

2.1 State（状态）

State 是在节点之间传递的数据容器，定义了图中所有节点共享的数据结构：

from typing_extensions import TypedDict

class MyState(TypedDict):
    input_text: str       # 用户输入
    result: str           # 处理结果
    count: int            # 计数器

LangGraph 还提供了预定义的 MessagesState，专门用于 Agent 对话场景：

from langgraph.graph import MessagesState

# MessagesState 内部定义：
# class MessagesState(TypedDict):
#     messages: Annotated[list, add_messages]
#
# add_messages 是一个 reducer：新消息追加到列表，而不是覆盖

2.2 Node（节点）

节点是执行具体逻辑的函数，接收 State 并返回更新：

def my_node(state: MyState) -> dict:
    """节点函数：处理 state 并返回更新"""
    text = state["input_text"]
    result = text.upper()
    return {"result": result}  # 只返回需要更新的字段

2.3 Edge（边）

边定义节点之间的执行顺序：

from langgraph.graph import StateGraph, START, END

builder = StateGraph(MyState)
builder.add_node("step1", step1_fn)
builder.add_node("step2", step2_fn)

builder.add_edge(START, "step1")      # START → step1
builder.add_edge("step1", "step2")    # step1 → step2
builder.add_edge("step2", END)        # step2 → END

2.4 Conditional Edge（条件边）

条件边根据运行时状态动态选择下一个节点：

def router(state: MyState) -> Literal["path_a", "path_b"]:
    if state["count"] > 10:
        return "path_a"
    return "path_b"

builder.add_conditional_edges(
    "step1",           # 从哪个节点出发
    router,            # 路由函数
    {                  # 路由映射
        "path_a": "node_a",
        "path_b": "node_b",
    },
)

2.5 Checkpointer（状态持久化）

Checkpointer 负责将 Agent 的状态（对话历史、中间步骤）持久化，实现多轮对话记忆：

Checkpointer	存储位置	适用场景
MemorySaver	内存	开发调试，重启丢失
SqliteSaver	SQLite 文件	单机持久化
PostgresSaver	PostgreSQL	生产环境
RedisSaver	Redis Stack	分布式场景
MySQLSaver	MySQL	已有 MySQL 基础设施

使用方式：

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

# 相同 thread_id 的调用共享同一份记忆
config = {"configurable": {"thread_id": "session-001"}}
graph.invoke({"messages": [...]}, config=config)

2.6 ToolNode 与 bind_tools

LangGraph 中的工具调用机制：

LLM.bind_tools(tools)  →  告诉 LLM 有哪些工具可用
                           LLM 在需要时生成 tool_calls

ToolNode(tools)        →  自动执行工具调用
                           从 AIMessage 中提取 tool_calls
                           调用对应工具函数
                           将结果封装为 ToolMessage

---

三、Part 1：基础节点与边

3.1 实现概述

Part 1 实现一个文本处理流水线，演示 StateGraph 的核心概念：

3.2 State 定义

class TextProcessingState(TypedDict):
    """文本处理流水线的状态"""
    input_text: str           # 用户输入的原始文本
    processed_text: str       # 经过处理的文本
    analysis: str             # 统计分析结果
    final_output: str         # 最终输出

3.3 节点函数

def node_uppercase(state: TextProcessingState) -> dict:
    """节点1：将输入文本转为大写"""
    text = state["input_text"]
    result = text.upper()
    return {"processed_text": result}

def node_analyze(state: TextProcessingState) -> dict:
    """节点2：分析文本的长度和字符构成"""
    text = state["processed_text"]
    total_chars = len(text)
    letter_count = sum(1 for c in text if c.isalpha())
    # ... 统计分析
    return {"analysis": analysis}

3.4 条件路由

def router_by_length(state: TextProcessingState) -> Literal["summarize", "direct_output"]:
    """条件边：根据处理后的文本长度路由到不同节点"""
    length = len(state["processed_text"])
    if length <= 50:
        return "summarize"      # 短文本 → 生成详细摘要
    else:
        return "direct_output"  # 长文本 → 直接输出

# 注册条件边
builder.add_conditional_edges("analyze", router_by_length, {
    "summarize": "summarize",
    "direct_output": "direct_output",
})

3.5 编译与执行

builder = StateGraph(TextProcessingState)
builder.add_node("uppercase", node_uppercase)
builder.add_node("analyze", node_analyze)
builder.add_node("summarize", node_summarize)
builder.add_node("direct_output", node_direct_output)

builder.add_edge(START, "uppercase")
builder.add_edge("uppercase", "analyze")
builder.add_conditional_edges("analyze", router_by_length, {...})
builder.add_edge("summarize", END)
builder.add_edge("direct_output", END)

graph = builder.compile()
result = graph.invoke({"input_text": "Hello LangGraph!"})

---

四、Part 2/3：ReAct Agent（高级 API）

4.1 create_agent 内部图结构

langchain.agents.create_agent 内部自动构建了以下 LangGraph 图：

START
  │
  ▼
┌──────────┐
│  agent   │  ← LLM 思考 + 决定是否调用工具
└────┬─────┘
     │
┌────▼────────────┐
│ should_continue  │  ← 条件边：有 tool_calls 吗？
└──┬──────────┬────┘
 有│          │无
┌──▼────┐  ┌──▼────┐
│ tools │  │  END  │
└──┬────┘  └───────┘
   │
   └──────→ 回到 agent（继续思考）

4.2 内部实现细节

组件	职责	实现方式
agent 节点	调用 LLM，决定是否调用工具	llm.bind_tools(tools) → llm.invoke(messages)
should_continue	条件路由	检查 last_message.tool_calls 是否为空
tools 节点	执行工具调用	ToolNode(tools) 自动处理

4.3 工具定义

工具使用 @tool 装饰器定义，args_schema 必须使用 Pydantic BaseModel：

from pydantic import BaseModel, Field

class CalcInput(BaseModel):
    a: int = Field(description="第一个数字")
    b: int = Field(description="第二个数字")

@tool(args_schema=CalcInput)
def calculator(a: int, b: int) -> str:
    """计算两个整数的加减乘除结果"""
    return json.dumps({
        "加法": a + b,
        "减法": a - b,
        "乘法": a * b,
        "除法": a / b if b != 0 else "除数不能为零",
    }, ensure_ascii=False)

4.4 强制工具调用的 system_prompt

ReAct Agent 中 LLM 可能自行计算而不调用工具，通过强 system_prompt 约束：

system_prompt = (
    "你是一个严格使用工具的助手。你绝对禁止自行计算或编造数据。\n"
    "\n"
    "强制规则（违反将导致严重后果）：\n"
    "- 任何数学运算 → 必须先调用 calculator 工具\n"
    "- 任何时间/日期查询 → 必须先调用 current_time 工具\n"
    "- 任何代码执行 → 必须先调用 python_repl 工具\n"
    "\n"
    "记住：永远先调用工具，再回答。不要直接给出答案。\n"
)
agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt=system_prompt,
)

4.5 MemorySaver 多轮对话

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
agent = create_agent(
    model=llm,
    tools=tools,
    checkpointer=memory,    # ← 接入 MemorySaver
    system_prompt=system_prompt,
)

# 相同 thread_id → 共享记忆
config = {"configurable": {"thread_id": "session-001"}}
agent.invoke({"messages": [("user", "我叫小明")]}, config=config)
agent.invoke({"messages": [("user", "我叫什么？")]}, config=config)
# → Agent 能回答出 "小明"

# 不同 thread_id → 隔离会话
new_config = {"configurable": {"thread_id": "session-002"}}
agent.invoke({"messages": [("user", "我叫什么？")]}, config=new_config)
# → Agent 不知道你是谁

4.6 调试：stream_mode 选择

stream_mode	输出内容	适用场景
"values"	每次节点执行后的完整 State	查看最终结果
"updates"	每个节点的增量更新	定位卡在哪一步
"messages"	每条消息的生成过程	实时流式输出
"debug"	详细的节点进出日志	深度调试

# 推荐调试方式：stream_mode="updates"
for event in agent.stream({"messages": [...]}, stream_mode="updates"):
    for node_name, update in event.items():
        print(f"[{node_name}] {update}")  # 实时看到每个节点的输出

---

五、Part 4：手动构建 ReAct Agent

5.1 为什么需要手动构建？

场景	create_agent	手动 StateGraph
标准 ReAct Agent	✅ 一行代码	❌ 代码冗余
插入日志节点	❌ 不支持	✅ 自定义节点
人工审批/中断	❌ 不支持	✅ interrupt_before
限制最大循环次数	❌ 不支持	✅ 自定义 should_continue
工具执行失败降级	❌ 内置基础	✅ 自定义错误处理
多工具组/子图	❌ 不支持	✅ 嵌套 StateGraph

5.2 手动构建步骤

Step 1：定义 State

from langgraph.graph import MessagesState
# MessagesState 已预定义 messages 字段和 add_messages reducer
# 等价于：
# class AgentState(TypedDict):
#     messages: Annotated[list, add_messages]

Step 2：准备 LLM + 工具绑定

llm_with_tools = llm.bind_tools(tools)  # 让 LLM 感知可用工具
system_prompt = SystemMessage(content="你是一个严格使用工具的助手...")

Step 3：定义节点函数

def agent_node(state: MessagesState) -> dict:
    """Agent 节点：调用 LLM 决定是直接回答还是调用工具"""
    messages = [system_prompt] + state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

# tools 节点使用预构建的 ToolNode
tools_node = ToolNode(tools)

Step 4：定义条件路由

def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
    """检查 LLM 是否要调用工具"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"      # 有工具调用 → 执行工具
    return "__end__"        # 无工具调用 → Agent 已给出最终回答

Step 5：构建 StateGraph

builder = StateGraph(MessagesState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tools_node)

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")    # 工具执行后回到 agent 继续思考

Step 6：编译并接入 MemorySaver

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

5.3 手动构建的图结构

5.4 自定义扩展示例

手动构建的核心优势是可以插入自定义节点：

# 示例：添加日志节点
def log_node(state: MessagesState) -> dict:
    """记录每次工具调用的日志"""
    last = state["messages"][-1]
    if isinstance(last, ToolMessage):
        print(f"[LOG] 工具 {last.name} 返回: {last.content[:50]}...")
    return {}

# 示例：限制最大循环次数
MAX_ITERATIONS = 5
iteration_count = 0

def should_continue_with_limit(state: MessagesState) -> Literal["tools", "__end__"]:
    global iteration_count
    iteration_count += 1
    if iteration_count > MAX_ITERATIONS:
        print(f"[WARN] 达到最大循环次数 {MAX_ITERATIONS}，强制结束")
        return "__end__"
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return "__end__"

---

六、create_agent vs 手动构建对比

维度	create_agent（Part 2/3）	手动 StateGraph（Part 4）
代码量	1 行 create_agent(...)	~30 行手动定义
可控性	低（黑盒）	高（每个节点可定制）
自定义节点	❌ 不支持	✅ 日志/审批/重试
错误处理	内置基础处理	可自定义降级策略
条件路由	固定（tool_calls 判断）	可自定义任意逻辑
循环控制	不支持限制次数	可自定义 should_continue
理解原理	不需要	需要理解图运行机制
system_prompt	通过参数传入	在 agent_node 中手动拼接
记忆	通过 checkpointer 参数	编译时传入 checkpointer
调试	debug=True	debug=True + 自定义日志
适用场景	标准 ReAct	复杂工作流

结论：

• 快速原型、标准 Agent → create_agent
• 生产系统、需要定制 → 手动 StateGraph
• 两者底层都是 LangGraph，图结构完全一致

---

七、开发踩坑记录

7.1 args_schema 必须使用 Pydantic BaseModel

问题： @tool(args_schema=CalcInput) 报 TypeError: args_schema must be a subclass of pydantic BaseModel。

原因： CalcInput 继承了 TypedDict，而 LangChain 要求 args_schema 必须是 pydantic.BaseModel 的子类。

解决：

# ❌ 错误
class CalcInput(TypedDict):
    a: int
    b: int

# ✅ 正确
from pydantic import BaseModel, Field
class CalcInput(BaseModel):
    a: int = Field(description="第一个数字")
    b: int = Field(description="第二个数字")

7.2 create_react_agent 已废弃

问题： from langgraph.prebuilt import create_react_agent 报 LangGraphDeprecatedSinceV10 警告。

解决： 迁移到 from langchain.agents import create_agent（注意函数名也变了）。

# ❌ 已废弃
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(model=llm, tools=tools)

# ✅ 新方式
from langchain.agents import create_agent
agent = create_agent(model=llm, tools=tools)

7.3 LLM 自行计算不调用工具

问题： 数学计算问题中，LLM 自己算出结果而不调用 calculator 工具。

原因： 大模型在训练时见过大量数学题，倾向于"自己算"。这是 ReAct Agent 的固有限制。

解决（三重加固）：

1. system_prompt 强制要求调用工具
2. 工具描述 中明确标注"必须使用此工具"
3. 用户问题 中显式提及工具名

# 1. system_prompt
system_prompt = "任何数学运算 → 必须先调用 calculator 工具..."

# 2. 工具描述
@tool(args_schema=CalcInput)
def calculator(a: int, b: int) -> str:
    """计算两个整数的加减乘除结果。当用户问数学计算问题时，必须使用此工具，绝不能自己计算。"""

# 3. 用户问题
messages = [("user", "请使用 calculator 工具计算 987654321 和 123456789 的乘积")]

7.4 Agent 卡住无输出

问题： Agent 长时间没有输出，不知道卡在哪一步。

解决： 使用 stream_mode="updates" 查看每个节点的实时执行状态：

for event in agent.stream({"messages": [...]}, stream_mode="updates"):
    for node_name, update in event.items():
        print(f"[{node_name}] {update}")

节点说明：

• [agent] → 等待 LLM 响应（网络延迟或模型计算）
• [tools] → 执行工具调用（工具执行中）

7.5 deepseek 模型工具调用不稳定

问题： deepseek-v4-pro 在工具调用方面不如 OpenAI GPT-4 稳定，有时忽略 system_prompt 中的强制规则。

缓解方案：

• 使用更强的 system_prompt（明确禁止自行计算）
• 在用户问题中显式提及工具名
• 考虑切换到工具调用更稳定的模型（如 GPT-4o、Claude）

---

八、关键 API 速查

# ── 基础图构建 ──
from langgraph.graph import StateGraph, START, END, MessagesState

builder = StateGraph(MyState)           # 创建图构建器
builder.add_node("name", fn)            # 注册节点
builder.add_edge(START, "name")         # 普通边
builder.add_conditional_edges(          # 条件边
    "from_node", router_fn, {"a": "node_a", "b": "node_b"}
)
graph = builder.compile()               # 编译图
graph.invoke({"key": "value"})          # 同步执行

# ── 高级 API ──
from langchain.agents import create_agent

agent = create_agent(
    model=llm,                          # LLM 实例
    tools=tools,                        # 工具列表
    system_prompt="...",                # 系统提示词
    checkpointer=memory,                # 状态持久化
    debug=True,                         # 调试模式
)

# ── 手动 ReAct Agent ──
from langgraph.prebuilt import ToolNode

llm_with_tools = llm.bind_tools(tools)  # 绑定工具
tools_node = ToolNode(tools)            # 工具节点

# ── 记忆 ──
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "session-001"}}

# ── 调试 ──
for event in graph.stream(input, stream_mode="updates"):
    for node, update in event.items():
        print(f"[{node}] {update}")

---

附录：完整源码

A.1 langgraph_test.py

"""
LangGraph 功能测试

覆盖四个核心场景：
    Part 1 - 基础节点与边（StateGraph + Node + Edge + Conditional Edge）
    Part 2 - ReAct Agent 流程（create_agent 高级 API + 自定义工具）
    Part 3 - ReAct Agent 带记忆（create_agent + MemorySaver + 多轮对话）
    Part 4 - 手动构建 ReAct Agent（StateGraph 手动定义节点/边 + MemorySaver）

Part 2/3 vs Part 4 对比：
    - Part 2/3 使用 create_agent() 高级 API：一行代码创建完整 ReAct Agent
      内部自动构建了 agent→tools→agent 的循环图，适合标准场景快速上手
    - Part 4 使用 StateGraph 手动构建：完全控制节点、边、条件路由
      可以插入自定义节点（如日志、审批、人工介入），适合需要定制的复杂场景

运行方式：
    python -m app.langgraph_test.langgraph_test
"""

import sys
import os

_project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
if _project_root not in sys.path:
    sys.path.insert(0, _project_root)

import datetime
import json
from typing import Literal

from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import AIMessage
from typing_extensions import TypedDict

from app.common import llm


# 全局调试开关（默认关闭，取消注释可查看详细执行日志）
# import logging
# logging.basicConfig(level=logging.INFO)

print(f"[INFO] 当前使用模型: {llm.LLM_MODEL_NAME}")
print(f"[INFO] 当前 Base URL: {llm.LLM_BASE_URL}")
print()


# ====================================================================
# Part 1: 基础节点与边
# ====================================================================
# StateGraph 是 LangGraph 的核心概念：
#   - State（状态）：在节点之间传递的数据容器
#   - Node（节点）：接收 State、处理、返回更新的函数
#   - Edge（边）：定义节点间的执行顺序
#   - Conditional Edge（条件边）：根据运行时条件选择下一步
#
# 本示例实现一个文本处理流水线：
#   输入文本 → 大写转换 → 统计分析 → 条件路由 → 输出
# ====================================================================

# 定义 State 类型
class TextProcessingState(TypedDict):
    """文本处理流水线的状态"""
    input_text: str           # 用户输入的原始文本
    processed_text: str       # 经过处理的文本
    analysis: str             # 统计分析结果
    final_output: str         # 最终输出


def node_uppercase(state: TextProcessingState) -> dict:
    """节点1：将输入文本转为大写"""
    text = state["input_text"]
    result = text.upper()
    print(f"  [node_uppercase] '{text[:20]}...' → '{result[:20]}...'")
    return {"processed_text": result}


def node_analyze(state: TextProcessingState) -> dict:
    """节点2：分析文本的长度和字符构成"""
    text = state["processed_text"]
    total_chars = len(text)
    letter_count = sum(1 for c in text if c.isalpha())
    digit_count = sum(1 for c in text if c.isdigit())
    space_count = sum(1 for c in text if c.isspace())
    other_count = total_chars - letter_count - digit_count - space_count

    analysis = (
        f"总长度: {total_chars} 字符 | "
        f"字母: {letter_count} | "
        f"数字: {digit_count} | "
        f"空格: {space_count} | "
        f"其他: {other_count}"
    )
    print(f"  [node_analyze] {analysis}")
    return {"analysis": analysis}


def node_summarize(state: TextProcessingState) -> dict:
    """节点3（短文本分支）：生成详细摘要"""
    text = state["processed_text"]
    analysis = state["analysis"]
    summary = (
        f"📝 文本摘要\n"
        f"  原文: {state['input_text']}\n"
        f"  大写: {text}\n"
        f"  统计: {analysis}\n"
        f"  说明: 这是一条短文本，已完成完整处理。"
    )
    print(f"  [node_summarize] 短文本 → 生成详细摘要")
    return {"final_output": summary}


def node_direct_output(state: TextProcessingState) -> dict:
    """节点4（长文本分支）：直接输出结果"""
    text = state["processed_text"]
    analysis = state["analysis"]
    output = (
        f"📄 处理结果\n"
        f"  大写: {text[:80]}...\n"
        f"  统计: {analysis}\n"
        f"  说明: 文本较长，仅展示处理摘要。"
    )
    print(f"  [node_direct_output] 长文本 → 直接输出")
    return {"final_output": output}


def router_by_length(state: TextProcessingState) -> Literal["summarize", "direct_output"]:
    """条件边：根据处理后的文本长度路由到不同节点"""
    length = len(state["processed_text"])
    if length <= 50:
        print(f"  [条件路由] 长度={length} → 走 summarize 分支")
        return "summarize"
    else:
        print(f"  [条件路由] 长度={length} → 走 direct_output 分支")
        return "direct_output"


def build_basic_graph():
    """构建基础节点-边图"""

    # 1. 创建图构建器，传入 State 类型
    builder = StateGraph(TextProcessingState)

    # 2. 注册节点
    builder.add_node("uppercase", node_uppercase)
    builder.add_node("analyze", node_analyze)
    builder.add_node("summarize", node_summarize)
    builder.add_node("direct_output", node_direct_output)

    # 3. 定义边的执行顺序
    builder.add_edge(START, "uppercase")    # Start → uppercase
    builder.add_edge("uppercase", "analyze")  # uppercase → analyze
    builder.add_conditional_edges(
        "analyze",                           # 从 analyze 节点出发
        router_by_length,                    # 条件判断函数
        {                                    # 路由映射
            "summarize": "summarize",
            "direct_output": "direct_output",
        },
    )
    builder.add_edge("summarize", END)       # summarize → End
    builder.add_edge("direct_output", END)   # direct_output → End

    # 4. 编译图
    graph = builder.compile()
    return graph


def part1_basic_graph():
    """Part 1 测试：基础节点与边"""

    print(f"\n{'=' * 60}")
    print("Part 1: 基础节点与边")
    print("  演示 StateGraph 的核心概念：")
    print("  - 节点（Node）：state → state 的转换函数")
    print("  - 边（Edge）：定义节点间的执行顺序")
    print("  - 条件边（Conditional Edge）：运行时动态路由")
    print(f"{'=' * 60}")

    graph = build_basic_graph()

    # 打印图的 Mermaid 结构
    print("\n[图结构]")
    print(graph.get_graph().draw_mermaid())

    # 测试用例 1：短文本
    print("\n" + "-" * 40)
    print("测试 1: 短文本")
    print("-" * 40)
    result1 = graph.invoke({"input_text": "Hello LangGraph!"})
    print(f"\n[输出]\n{result1['final_output']}")

    # 测试用例 2：长文本
    print("\n" + "-" * 40)
    print("测试 2: 长文本")
    print("-" * 40)
    long_text = (
        "LangGraph is a library for building stateful, multi-agent applications. "
        "It extends LangChain's capabilities by providing a graph-based execution model "
        "where nodes represent computation steps and edges define the flow of data."
    )
    result2 = graph.invoke({"input_text": long_text})
    print(f"\n[输出]\n{result2['final_output']}")

    print("\n✓ Part 1 完成\n")
    return graph


# ====================================================================
# Part 2: ReAct Agent 流程
# ====================================================================
# ReAct = Reasoning + Acting
#   create_agent 是 LangChain 提供的预构建 Agent：
#   - 自动处理"思考→行动→观察→思考→...→最终答案"的循环
#   - 内置 ToolNode，自动调度工具调用
#   - 支持自定义系统提示词和工具列表
# ====================================================================

class CalcInput(BaseModel):
    a: int = Field(description="第一个数字")
    b: int = Field(description="第二个数字")


@tool(args_schema=CalcInput)
def calculator(a: int, b: int) -> str:
    """计算两个整数的加减乘除结果。当用户问数学计算问题时，必须使用此工具，绝不能自己计算。"""
    return json.dumps({
        "加法": a + b,
        "减法": a - b,
        "乘法": a * b,
        "除法": a / b if b != 0 else "除数不能为零",
    }, ensure_ascii=False)


@tool
def python_repl(code: str) -> str:
    """执行 Python 代码并返回输出。入参是纯 Python 代码，不要包裹在 ```python 标记中。"""
    import io
    import contextlib
    output = io.StringIO()
    try:
        with contextlib.redirect_stdout(output):
            exec(code, {"__builtins__": __builtins__})
        return output.getvalue() or "(无输出)"
    except Exception as e:
        return f"执行错误: {e}"


@tool
def current_time(format_str: str = "%Y-%m-%d %H:%M:%S") -> str:
    """获取当前的日期和时间。当用户问当前时间、日期、年月日时分秒时，必须使用此工具，绝不能编造时间。format_str 是 strftime 格式字符串。"""
    return datetime.datetime.now().strftime(format_str)


react_tools = [calculator, python_repl, current_time]


def _stream_agent(agent, messages: list):
    """辅助函数：流式执行 Agent 并打印每个节点的执行状态

    使用 stream_mode="updates" 可以看到每个节点（agent/tools）的实时输出，
    方便定位 Agent 卡在哪一步。
    """
    for event in agent.stream(
        {"messages": messages},
        stream_mode="updates",
    ):
        for node_name, update in event.items():
            print(f"  [{node_name}] ", end="", flush=True)
            msgs = update.get("messages", [])
            for msg in msgs:
                if isinstance(msg, AIMessage) and msg.tool_calls:
                    for tc in msg.tool_calls:
                        print(f"→ 调用工具: {tc['name']}(args={tc['args']})")
                elif isinstance(msg, AIMessage) and msg.content:
                    content = msg.content
                    if len(content) > 200:
                        content = content[:200] + "..."
                    print(f"→ {content}")


def part2_react_agent():
    """Part 2 测试：ReAct Agent 流程（高级 API）"""

    print(f"\n{'=' * 60}")
    print("Part 2: ReAct Agent 流程")
    print("  演示 create_agent 的核心机制：")
    print("  - 思考（Thought）：分析当前问题和可用工具")
    print("  - 行动（Action）：调用一个工具获取信息")
    print("  - 观察（Observation）：工具返回的结果")
    print("  - 循环（Loop）：重复思考→行动→观察，直到得出答案")
    print(f"{'=' * 60}")

    system_prompt = (
        "你是一个严格使用工具的助手。你绝对禁止自行计算或编造数据。\n"
        "\n"
        "强制规则（违反将导致严重后果）：\n"
        "- 任何数学运算（加减乘除、取余等）→ 必须先调用 calculator 工具\n"
        "- 任何时间/日期查询 → 必须先调用 current_time 工具\n"
        "- 任何代码执行 → 必须先调用 python_repl 工具\n"
        "\n"
        "执行流程：\n"
        "1. 收到问题后，第一步就是调用对应的工具\n"
        "2. 等工具返回结果后，再根据工具的结果回答用户\n"
        "3. 如果不确定结果，调用工具验证\n"
        "\n"
        "记住：永远先调用工具，再回答。不要直接给出答案。\n"
    )
    agent = create_agent(
        model=llm.llm,
        tools=react_tools,
        system_prompt=system_prompt,
        debug=False,
    )

    # 问题 1：数学计算
    print("\n" + "-" * 40)
    print("问题 1: 数学计算")
    print("-" * 34)
    messages1 = [("user", "请使用 calculator 工具计算 987654321 和 123456789 的乘积")]
    _stream_agent(agent, messages1)

    # 问题 2：当前时间
    print("\n" + "-" * 40)
    print("问题 2: 当前时间")
    print("-" * 34)
    messages2 = [("user", "现在是什么时间？请告诉我年月日时分秒")]
    _stream_agent(agent, messages2)

    # 问题 3：Python 代码执行
    print("\n" + "-" * 40)
    print("问题 3: Python 代码执行")
    print("-" * 34)
    messages3 = [("user", "用 Python 生成斐波那契数列的前 20 个数字")]
    _stream_agent(agent, messages3)

    print("\n✓ Part 2 完成\n")
    return agent


# ====================================================================
# Part 3: ReAct Agent 带记忆
# ====================================================================

def part3_react_agent_with_memory():
    """Part 3 测试：ReAct Agent 带记忆"""

    print(f"\n{'=' * 60}")
    print("Part 3: ReAct Agent 带记忆")
    print("  演示带 MemorySaver 的多轮对话：")
    print("  - 相同 thread_id → Agent 记住上下文")
    print("  - 不同 thread_id → 隔离的会话")
    print("  - 记忆与工具调用的结合")
    print(f"{'=' * 60}")

    memory = MemorySaver()

    system_prompt = (
        "你是一个严格使用工具的助手。你绝对禁止自行计算或编造数据。\n"
        "- 任何数学运算 → 必须先调用 calculator 工具\n"
        "- 任何时间/日期查询 → 必须先调用 current_time 工具\n"
        "- 任何代码执行 → 必须先调用 python_repl 工具\n"
        "永远先调用工具，再回答。不要直接给出答案。\n"
    )
    agent = create_agent(
        model=llm.llm,
        tools=react_tools,
        checkpointer=memory,
        system_prompt=system_prompt,
    )

    thread1 = RunnableConfig(configurable={"thread_id": "react-session-001"})

    print("\n" + "-" * 40)
    print("【会话 1 第 1 轮】告诉名字")
    print("-" * 34)
    res1 = agent.invoke(
        {"messages": [("user", "你好，我叫小明，我是个数学爱好者")]},
        config=thread1,
    )
    print(f"Assistant: {res1['messages'][-1].content}")

    print("\n" + "-" * 40)
    print("【会话 1 第 2 轮】让 Agent 计算（测试记忆 + 工具调用结合）")
    print("-" * 34)
    res2 = agent.invoke(
        {"messages": [("user", "你还记得我是谁吗？另外请计算 123 * 456")]},
        config=thread1,
    )
    for msg in res2["messages"]:
        if isinstance(msg, AIMessage) and msg.tool_calls:
            for tc in msg.tool_calls:
                print(f"  [调用工具] {tc['name']}(args={tc['args']})")
    print(f"Assistant: {res2['messages'][-1].content}")

    print("\n" + "-" * 40)
    print("【会话 1 第 3 轮】继续对话，验证上下文保持")
    print("-" * 34)
    res3 = agent.invoke(
        {"messages": [("user", "我刚才问你的是什么计算题？我的名字是什么？")]},
        config=thread1,
    )
    print(f"Assistant: {res3['messages'][-1].content}")

    thread2 = RunnableConfig(configurable={"thread_id": "react-session-002"})

    print("\n" + "-" * 40)
    print("【会话 2 新对话】不同 thread_id，无记忆")
    print("-" * 34)
    res4 = agent.invoke(
        {"messages": [("user", "我们之前聊过什么？我叫什么名字？")]},
        config=thread2,
    )
    print(f"Assistant: {res4['messages'][-1].content}")

    print("\n✓ Part 3 完成\n")
    return agent


# ====================================================================
# Part 4: 手动构建 ReAct Agent（StateGraph + 记忆）
# ====================================================================

from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage
from langgraph.prebuilt import ToolNode


def part4_manual_react_agent():
    """Part 4 测试：手动构建 ReAct Agent（StateGraph + MemorySaver）"""

    print(f"\n{'=' * 60}")
    print("Part 4: 手动构建 ReAct Agent（StateGraph + MemorySaver）")
    print("  演示如何用 StateGraph 从零搭建一个 ReAct Agent：")
    print("  - 手动定义 agent 节点（LLM + 工具绑定）")
    print("  - 手动定义 tools 节点（ToolNode 执行工具）")
    print("  - 手动定义条件边（should_continue 路由）")
    print("  - 接入 MemorySaver（多轮对话记忆）")
    print(f"{'=' * 60}")

    from langgraph.graph import MessagesState

    tools = react_tools
    llm_with_tools = llm.llm.bind_tools(tools)

    system_prompt = SystemMessage(content=(
        "你是一个严格使用工具的助手。你绝对禁止自行计算或编造数据。\n"
        "- 任何数学运算 → 必须先调用 calculator 工具\n"
        "- 任何时间/日期查询 → 必须先调用 current_time 工具\n"
        "- 任何代码执行 → 必须先调用 python_repl 工具\n"
        "永远先调用工具，再回答。不要直接给出答案。"
    ))

    def agent_node(state: MessagesState) -> dict:
        """Agent 节点：调用 LLM，让它决定是直接回答还是调用工具"""
        messages = [system_prompt] + state["messages"]
        response = llm_with_tools.invoke(messages)
        return {"messages": [response]}

    tools_node = ToolNode(tools)

    def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
        """条件边：检查 LLM 是否要调用工具"""
        last_message = state["messages"][-1]
        if last_message.tool_calls:
            return "tools"
        return "__end__"

    builder = StateGraph(MessagesState)
    builder.add_node("agent", agent_node)
    builder.add_node("tools", tools_node)
    builder.add_edge(START, "agent")
    builder.add_conditional_edges("agent", should_continue)
    builder.add_edge("tools", "agent")

    memory = MemorySaver()
    graph = builder.compile(checkpointer=memory)

    print("\n[手动构建的图结构]")
    print(graph.get_graph().draw_mermaid())

    config = RunnableConfig(configurable={"thread_id": "manual-react-001"})

    print("\n" + "-" * 40)
    print("【第 1 轮】自我介绍")
    print("-" * 34)
    res1 = graph.invoke(
        {"messages": [HumanMessage(content="你好，我叫小明")]},
        config=config,
    )
    print(f"Assistant: {res1['messages'][-1].content}")

    print("\n" + "-" * 40)
    print("【第 2 轮】工具调用 + 记忆验证")
    print("-" * 34)
    res2 = graph.invoke(
        {"messages": [HumanMessage(content="你还记得我叫什么吗？另外请用 calculator 计算 987654321 和 123456789 的乘积")]},
        config=config,
    )
    for msg in res2["messages"]:
        if isinstance(msg, AIMessage) and msg.tool_calls:
            for tc in msg.tool_calls:
                print(f"  [调用工具] {tc['name']}(args={tc['args']})")
        elif isinstance(msg, ToolMessage):
            print(f"  [工具返回] {msg.content[:80]}...")
    print(f"Assistant: {res2['messages'][-1].content}")

    print("\n" + "-" * 40)
    print("【第 3 轮】回忆完整上下文")
    print("-" * 34)
    res3 = graph.invoke(
        {"messages": [HumanMessage(content="回顾一下我们的对话：我叫什么？我让你算了什么？结果是多少？")]},
        config=config,
    )
    print(f"Assistant: {res3['messages'][-1].content}")

    print("\n" + "-" * 40)
    print("【新会话】不同 thread_id")
    print("-" * 34)
    new_config = RunnableConfig(configurable={"thread_id": "manual-react-002"})
    res4 = graph.invoke(
        {"messages": [HumanMessage(content="我叫什么名字？")]},
        config=new_config,
    )
    print(f"Assistant: {res4['messages'][-1].content}")

    print("\n✓ Part 4 完成\n")
    return graph


# ====================================================================
# 运行全部测试
# ====================================================================

if __name__ == "__main__":
    print(f"\n{'=' * 60}")
    print("LangGraph 功能测试")
    print(f"{'=' * 60}")
    print(f"  测试时间: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print()

    # Part 1: 基础节点与边
    # part1_basic_graph()

    # Part 2: ReAct Agent（高级 API）
    # part2_react_agent()

    # Part 3: ReAct Agent 带记忆（高级 API + MemorySaver）
    # part3_react_agent_with_memory()

    # Part 4: 手动构建 ReAct Agent（StateGraph + MemorySaver）
    part4_manual_react_agent()

    print(f"\n{'=' * 60}")
    print("全部测试完成！")
    print(f"{'=' * 60}")

A.2 human_approval_test.py

"""
LangGraph 人工审批流程测试（Agent 智能处理反馈 + 记忆回退）

场景：Agent 为用户的公司撰写宣传内容
    1. Agent 生成初始宣传内容（v1），写入 xxx草稿.txt
    2. 用户在终端查看并给出自然语言意见
    3. Agent（LLM + Tools）智能理解用户意图：
       - 用户想修改 → 调用 modify_content 工具生成新版本
       - 用户想要某个历史版本 → 调用 recall_version 工具取回
    4. 用户看到新版本后，说「满意」→ 输出最终版
    5. 不满意 → 继续循环

图结构（5 节点）：
    START → generate → write_draft → human_review
                                          │
                              ├── 满意 ──→ write_final → END
                              └── 不满意 → process_feedback → write_draft → human_review → ...

核心设计理念：
    - 不用关键字匹配，由 LLM Agent 理解用户的自然语言意图
    - versions 列表自动存储所有版本，LLM 可随意调取任意历史版本
    - MemorySaver 确保同一 thread_id 下状态持久化

运行方式：
    python -m app.langgraph_test.human_approval_test
"""

import sys
import os
import json

_project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
if _project_root not in sys.path:
    sys.path.insert(0, _project_root)

import datetime
import operator
from typing import Literal, Annotated
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage, HumanMessage

from pydantic import BaseModel, Field

from app.common import llm


print(f"[INFO] 当前使用模型: {llm.LLM_MODEL_NAME}")
print(f"[INFO] 当前 Base URL: {llm.LLM_BASE_URL}")
print()


# ====================================================================
# State 定义
# ====================================================================

class PromotionState(TypedDict):
    """宣传内容生成流程的状态"""
    company_name: str              # 公司名称
    industry: str                  # 行业
    content: str                   # 当前展示的宣传内容
    versions: Annotated[list, operator.add]  # 所有版本（使用 reducer 追加）
    current_view: int              # 当前展示的版本号
    feedback: str                  # 用户反馈意见
    revision_count: int            # 新版本生成次数（不含初始版本）
    draft_dir: str                 # 草稿保存目录
    is_approved: bool              # 是否审批通过


# ====================================================================
# Agent 工具定义（LLM 通过工具调用来处理用户反馈）
# ====================================================================

class RecallVersionInput(BaseModel):
    version_number: int = Field(description="要取回的版本号，如 1 表示第一版")

@tool(args_schema=RecallVersionInput)
def recall_version(version_number: int) -> str:
    """从历史版本中取回指定版本的内容。当用户想要某个之前的版本时调用此工具。"""
    return f"VERSION_RECALLED:{version_number}"


class ModifyInput(BaseModel):
    current_content: str = Field(description="当前需要修改的内容")
    user_feedback: str = Field(description="用户的修改意见")

@tool(args_schema=ModifyInput)
def modify_content(current_content: str, user_feedback: str) -> str:
    """根据用户的修改意见生成新版本的宣传内容。当用户提出修改要求时调用此工具。"""
    prompt = SystemMessage(content=(
        f"你是一位资深的企业宣传文案专家。请根据用户的修改意见，对以下内容进行修改。\n"
        f"\n当前内容：\n{current_content}\n"
        f"\n用户修改意见：{user_feedback}\n"
        f"\n要求：\n"
        f"1. 根据用户意见进行调整\n"
        f"2. 保持整体结构和语言风格\n"
        f"3. 直接输出修改后的完整内容，不要添加额外说明\n"
    ))
    response = llm.llm.invoke([prompt, HumanMessage(content="请修改")])
    return response.content


# Agent 可用工具列表
AGENT_TOOLS = [recall_version, modify_content]
llm_with_tools = llm.llm.bind_tools(AGENT_TOOLS)

# Agent 系统提示词
AGENT_SYSTEM_PROMPT = SystemMessage(content=(
    "你是一个宣传内容审批助手。用户正在审阅公司宣传内容的不同版本，并给出反馈。\n"
    "你的任务是理解用户的意图，选择合适的工具：\n"
    "\n"
    "1. 如果用户想要某个之前的版本（如\"第一版\"、\"v2\"、\"之前的那个\"、\"第二个版本好\"等），"
    "调用 recall_version 工具，传入对应的版本号。\n"
    "2. 如果用户提出修改意见（如\"改短一点\"、\"加点创意\"、\"更正式些\"等），"
    "调用 modify_content 工具。\n"
    "\n"
    "请根据用户的反馈智能判断应该调用哪个工具。\n"
    "注意：可用的版本列表会在每次对话时提供，请据此判断版本号是否有效。"
))


# ====================================================================
# 节点函数
# ====================================================================

def generate_content(state: PromotionState) -> dict:
    """节点1：LLM 生成初始宣传内容（v1）"""
    company = state["company_name"]
    industry = state.get("industry", "科技")

    print(f"\n  [generate] 正在为「{company}」生成宣传内容...")

    prompt = SystemMessage(content=(
        f"你是一位资深的企业宣传文案专家。请为以下公司撰写一篇宣传内容：\n"
        f"- 公司名称：{company}\n"
        f"- 所属行业：{industry}\n"
        f"\n"
        f"要求：\n"
        f"1. 内容包含：公司简介、核心优势、服务理念、未来愿景\n"
        f"2. 语言风格：专业、大气、有感染力\n"
        f"3. 长度：300~500 字\n"
        f"4. 直接输出宣传内容，不要添加额外的说明文字\n"
    ))

    response = llm.llm.invoke([prompt, HumanMessage(content="请开始撰写")])
    content = response.content

    print(f"  [generate] 内容生成完成（{len(content)} 字）")

    return {
        "content": content,
        "versions": [{"v": 1, "content": content}],
        "current_view": 1,
        "revision_count": 0,
        "feedback": "",
        "is_approved": False,
    }


def write_draft(state: PromotionState) -> dict:
    """节点2：将当前内容写入草稿文件"""
    draft_dir = state.get("draft_dir", "drafts")
    company = state["company_name"]
    content = state["content"]
    rev = state["revision_count"]

    os.makedirs(draft_dir, exist_ok=True)

    if rev == 0:
        filename = f"{company}_草稿.txt"
    else:
        filename = f"{company}_草稿_v{rev + 1}.txt"

    filepath = os.path.join(draft_dir, filename)
    with open(filepath, "w", encoding="utf-8") as f:
        f.write(content)

    print(f"\n  [write_draft] 已写入: {filepath}")
    print(f"  {'─' * 50}")
    preview = content[:200] + "..." if len(content) > 200 else content
    print(f"  {preview}")
    print(f"  {'─' * 50}")

    return {"draft_dir": draft_dir}


def human_review(state: PromotionState) -> dict:
    """节点3：人工审批 — 用户在终端输入自然语言意见

    用户可以用自然语言表达任何想法：
    - "满意" → 审批通过
    - "还是第一版好" → Agent 会智能调取第一版
    - "改短一点" → Agent 会根据意见修改
    - "第二版比第三版好" → Agent 会调取第二版
    """
    rev = state["revision_count"]
    view = state.get("current_view", 1)
    total_versions = len(state.get("versions", []))

    print(f"\n{'=' * 50}")
    print(f"  📋 人工审批（当前：第 {view} 版，共 {total_versions} 个版本）")
    print(f"{'=' * 50}")
    print(f"  请查看上方内容，输入您的意见。")
    print(f"  输入「满意」→ 当前版本作为最终版")
    if total_versions > 1:
        print(f"  您也可以说「第X版」来查看历史版本")
    print(f"{'─' * 50}")

    try:
        feedback = input("  您的意见：").strip()
    except (EOFError, KeyboardInterrupt):
        feedback = "满意"

    if not feedback:
        feedback = "满意"

    is_approved = feedback in ("满意", "通过", "ok", "OK", "好", "可以", "没问题")

    if is_approved:
        print(f"\n  ✅ 审批通过！")
    else:
        print(f"\n  💬 用户反馈：{feedback}")
        print(f"  → Agent 正在智能处理...")

    return {
        "feedback": feedback,
        "is_approved": is_approved,
    }


def process_feedback(state: PromotionState) -> dict:
    """节点4：Agent 智能处理用户反馈（LLM + Tools）

    这是核心 Agent 节点：
    1. 将所有历史版本信息注入 LLM 上下文
    2. LLM 理解用户意图后，调用对应工具：
       - recall_version(n) → 从 versions 中取回第 n 版
       - modify_content(content, feedback) → 根据意见生成新版本
    3. 返回更新后的 content 和 versions
    """
    versions = state.get("versions", [])
    feedback = state["feedback"]
    content = state["content"]
    rev = state["revision_count"]

    # 构建可用版本描述（让 LLM 知道有哪些版本）
    versions_desc = "可用版本列表：\n"
    for v in versions:
        preview = v["content"][:80].replace("\n", " ")
        versions_desc += f"  - 第 {v['v']} 版（{len(v['content'])} 字）：{preview}...\n"

    # 构建 LLM 消息
    messages = [
        AGENT_SYSTEM_PROMPT,
        HumanMessage(content=(
            f"{versions_desc}\n"
            f"当前展示的是第 {state.get('current_view', 1)} 版。\n"
            f"用户反馈：{feedback}\n"
            f"\n请根据用户反馈选择合适的工具。"
        )),
    ]

    # LLM 决策调用哪个工具
    response = llm_with_tools.invoke(messages)

    # 处理工具调用
    if response.tool_calls:
        tool_call = response.tool_calls[0]
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]

        print(f"\n  🤖 [Agent] 决策：调用 {tool_name} 工具")

        if tool_name == "recall_version":
            target_v = tool_args.get("version_number", 1)
            found = None
            for v in versions:
                if v["v"] == target_v:
                    found = v
                    break

            if found:
                new_content = found["content"]
                print(f"  🔄 [Agent] 已取回第 {target_v} 版内容（{len(new_content)} 字）")
                preview = new_content[:150] + "..." if len(new_content) > 150 else new_content
                print(f"  {'─' * 50}")
                print(f"  {preview}")
                print(f"  {'─' * 50}")
                return {
                    "content": new_content,
                    "current_view": target_v,
                }
            else:
                print(f"  ⚠️ [Agent] 未找到第 {target_v} 版，保持当前内容")
                return {}

        elif tool_name == "modify_content":
            new_content = modify_content.invoke(tool_args)
            new_rev = rev + 1
            new_v = new_rev + 1
            print(f"  ✏️ [Agent] 已根据意见生成第 {new_v} 版（{len(new_content)} 字）")
            return {
                "content": new_content,
                "versions": [{"v": new_v, "content": new_content}],
                "current_view": new_v,
                "revision_count": new_rev,
            }
    else:
        print(f"  ⚠️ [Agent] 未调用任何工具，保持当前内容")

    return {}


def write_final(state: PromotionState) -> dict:
    """节点5：将最终版本写入 xxx最终版.txt"""
    draft_dir = state.get("draft_dir", "drafts")
    company = state["company_name"]
    content = state["content"]
    rev = state["revision_count"]
    view = state.get("current_view", 1)

    filepath = os.path.join(draft_dir, f"{company}_最终版.txt")

    header = (
        f"{'=' * 50}\n"
        f"  {company} — 宣传内容（最终版 · 第 {view} 版）\n"
        f"  生成时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
        f"  共产生 {len(state.get('versions', []))} 个版本，修改 {rev} 次\n"
        f"{'=' * 50}\n\n"
    )

    with open(filepath, "w", encoding="utf-8") as f:
        f.write(header)
        f.write(content)

    print(f"\n{'=' * 50}")
    print(f"  🎉 最终版已生成：{filepath}")
    print(f"  来源：第 {view} 版")
    print(f"  共产生 {len(state.get('versions', []))} 个版本，修改 {rev} 次")
    print(f"{'=' * 50}")

    return {}


# ====================================================================
# 条件路由
# ====================================================================

def should_approve(state: PromotionState) -> Literal["write_final", "process_feedback"]:
    """条件边：用户是否满意

    满意 → write_final（生成最终版）
    不满意 → process_feedback（Agent 智能处理）
    """
    if state.get("is_approved", False):
        return "write_final"
    return "process_feedback"


# ====================================================================
# 构建图
# ====================================================================

def build_approval_graph(checkpointer=None):
    """构建人工审批流程图（Agent 智能处理反馈）"""
    builder = StateGraph(PromotionState)

    builder.add_node("generate", generate_content)
    builder.add_node("write_draft", write_draft)
    builder.add_node("human_review", human_review)
    builder.add_node("process_feedback", process_feedback)
    builder.add_node("write_final", write_final)

    builder.add_edge(START, "generate")
    builder.add_edge("generate", "write_draft")
    builder.add_edge("write_draft", "human_review")
    builder.add_conditional_edges(
        "human_review",
        should_approve,
        {
            "write_final": "write_final",
            "process_feedback": "process_feedback",
        },
    )
    builder.add_edge("process_feedback", "write_draft")
    builder.add_edge("write_final", END)

    graph = builder.compile(checkpointer=checkpointer)
    return graph


# ====================================================================
# 运行测试
# ====================================================================

def run_human_approval_test():
    """运行人工审批流程测试（Agent 智能处理 + 记忆回退）"""

    print(f"\n{'=' * 60}")
    print("LangGraph 人工审批流程测试（Agent 智能处理 + 记忆）")
    print("  场景：Agent 智能理解反馈，调取历史版本或生成新版本")
    print(f"{'=' * 60}")

    memory = MemorySaver()
    graph = build_approval_graph(checkpointer=memory)

    print("\n[图结构]")
    print(graph.get_graph().draw_mermaid())

    print("\n请输入公司信息：")
    try:
        company_name = input("  公司名称（默认：星辰科技）：").strip()
        industry = input("  所属行业（默认：人工智能）：").strip()
        draft_dir = input("  草稿保存目录（默认：drafts）：").strip()
    except (EOFError, KeyboardInterrupt):
        company_name = ""
        industry = ""
        draft_dir = ""

    company_name = company_name or "星辰科技"
    industry = industry or "人工智能"
    draft_dir = draft_dir or "drafts"

    if not all('\u4e00' <= c <= '\u9fff' or c.isalnum() for c in company_name):
        company_name = "星辰科技"
    if not all('\u4e00' <= c <= '\u9fff' or c.isalnum() for c in industry):
        industry = "人工智能"

    print(f"\n  公司：{company_name}")
    print(f"  行业：{industry}")
    print(f"  目录：{draft_dir}")

    thread_id = "approval-agent-test"
    config = {"configurable": {"thread_id": thread_id}}

    print(f"\n{'━' * 60}")
    print(f"  开始执行（thread_id: {thread_id}）")
    print(f"{'━' * 60}")

    result = graph.invoke({
        "company_name": company_name,
        "industry": industry,
        "draft_dir": draft_dir,
        "content": "",
        "versions": [],
        "current_view": 0,
        "feedback": "",
        "revision_count": 0,
        "is_approved": False,
    }, config=config)

    # ── 总结 ────────────────────────────────────────────────
    final_path = os.path.join(result.get('draft_dir', 'drafts'), f"{result['company_name']}_最终版.txt")
    total_versions = len(result.get('versions', []))

    print(f"\n{'=' * 60}")
    print("测试完成！Agent 智能处理总结：")
    print(f"  公司名称：{result['company_name']}")
    print(f"  总版本数：{total_versions}")
    print(f"  修改次数：{result['revision_count']}")
    print(f"  最终来源：第 {result.get('current_view', '?')} 版")
    print(f"  最终文件：{final_path}")

    chosen_view = result.get('current_view', 0)
    versions = result.get('versions', [])
    if versions and chosen_view > 0:
        chosen = next((v for v in versions if v["v"] == chosen_view), None)
        if chosen and chosen["content"] == result.get("content", ""):
            print(f"\n  ✅ 验证通过：最终版内容 == 第 {chosen_view} 版内容")
        else:
            print(f"\n  ❌ 验证失败：内容不匹配")

    print(f"{'=' * 60}")

    return result


if __name__ == "__main__":
    run_human_approval_test()

---

Part 5：人工审批流程 — Agent 智能处理反馈 + 记忆回退

5.1 设计理念

传统做法使用硬编码关键字匹配（如 if feedback in ("满意", "ok")）来判断用户意图，这种方式：

• 无法理解自然语言（“第一版比较好” 不会被识别为"想要第一版"）
• 扩展性差（每增加一种意图就要写一堆 if/elif）
• 不够灵活（用户表达方式千变万化）

正确做法：让 LLM Agent 来理解用户意图。

方案	实现方式	优点	缺点
关键字匹配	if feedback in (...)	简单、确定性强	死板、无法理解自然语言
LLM Agent + Tools	LLM 理解意图 → 调用对应工具	灵活、可扩展、支持自然语言	多一次 LLM 调用

5.2 图结构总览

关键设计：

• human_review 只判断"满意/不满意"，不关心用户具体说了什么
• process_feedback 是真正的"大脑"，由 LLM Agent 决定如何处理反馈
• versions 列表存储所有历史版本，Agent 可随时调取任意版本

5.3 State 定义 — 使用 Reducer 追加版本

import operator
from typing import Annotated
from typing_extensions import TypedDict

class PromotionState(TypedDict):
    company_name: str                          # 公司名称
    industry: str                              # 行业
    content: str                               # 当前展示的内容
    versions: Annotated[list, operator.add]    # 所有版本（reducer 追加）
    current_view: int                          # 当前展示的版本号
    feedback: str                              # 用户反馈意见
    revision_count: int                        # 修改次数
    draft_dir: str                             # 草稿目录
    is_approved: bool                          # 是否审批通过

重点：Annotated[list, operator.add]

字段类型	行为	示例
list（无 reducer）	节点返回值覆盖整个列表	{"versions": [v3]} → 只剩 v3
Annotated[list, operator.add]	节点返回值追加到列表	{"versions": [v3]} → [v1, v2, v3]

这是版本历史能正确累积的关键。

5.4 Agent 工具定义

Agent 拥有两个工具，由 LLM 自主决定调用哪个：

# 工具1：取回历史版本
class RecallVersionInput(BaseModel):
    version_number: int = Field(description="要取回的版本号，如 1 表示第一版")

@tool(args_schema=RecallVersionInput)
def recall_version(version_number: int) -> str:
    """从历史版本中取回指定版本的内容。当用户想要某个之前的版本时调用此工具。"""
    return f"VERSION_RECALLED:{version_number}"


# 工具2：根据意见修改内容
class ModifyInput(BaseModel):
    current_content: str = Field(description="当前需要修改的内容")
    user_feedback: str = Field(description="用户的修改意见")

@tool(args_schema=ModifyInput)
def modify_content(current_content: str, user_feedback: str) -> str:
    """根据用户的修改意见生成新版本的宣传内容。当用户提出修改要求时调用此工具。"""
    prompt = SystemMessage(content=f"...根据意见修改...")
    response = llm.llm.invoke([prompt, HumanMessage(content="请修改")])
    return response.content

设计原则：

• 工具描述必须清晰，LLM 根据描述决定何时调用
• recall_version 返回标记字符串，由节点函数执行实际的状态查找
• modify_content 内部调用 LLM 生成新内容，返回纯文本

5.5 核心节点：process_feedback（Agent 智能处理）

这是整个流程的"大脑"：

AGENT_TOOLS = [recall_version, modify_content]
llm_with_tools = llm.llm.bind_tools(AGENT_TOOLS)

AGENT_SYSTEM_PROMPT = SystemMessage(content=(
    "你是一个宣传内容审批助手。理解用户意图，选择合适的工具：\n"
    "1. 用户想要某个版本 → 调用 recall_version\n"
    "2. 用户提出修改意见 → 调用 modify_content\n"
))

def process_feedback(state: PromotionState) -> dict:
    versions = state.get("versions", [])
    feedback = state["feedback"]
    content = state["content"]

    # 1. 构建版本列表描述，让 LLM 知道有哪些版本
    versions_desc = "可用版本列表：\n"
    for v in versions:
        versions_desc += f"  - 第 {v['v']} 版（{len(v['content'])} 字）\n"

    # 2. LLM 理解用户意图，决策调用哪个工具
    response = llm_with_tools.invoke([
        AGENT_SYSTEM_PROMPT,
        HumanMessage(content=f"{versions_desc}\n用户反馈：{feedback}")
    ])

    # 3. 处理工具调用结果
    if response.tool_calls:
        tool_call = response.tool_calls[0]
        if tool_call["name"] == "recall_version":
            # 从 versions 中找到对应版本，覆盖 content
            ...
        elif tool_call["name"] == "modify_content":
            # 执行修改，追加到 versions 列表
            new_content = modify_content.invoke(tool_call["args"])
            return {
                "content": new_content,
                "versions": [{"v": new_v, "content": new_content}],  # reducer 自动追加
                "current_view": new_v,
                "revision_count": rev + 1,
            }

执行流程图：

用户输入 "还是第一版好"
        │
        ▼
  ┌─ process_feedback ─────────────────────┐
  │                                         │
  │  1. 构建 versions_desc 注入 LLM 上下文  │
  │  2. LLM 理解 → 应调用 recall_version(1) │
  │  3. 从 versions 列表查找 v1             │
  │  4. 返回 {"content": v1_content,        │
  │           "current_view": 1}            │
  │                                         │
  └─────────────────────────────────────────┘
        │
        ▼
  write_draft（写入草稿供用户确认）
        │
        ▼
  human_review（用户看到 v1，说"满意"）
        │
        ▼
  write_final → END

5.6 完整数据流示例

以下是一次完整的测试运行：

┌─────────────────────────────────────────────────────────────────┐
│  generate → v1（588字）→ 存入 versions[0]                       │
│  write_draft → 星辰科技_草稿.txt                                 │
│  human_review → 用户："make it more creative" → is_approved=F   │
├─────────────────────────────────────────────────────────────────┤
│  process_feedback → LLM 决策：modify_content                    │
│    → v2（134字）→ 追加到 versions[1]                             │
│  write_draft → 星辰科技_草稿_v2.txt                              │
│  human_review → 用户："make it shorter" → is_approved=F         │
├─────────────────────────────────────────────────────────────────┤
│  process_feedback → LLM 决策：modify_content                    │
│    → v3（64字）→ 追加到 versions[2]                              │
│  write_draft → 星辰科技_草稿_v3.txt                              │
│  human_review → 用户："I prefer version 1" → is_approved=F      │
├─────────────────────────────────────────────────────────────────┤
│  process_feedback → LLM 决策：recall_version(1)                 │
│    → 从 versions 取回 v1（588字）→ 覆盖 content                  │
│  write_draft → 星辰科技_草稿_v3.txt（内容已变回 v1）              │
│  human_review → 用户："ok" → is_approved=T                      │
├─────────────────────────────────────────────────────────────────┤
│  write_final → 星辰科技_最终版.txt（v1 内容）                    │
│  ✅ 验证通过：最终版内容 == 第 1 版内容                           │
└─────────────────────────────────────────────────────────────────┘

State 中 versions 列表的变化过程：

generate 后:     [{"v":1, "content":"v1内容"}]
第1次 revise 后: [{"v":1, ...}, {"v":2, "content":"v2内容"}]
第2次 revise 后: [{"v":1, ...}, {"v":2, ...}, {"v":3, "content":"v3内容"}]
recall v1 后:    不变（只修改 content 和 current_view）

5.7 为什么不用硬编码关键字

用户输入	关键字匹配	LLM Agent
“满意”	✅ 识别	✅ 识别
“ok”	✅ 识别	✅ 识别
“I prefer version 1”	❌ 无法识别	✅ 调用 recall_version(1)
“还是第一版好”	❌ 需要加中文关键字	✅ 调用 recall_version(1)
“第二版比第三版好”	❌ 复杂逻辑	✅ 调用 recall_version(2)
“加点创意，但保留第一版的开头”	❌ 无法处理	✅ 调用 modify_content 并传入具体意见
“都不行，重新写”	❌ 无法处理	✅ 调用 modify_content

结论：LLM Agent 方案用少量代码实现了高度灵活的用户交互，且能处理关键字方案无法覆盖的复杂意图。

5.8 关键设计决策

1. 为什么 process_feedback 内部不执行完整的 ReAct 循环？

process_feedback 只做单轮工具调用，不需要 while 循环。因为：

• 用户意图通常很明确（要么修改、要么选历史版本）
• 单轮调用足够，不需要"思考→行动→观察→再思考"的完整循环
• 简化了实现，避免无限循环风险

2. 为什么 recall_version 返回标记字符串而不是直接操作 state？

# 工具内部
return f"VERSION_RECALLED:{version_number}"  # 返回标记

# 节点函数中
if tool_name == "recall_version":
    target_v = tool_args.get("version_number", 1)
    found = next((v for v in versions if v["v"] == target_v), None)
    return {"content": found["content"], "current_view": target_v}

原因：LangChain 的 @tool 函数无法直接访问 LangGraph 的 state。工具只负责"声明意图"，实际的状态操作由节点函数完成。

3. versions 的 reducer 设计

versions: Annotated[list, operator.add]  # reducer 追加模式

操作	节点返回	versions 实际变化
generate	[{"v":1, "content":"..."}]	[v1]
revise	[{"v":2, "content":"..."}]	[v1, v2]（追加）
recall	不返回 versions	[v1, v2]（不变）

4. MemorySaver 的作用

memory = MemorySaver()
graph = build_approval_graph(checkpointer=memory)
config = {"configurable": {"thread_id": "approval-agent-test"}}
result = graph.invoke(initial_state, config=config)

MemorySaver 确保同一 thread_id 下的状态持久化：

• 多轮 invoke 之间，state 自动保持
• versions 列表跨调用累积
• 即使程序重启（配合持久化 checkpointer），也能恢复状态

5.9 踩坑记录

问题	原因	解决方案
versions 列表只保留最后一个版本	TypedDict 默认覆盖而非追加	使用 Annotated[list, operator.add] reducer
Agent 无法理解"第一版"	硬编码关键字匹配	改用 LLM Agent + Tools 理解自然语言
recall_version 无法访问 state	@tool 函数无法访问 LangGraph state	工具返回标记，节点函数操作 state
LLM 有时不调用工具	系统提示不够明确	在 system prompt 中列出所有可用工具和使用时机
版本号计算错误	len(versions) 在 reducer 下不可靠	改用 revision_count + 1 计算版本号