上周我用 ReAct Agent 搭建了房产推荐系统。这周我发现了瓶颈：复杂查询需要多步操作，顺序执行太慢。我用 LangGraph 重构了整个工作流，性能提升 40%，代码也更清晰了。

前言

很多人做 Agent 时，会陷入一个误区：觉得"能工作"就等于"工作得好"。

我的房产 Agent 就是这样：

• ✅ 能回答问题
• ✅ 能调用工具
• ❌ 但复杂查询很慢
• ❌ 无法保存中间状态
• ❌ 无法并行执行

这周我决定升级工作流。不是加模型，不是加工具，而是把执行引擎做对。

---

第一部分：问题诊断（ReAct 的局限性）

场景 1：复杂查询很慢

ReAct 顺序执行：对比 (2s) → 评分1 (1s) → 评分2 (1s) → 评分3 (1s) = 5s

LangGraph 并行执行：对比 (2s) → 评分1/2/3 并行 (1s) = 3s

真实数据：100 个查询平均 4.8s，LLM 占 79%，工具占 21%，可并行化 60%

场景 2：无法保存中间状态

12 分钟的长任务，中间崩溃全部丢失。加 Checkpoint 后发现 State 过期问题。教训：保存时间戳和版本号。

场景 3：无法可视化工作流

用户投诉推荐不符合要求，花 2 小时才找到意图识别错误（87% 准确率）。加入可视化后能追踪每一步。

场景 4：无法支持复杂的条件分支

问题：
- 不同类型的查询需要不同的处理流程
  * 推荐查询 → 检索 + 生成
  * 对比查询 → 对比 + 分析
  * 投资查询 → 评分 + 趋势分析
  * 政策查询 → 知识库查询
  * 首次购房 → 多步骤指南
- ReAct 无法灵活地根据条件选择不同的路径
- 代码变得复杂且难以维护

【我踩过的坑】
- 用 if-else 链处理 14 种意图，代码变成了 500 行的"意大利面"
- 添加新意图时，需要修改 5 个地方
- 一次改动导致其他意图的行为改变（副作用）
- 代码审查时，没人敢改这部分
- 解决方案：用 LangGraph 的条件边，每种意图独立一个路由函数

场景 5：LLM 意图识别不准确

【真实问题】
用户查询："翠湖花园怎么样？"

ReAct Agent 的识别结果：
- 第1次：query_house（准确）
- 第2次：value_analysis（错误）
- 第3次：general（错误）

准确率只有 33%！

原因分析：
- 查询太简洁，LLM 无法确定用户真实意图
- 没有上下文信息
- Prompt 不够清晰

【我的解决方案】
1. 加入多轮确认：如果置信度 < 0.8，询问用户
2. 加入上下文记忆：记住用户之前的查询
3. 优化 Prompt：给 LLM 更多的示例和指导

【改进后的准确率】
- 简单查询：95%（从 87% 提升）
- 复杂查询：92%（从 85% 提升）
- 模糊查询：78%（从 33% 提升）

场景 6：工具调用失败的恢复

【真实问题】
查询："推荐朝阳区 500 万的房子"

执行过程：
1. 思考：识别为 recommend
2. 行动1：调用 search_houses 工具
3. 观察1：工具超时（数据库连接失败）
4. 思考：？？？（Agent 不知道怎么办）
5. 行动2：重新调用 search_houses
6. 观察2：再次超时
7. 思考：放弃
8. 最终答案：抱歉，我无法完成这个查询

用户体验：很差

【我的解决方案】
1. 加入重试机制：最多重试 3 次，每次等待时间递增
2. 加入降级方案：如果 search_houses 失败，使用缓存数据
3. 加入错误传播：如果都失败，返回有意义的错误信息

【改进后的成功率】
- 第一次成功：92%
- 重试后成功：6%
- 最终失败：2%

---

第二部分：LangGraph 是什么？

LangGraph 是一个基于状态图的工作流框架，用来构建复杂的 Agent 工作流。

核心思想：用图论来表示工作流。

ReAct Agent：
思考 → 行动 → 观察 → 思考 → ...（线性流程）

LangGraph：
    ┌─ 思考 ─┐
    │        ↓
    │      是否继续?
    │      /      \
    │    是        否
    │    ↓        ↓
    │   行动      验证
    │    ↓        ↓
    │   观察      完成
    │    ↓
    └─ 返回

为什么用图而不是链？

理论基础：

• ReAct 是一个有限状态机（FSM），但用链式结构实现，导致代码复杂
• LangGraph 用有向无环图（DAG）表示，更符合工作流的本质
• 图的优势：
  1. 可视化：一眼看出所有可能的路径
  2. 可维护：添加新路径不影响现有代码
  3. 可扩展：支持复杂的条件分支和并行执行
  4. 可调试：能追踪每一步的决策

对比：

特性	链式（ReAct）	图式（LangGraph）
代码行数	500+	200+
添加新意图	修改 5 个地方	添加 1 个函数
调试难度	困难	容易
性能	中等	高
学习曲线	平缓	陡峭

---

核心概念 1：State（状态）

定义：工作流的"记忆"，记录执行过程中的所有信息。

class WorkflowState(TypedDict):
    # 输入
    query: str  # 用户查询
    
    # 过程
    thoughts: List[str]  # 思考历史
    actions: List[Dict]  # 行动历史
    observations: List[str]  # 观察历史
    
    # 结果
    final_answer: str  # 最终答案
    confidence: float  # 置信度
    
    # 控制
    current_iteration: int  # 当前迭代
    max_iterations: int  # 最大迭代
    
    # 性能
    total_time: float  # 总耗时
    tool_call_count: int  # 工具调用次数

例子：

初始状态：
{
  query: "推荐朝阳区500万的房子",
  thoughts: [],
  actions: [],
  observations: [],
  final_answer: "",
  confidence: 0.0,
  current_iteration: 0,
  max_iterations: 3,
  total_time: 0,
  tool_call_count: 0
}

执行后：
{
  query: "推荐朝阳区500万的房子",
  thoughts: ["识别为推荐任务", "需要检索楼盘"],
  actions: [{"tool": "search_houses", "params": {...}}],
  observations: ["找到10个楼盘"],
  final_answer: "推荐璞樾、紫京宸园...",
  confidence: 0.85,
  current_iteration: 2,
  max_iterations: 3,
  total_time: 3.5,
  tool_call_count: 2
}

---

核心概念 2：节点（Node）

定义：工作流的"步骤"，每个节点是一个函数。

def think_node(state: WorkflowState) -> WorkflowState:
    """思考节点：分析当前状态，决定下一步"""
    thought = analyze_state(state)
    state['thoughts'].append(thought)
    return state

def act_node(state: WorkflowState) -> WorkflowState:
    """行动节点：选择工具并执行"""
    action = choose_tool(state)
    state['actions'].append(action)
    return state

def observe_node(state: WorkflowState) -> WorkflowState:
    """观察节点：观察工具结果"""
    observation = execute_tool(state['actions'][-1])
    state['observations'].append(observation)
    return state

def verify_node(state: WorkflowState) -> WorkflowState:
    """验证节点：验证答案准确性"""
    verification = verify_answer(state)
    state['verification'] = verification
    return state

节点类型：

节点	作用	输入	输出
思考	分析状态，决定下一步	当前状态	更新后的状态
行动	选择并执行工具	当前状态	更新后的状态
观察	观察工具结果	当前状态	更新后的状态
验证	验证答案准确性	当前状态	更新后的状态
完成	生成最终答案	当前状态	最终状态

---

核心概念 3：边（Edge）

定义：节点之间的连接。

# 普通边：固定连接
workflow.add_edge("思考", "行动")  # 思考 → 行动

# 条件边：根据条件选择
workflow.add_conditional_edges(
    "思考",
    should_continue,  # 条件函数
    {
        "行动": "行动",      # 如果返回 "行动"，去行动节点
        "完成": "完成"  # 如果返回 "完成"，去完成节点
    }
)

工作流图：

[开始]
    ↓
[思考]  ← ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐
    ↓                           │
[是否继续?]                      │
    ├─ 是 → [行动]              │
    │           ↓               │
    │       [观察] ─ ─ ─ ─ ─ ┘
    │
    └─ 否 → [验证]
                ↓
            [完成]
                ↓
             [END]

---

核心概念 4：条件边（Conditional Edge）

定义：根据状态选择下一个节点。

def should_continue(state: WorkflowState) -> str:
    """判断是否继续迭代"""
    # 检查迭代次数
    if state['current_iteration'] >= state['max_iterations']:
        return "验证"
    
    # 检查思考内容
    if "完成" in state['thoughts'][-1]:
        return "验证"
    
    return "行动"

def route_by_task_type(state: WorkflowState) -> str:
    """根据任务类型路由"""
    if state['task_type'] == TaskType.RECOMMEND:
        return "搜索路由"
    elif state['task_type'] == TaskType.COMPARE:
        return "对比路由"
    else:
        return "通用路由"

---

核心概念 5：检查点（Checkpoint）

定义：保存工作流的中间状态，支持断点续传。

# 保存检查点
checkpoint_id = checkpoint_manager.save_checkpoint(state)

# 加载检查点
state = checkpoint_manager.restore_state(checkpoint_id)

# 从检查点恢复
result = graph.run(query, resume_from=checkpoint_id)

使用场景：

场景1：长时间任务
- 任务1：检索楼盘（5分钟）
- 保存检查点
- 任务2：生成答案（3分钟）
- 如果中断，可以从检查点恢复

场景2：调试
- 运行工作流
- 保存检查点
- 修改代码
- 从检查点恢复，继续调试

---

第三部分：实战优化

优化 1：支持并行执行

问题：计算多个楼盘的评分时，顺序执行太慢

解决方案：在 observe_node 中检测并行机会

def observe_node(state: WorkflowState) -> WorkflowState:
    """观察节点：支持并行执行"""
    
    # 检查是否有多个行动
    if len(state['actions']) > 1:
        # 并行执行
        state['execution_mode'] = ExecutionMode.PARALLEL
        results = execute_actions_parallel(state['actions'], state)
    else:
        # 顺序执行
        state['execution_mode'] = ExecutionMode.SEQUENTIAL
        results = execute_actions_sequential(state['actions'], state)
    
    # 更新观察
    state['observations'].extend(results)
    return state

效果：

查询："对比项目 A、项目 B 和项目 C，并计算评分"

顺序执行：
- 对比楼盘: 2s
- 计算评分1: 1s
- 计算评分2: 1s
- 计算评分3: 1s
- 总耗时: 5s

并行执行：
- 对比楼盘: 2s
- 计算评分1/2/3（并行）: 1s
- 总耗时: 3s

性能提升：40%

---

优化 2：智能路由

问题：不同类型的查询需要不同的处理流程

解决方案：使用条件边实现 14 种任务类型的自动路由

def route_by_task_type(state: WorkflowState) -> str:
    """根据任务类型路由"""
    task_type = state['task_type']
    
    if task_type == TaskType.RECOMMEND:
        return "搜索路由"
    elif task_type == TaskType.COMPARE:
        return "对比路由"
    elif task_type == TaskType.VALUE_ANALYSIS:
        return "分析路由"
    elif task_type == TaskType.INVESTMENT:
        return "投资路由"
    # ... 14 种任务类型
    else:
        return "通用路由"

支持的任务类型：

基础任务（4个）：
- recommend - 推荐楼盘
- query_house - 查询具体楼盘信息
- compare - 对比楼盘
- query_facility - 查询配套信息

细分任务（10个）：
- value_analysis - 房产价值解读
- investment - 投资房咨询
- school_district - 学区房咨询
- first_time_buyer - 首次购房指南
- second_hand - 二手房交易指南
- decoration - 装修指南
- out_of_town - 外地房指南
- short_term - 短期租赁指南
- policy - 政策解读
- general - 通用对话/兜底

复杂查询示例：

查询1："我是首次购房者，预算 500 万，想要朝阳区的学区房，
需要分析投资价值、对比周边楼盘、查询政策支持"

识别的意图：
- first_time_buyer（首次购房）
- school_district（学区房）
- investment（投资分析）
- compare（对比）
- policy（政策查询）

执行路由：
1. 首次购房指南路由 → 生成购房流程
2. 学区房路由 → 检索学区房
3. 投资分析路由 → 计算投资评分
4. 对比路由 → 对比周边楼盘
5. 政策查询路由 → 查询相关政策
6. 综合路由 → 生成综合答案

查询2："翠湖花园和星河湾哪个更适合投资？
考虑升值潜力、租赁收益、政策支持"

识别的意图：
- compare（对比）
- investment（投资分析）
- policy（政策查询）

执行路由：
1. 对比路由 → 对比两个楼盘
2. 投资分析路由 → 计算升值潜力和租赁收益
3. 政策查询路由 → 查询相关政策
4. 综合路由 → 生成投资建议

---

优化 3：状态持久化

问题：无法保存中间状态，长时间任务容易丢失

解决方案：实现 CheckpointManager

class CheckpointManager:
    def save_checkpoint(self, state: WorkflowState) -> str:
        """保存检查点"""
        checkpoint_id = generate_checkpoint_id()
        checkpoint_data = {
            'checkpoint_id': checkpoint_id,
            'timestamp': datetime.now().isoformat(),
            'query': state['query'],
            'thoughts': state['thoughts'],
            'actions': state['actions'],
            'observations': state['observations'],
            'final_answer': state['final_answer'],
            'confidence': state['confidence'],
            'total_time': state['total_time'],
            'tool_call_count': state['tool_call_count']
        }
        
        # 保存到文件
        checkpoint_path = f"data/checkpoints/{checkpoint_id}.json"
        with open(checkpoint_path, 'w') as f:
            json.dump(checkpoint_data, f, indent=2)
        
        return checkpoint_id
    
    def restore_state(self, checkpoint_id: str) -> WorkflowState:
        """恢复检查点"""
        checkpoint_path = f"data/checkpoints/{checkpoint_id}.json"
        with open(checkpoint_path, 'r') as f:
            checkpoint_data = json.load(f)
        
        # 恢复状态
        state = WorkflowState(...)
        state['query'] = checkpoint_data['query']
        state['thoughts'] = checkpoint_data['thoughts']
        # ... 恢复其他字段
        
        return state

效果：

检查点大小：5-8 KB（很小）
保存时间：< 100ms
恢复时间：< 100ms
支持断点续传：✓

---

优化 4：工作流可视化

问题：无法看到工作流的执行过程

解决方案：生成工作流图

def visualize(self, output_path: str = "workflow_graph.png"):
    """生成工作流图"""
    # 获取图结构
    structure = self.get_graph_structure()
    
    # 使用 graphviz 生成图
    graph = graphviz.Digraph()
    
    # 添加节点
    for node in structure['nodes']:
        graph.node(node, node)
    
    # 添加边
    for source, target in structure['edges']:
        graph.edge(source, target)
    
    # 保存
    graph.render(output_path)

输出：

节点:
  - think
  - act
  - observe
  - verify
  - finish
  - save_checkpoint

边:
  think → act
  think → verify
  think → finish
  act → observe
  observe → think
  verify → finish
  finish → save_checkpoint
  save_checkpoint → END

---

第四部分：性能对比

ReAct vs LangGraph

功能	ReAct Agent	LangGraph
顺序执行	✓	✓
并行执行	✗	✓
状态持久化	✗	✓
条件分支	简单	复杂
可视化	✗	✓
错误恢复	✗	✓
性能	中等	高

执行时间对比

查询1："推荐朝阳区 500 万的房子"
- ReAct: 3.5s
- LangGraph: 3.5s
- 提升: 0%（简单查询没有优势）

查询2："对比翠湖花园、星河湾和滨江豪庭"
- ReAct: 4s
- LangGraph: 3s
- 提升: 25%

查询3："对比翠湖花园、星河湾和滨江豪庭，并计算它们的综合评分，我想要地铁便利、学区优质的三居室"
- ReAct: 5s
- LangGraph: 3s
- 提升: 40%

平均提升：20%

---

第四部分 B：成本-收益分析

什么时候用 ReAct，什么时候用 LangGraph？

ReAct 足够的场景：

1. 简单查询（单一意图）
   - 推荐楼盘
   - 查询房价
   - 查询配套
   - 性能：3-4s
   - 代码复杂度：低
   - 维护成本：低

2. 实时性要求高
   - 用户期望 < 2s 响应
   - ReAct 的 LLM 调用时间已经是瓶颈
   - 并行执行收益不大

3. 工具调用少（< 3 个）
   - 并行化收益不明显
   - 状态管理简单

必须用 LangGraph 的场景：

1. 复杂多意图查询
   - 首次购房 + 学区房 + 投资分析
   - 需要 5+ 个工具调用
   - 需要条件分支

2. 长时间任务（> 5 分钟）
   - 需要 Checkpoint 保存中间状态
   - 需要错误恢复

3. 需要可视化和调试
   - 用户投诉"为什么推荐这个？"
   - 需要追踪决策过程

4. 需要扩展性
   - 计划添加更多意图
   - 计划添加更多工具

迁移成本分析

从 ReAct 迁移到 LangGraph 的成本：

时间成本：
- 学习 LangGraph：2-3 天
- 重构代码：1-2 天
- 测试和调试：1-2 天
- 总计：4-7 天

代码成本：
- 原有代码：500 行（if-else 链）
- 新代码：200 行（LangGraph）
- 删除代码：300 行
- 收益：代码更清晰，更易维护

性能成本：
- LangGraph 框架开销：< 100ms
- 可以忽略不计

收益：
- 代码行数减少 60%
- 添加新意图时间减少 80%
- 调试时间减少 70%
- 用户投诉减少 50%（因为能追踪决策）

什么时候不值得迁移？

1. 系统已经稳定运行
   - 没有新需求
   - 没有用户投诉
   - 迁移风险 > 收益

2. 团队规模小
   - 只有 1-2 个开发者
   - 代码复杂度不是主要问题
   - 学习成本相对较高

3. 项目即将下线
   - 不值得投入

---

第五部分：完整代码示例

基础使用

from src.workflows.house_graph import HouseGraph

# 创建工作流
graph = HouseGraph()

# 运行复杂查询
result = graph.run(
    query="我是首次购房者，预算 500 万，想要朝阳区的学区房，需要分析投资价值、对比周边楼盘、查询政策支持",
    max_iterations=5
)

# 查看结果
print(f"识别的意图: {result['intents']}")
print(f"最终答案: {result['final_answer']}")
print(f"置信度: {result['confidence']:.2f}")
print(f"耗时: {result['total_time']:.2f}s")
print(f"迭代: {result['iterations']}")
print(f"工具调用: {result['tool_calls']}")

检查点功能

# 第一次运行
result1 = graph.run(
    query="我是首次购房者，预算 500 万，想要朝阳区的学区房，需要分析投资价值、对比周边楼盘、查询政策支持",
    max_iterations=3
)

checkpoint_id = result1['checkpoint_id']
print(f"检查点已保存: {checkpoint_id}")

# 从检查点恢复
result2 = graph.run(
    query="我是首次购房者，预算 500 万，想要朝阳区的学区房，需要分析投资价值、对比周边楼盘、查询政策支持",
    max_iterations=5,
    resume_from=checkpoint_id
)

print(f"恢复后的迭代次数: {result2['iterations']}")
print(f"恢复后的置信度: {result2['confidence']:.2f}")

流式输出

# 流式运行工作流
for event in graph.run_with_streaming(
    query="翠湖花园和星河湾哪个更适合投资？考虑升值潜力、租赁收益、政策支持",
    max_iterations=4
):
    print(f"事件: {event}")
    # 实时查看执行过程

工作流可视化

# 生成工作流图
graph.visualize(output_path="workflow_graph.png")

# 获取图结构
structure = graph.get_graph_structure()
print("节点:", structure['nodes'])
print("边:", structure['edges'])

---

第六部分：生产环境的考虑

1. 监控和日志

问题：如何知道工作流在哪一步出问题了？

解决方案：

import logging
from datetime import datetime

class MonitoredGraph:
    def __init__(self, graph):
        self.graph = graph
        self.logger = logging.getLogger(__name__)
    
    def run(self, query, **kwargs):
        """运行工作流，并记录每一步"""
        start_time = datetime.now()
        
        try:
            result = self.graph.run(query, **kwargs)
            
            # 记录成功
            self.logger.info(f"Query succeeded: {query[:50]}...")
            self.logger.info(f"Total time: {(datetime.now() - start_time).total_seconds():.2f}s")
            self.logger.info(f"Iterations: {result['iterations']}")
            self.logger.info(f"Confidence: {result['confidence']:.2f}")
            
            return result
        
        except Exception as e:
            # 记录失败
            self.logger.error(f"Query failed: {query[:50]}...")
            self.logger.error(f"Error: {str(e)}")
            self.logger.error(f"Time elapsed: {(datetime.now() - start_time).total_seconds():.2f}s")
            
            raise

监控指标：

- 平均响应时间
- 成功率
- 各步骤的耗时分布
- 错误率和错误类型
- LLM 调用次数
- 工具调用次数

2. 超时处理

问题：如果某个工具调用超时怎么办？

解决方案：

import asyncio
from functools import wraps

def with_timeout(timeout_seconds):
    """为节点添加超时保护"""
    def decorator(func):
        @wraps(func)
        async def wrapper(state):
            try:
                return await asyncio.wait_for(
                    func(state),
                    timeout=timeout_seconds
                )
            except asyncio.TimeoutError:
                state['error'] = f"Timeout after {timeout_seconds}s"
                state['fallback_used'] = True
                return state
        return wrapper
    return decorator

@with_timeout(5)
async def search_houses_with_timeout(state):
    """搜索楼盘，5 秒超时"""
    return await search_houses(state)

3. 错误恢复和降级

问题：如果工具调用失败，怎么提供有意义的答案？

解决方案：

class ResilientGraph:
    def __init__(self, graph, cache):
        self.graph = graph
        self.cache = cache  # 缓存上次的结果
    
    def run(self, query, **kwargs):
        """运行工作流，支持降级"""
        try:
            # 第一次尝试
            return self.graph.run(query, **kwargs)
        
        except ToolCallError as e:
            # 工具调用失败，尝试使用缓存
            cached_result = self.cache.get(query)
            
            if cached_result:
                # 使用缓存，但标记为降级
                cached_result['degraded'] = True
                cached_result['error'] = str(e)
                return cached_result
            else:
                # 没有缓存，返回错误
                raise

4. LLM 不稳定性处理

问题：LLM 的输出不稳定，有时识别错误，有时工具选择错误

真实数据：

意图识别准确率：
- 第一次：87%
- 重试后：92%
- 多轮确认后：95%

工具选择准确率：
- 第一次：89%
- 使用 Few-shot 后：94%

解决方案：

class RobustThinkNode:
    def __init__(self, llm, confidence_threshold=0.8):
        self.llm = llm
        self.confidence_threshold = confidence_threshold
    
    def __call__(self, state):
        """思考节点，支持多轮确认"""
        
        # 第一次识别
        thought, confidence = self.llm.think(state['query'])
        
        # 如果置信度低，进行多轮确认
        if confidence < self.confidence_threshold:
            # 使用不同的 Prompt 重新识别
            thought2, confidence2 = self.llm.think_with_examples(state['query'])
            
            # 取置信度更高的结果
            if confidence2 > confidence:
                thought = thought2
                confidence = confidence2
        
        state['thoughts'].append(thought)
        state['confidence'] = confidence
        
        return state

5. 成本控制

问题：LLM API 调用很贵，如何控制成本？

解决方案：

class CostAwareGraph:
    def __init__(self, graph, max_cost_per_query=0.1):
        self.graph = graph
        self.max_cost_per_query = max_cost_per_query
        self.current_cost = 0
    
    def run(self, query, **kwargs):
        """运行工作流，监控成本"""
        self.current_cost = 0
        
        # 使用便宜的模型进行意图识别
        intent = self.cheap_llm.identify_intent(query)
        self.current_cost += 0.001
        
        # 根据意图选择模型
        if intent == "simple":
            # 简单查询用便宜模型
            result = self.cheap_llm.generate(query)
            self.current_cost += 0.01
        else:
            # 复杂查询用贵模型
            result = self.expensive_llm.generate(query)
            self.current_cost += 0.05
        
        # 检查成本
        if self.current_cost > self.max_cost_per_query:
            raise CostExceededError(f"Cost {self.current_cost} exceeds limit")
        
        return result

---

第七部分：扩展性讨论

从 5 个节点到 50 个节点

问题：如果系统变得更复杂，怎么管理？

当前架构（5 个节点）：

think → act → observe → verify → finish

扩展后的架构（50 个节点）：

think → route → [
  search_route → search → observe_search → think,
  compare_route → compare → observe_compare → think,
  analyze_route → analyze → observe_analyze → think,
  policy_route → policy → observe_policy → think,
  ...
] → verify → finish

管理方案：

class ModularGraph:
    """模块化的工作流管理"""
    
    def __init__(self):
        self.routes = {}  # 存储各个路由
        self.graph = StateGraph(WorkflowState)
    
    def register_route(self, route_name, route_func):
        """注册新的路由"""
        self.routes[route_name] = route_func
    
    def build(self):
        """构建完整的图"""
        # 添加通用节点
        self.graph.add_node("think", think_node)
        self.graph.add_node("route", self.route_node)
        self.graph.add_node("verify", verify_node)
        self.graph.add_node("finish", finish_node)
        
        # 添加各个路由
        for route_name, route_func in self.routes.items():
            self.graph.add_node(route_name, route_func)
        
        # 添加边
        self.graph.set_entry_point("think")
        self.graph.add_conditional_edges("think", self.route_node, self.routes.keys())
        # ... 其他边
        
        return self.graph.compile()
    
    def route_node(self, state):
        """根据意图选择路由"""
        intent = state['intent']
        return self.routes[intent]

从 14 种意图到 100 种意图

问题：如何避免意图识别的"组合爆炸"？

解决方案：分层意图识别

class HierarchicalIntentRouter:
    """分层意图识别"""
    
    def __init__(self):
        # 第一层：大类
        self.level1_intents = {
            "search": ["recommend", "query_house", "query_facility"],
            "compare": ["compare", "compare_price", "compare_location"],
            "analyze": ["value_analysis", "investment", "risk_analysis"],
            "guide": ["first_time_buyer", "second_hand", "decoration"],
            "policy": ["policy", "tax", "subsidy"]
        }
        
        # 第二层：细分
        self.level2_intents = {
            "search": ["recommend", "query_house", "query_facility"],
            # ...
        }
    
    def identify(self, query):
        """两层识别"""
        # 第一层：识别大类（快速）
        level1 = self.llm.identify_level1(query)
        
        # 第二层：识别细分（精准）
        level2 = self.llm.identify_level2(query, level1)
        
        return level1, level2

避免状态爆炸

问题：State 中的字段越来越多，变得难以管理

解决方案：使用嵌套 State

class NestedWorkflowState(TypedDict):
    # 基础信息
    query: str
    intent: str
    
    # 搜索结果
    search_state: SearchState
    
    # 对比结果
    compare_state: CompareState
    
    # 分析结果
    analyze_state: AnalyzeState
    
    # 最终结果
    final_answer: str
    confidence: float

class SearchState(TypedDict):
    query: str
    results: List[Dict]
    total_time: float

class CompareState(TypedDict):
    items: List[str]
    comparison: Dict
    winner: str

性能瓶颈分析

真实数据（100 个查询的平均值）：

总耗时：4.2s

时间分布：
- LLM 调用（意图识别）：0.5s（12%）
- LLM 调用（工具选择）：0.3s（7%）
- LLM 调用（答案生成）：2.5s（60%）
- 工具执行：0.8s（19%）
- 框架开销：0.1s（2%）

瓶颈：
1. 答案生成 LLM 调用（60%）
   - 无法优化（必须调用 LLM）
   - 可以用缓存减少重复调用

2. 工具执行（19%）
   - 可以并行化（已实现）
   - 可以缓存结果

3. 意图识别 LLM 调用（12%）
   - 可以用更快的模型
   - 可以缓存常见查询

优化空间：
- 缓存：可以减少 30% 的 LLM 调用
- 并行化：已经实现，收益 5-10%
- 模型选择：用更快的模型做意图识别，可以减少 50% 的时间

---

总结：我学到了什么

1. ReAct 的局限性

ReAct Agent：
- 顺序执行（无法并行）
- 简单状态管理（无法持久化）
- 无条件分支（无法复杂路由）
- 无可视化（难以调试）

2. LangGraph 的优势

LangGraph：
- 支持并行执行（性能 +40%）
- 完整状态管理（支持持久化）
- 复杂条件分支（支持 14 种任务类型）
- 工作流可视化（易于理解和调试）
- 错误恢复（支持断点续传）

3. 升级的关键

第一步：定义 State（工作流的记忆）
第二步：定义 Node（工作流的步骤）
第三步：定义 Edge（节点之间的连接）
第四步：定义 Conditional Edge（智能路由）
第五步：实现 Checkpoint（状态持久化）

第八部分：实施指南

如何评估自己的项目是否需要 LangGraph？

评分表：

1. 意图类型数量
   - < 5 种：0 分
   - 5-10 种：1 分
   - 10-20 种：2 分
   - > 20 种：3 分

2. 工具调用复杂度
   - 简单（< 3 个工具）：0 分
   - 中等（3-10 个工具）：1 分
   - 复杂（> 10 个工具）：2 分

3. 条件分支复杂度
   - 简单（线性流程）：0 分
   - 中等（2-3 个分支）：1 分
   - 复杂（> 3 个分支）：2 分

4. 需要可视化和调试
   - 不需要：0 分
   - 需要：2 分

5. 需要状态持久化
   - 不需要：0 分
   - 需要：2 分

总分：
- 0-2 分：不需要 LangGraph，用 if-else 链就够了
- 3-5 分：可以考虑 LangGraph，但不是必须
- 6+ 分：必须用 LangGraph

从 ReAct 迁移到 LangGraph 的实战步骤

第 1 步：定义 State

# 列出所有需要保存的信息
class WorkflowState(TypedDict):
    query: str
    intent: str
    search_results: List[Dict]
    comparison_result: Dict
    analysis_result: Dict
    final_answer: str
    confidence: float
    error: Optional[str]

第 2 步：定义 Node

# 将现有的函数改成 Node
def think_node(state: WorkflowState) -> WorkflowState:
    """原来的 think 函数"""
    intent = identify_intent(state['query'])
    state['intent'] = intent
    return state

def search_node(state: WorkflowState) -> WorkflowState:
    """原来的 search 函数"""
    try:
        results = search_houses(state['query'])
        state['search_results'] = results
    except Exception as e:
        state['error'] = str(e)
    return state

第 3 步：定义 Edge 和条件边

# 定义节点之间的连接
workflow = StateGraph(WorkflowState)

workflow.add_node("think", think_node)
workflow.add_node("search", search_node)
workflow.add_node("compare", compare_node)
workflow.add_node("finish", finish_node)

# 条件边：根据意图选择路由
def route_by_intent(state):
    if state['intent'] == "recommend":
        return "search"
    elif state['intent'] == "compare":
        return "compare"
    else:
        return "finish"

workflow.add_conditional_edges("think", route_by_intent)
workflow.add_edge("search", "finish")
workflow.add_edge("compare", "finish")

第 4 步：编译和测试

# 编译 Graph
graph = workflow.compile()

# 用现有的测试用例验证
test_queries = [
    "推荐朝阳区 500 万的房子",
    "翠湖花园和星河湾哪个好？",
    "璞樾值不值得买？"
]

for query in test_queries:
    result = graph.run(query)
    assert result['final_answer'] is not None
    print(f"✓ {query}")

第 5 步：逐步迁移

- 先迁移简单的意图（recommend、query_house）
- 再迁移中等复杂的意图（compare、value_analysis）
- 最后迁移复杂的意图（first_time_buyer、investment）
- 每迁移一个意图，就运行测试
- 最后删除旧代码

常见的迁移陷阱和解决方案

陷阱 1：State 设计不当导致状态爆炸

❌ 错误做法：
class WorkflowState(TypedDict):
    query: str
    temp_var1: str
    temp_var2: str
    temp_var3: str
    # ... 30 个字段

✅ 正确做法：
class WorkflowState(TypedDict):
    query: str
    search_state: SearchState  # 嵌套 State
    compare_state: CompareState
    final_answer: str

# 临时变量在 Node 内部处理
def search_node(state):
    temp_results = search_houses(state['query'])
    state['search_state'] = process_results(temp_results)
    return state

陷阱 2：循环依赖导致死循环

❌ 错误做法：
workflow.add_edge("think", "act")
workflow.add_edge("act", "think")  # 无限循环

✅ 正确做法：
def should_continue(state):
    if state['current_iteration'] >= state['max_iterations']:
        return "finish"
    return "act"

workflow.add_conditional_edges("think", should_continue, {
    "act": "act",
    "finish": "finish"
})

陷阱 3：条件边返回的节点不存在

❌ 错误做法：
def route(state):
    if state['intent'] == "recommend":
        return "search"  # 但没有定义 search 节点
    return "finish"

✅ 正确做法：
workflow.add_node("search", search_node)  # 先定义节点

def route(state):
    if state['intent'] == "recommend":
        return "search"  # 现在节点存在了
    return "finish"

陷阱 4：忘记处理错误导致 State 不一致

❌ 错误做法：
def search_node(state):
    results = search_houses(state['query'])  # 可能抛异常
    state['search_results'] = results
    return state

✅ 正确做法：
def search_node(state):
    try:
        results = search_houses(state['query'])
        state['search_results'] = results
        state['error'] = None
    except Exception as e:
        state['search_results'] = []
        state['error'] = str(e)
    return state

陷阱 5：Checkpoint 恢复时 State 过期

❌ 错误做法：
checkpoint = save_checkpoint(state)
# 1 小时后恢复
state = restore_checkpoint(checkpoint)
# State 中的 LLM 结果可能已经过期

✅ 正确做法：
checkpoint = {
    'state': state,
    'timestamp': datetime.now(),
    'version': 1
}

def restore_checkpoint(checkpoint):
    # 检查时间戳
    if datetime.now() - checkpoint['timestamp'] > timedelta(hours=1):
        # 重新执行，不使用过期的 State
        return None
    return checkpoint['state']

---

学习进度

第01周：从零到一：用 LangChain 搭建房产 RAG 系统 ✅
第02周：LlamaIndex 框架与 Transformer 原理 ⏳
第03周：RAG 优化与向量数据库 ✅
第04周：RAG 评估与数据处理 ✅
第05周：LangGraph 与 Agent 开发 ✅
├─ ReAct Agent 基础
├─ LangGraph 核心概念
├─ 并行执行优化
├─ 状态持久化
├─ 智能路由
└─ 工作流可视化
第06周：AutoGen 与 CrewAI 框架对比 ⏳
├─ AutoGen 简单 Agent
├─ CrewAI 简单 Agent
├─ 三框架对比（LangGraph vs AutoGen vs CrewAI）
└─ 框架选择指南
第07周：Verification RAG 验证机制增强 ⏳
第08周：GraphRAG 知识图谱 ⏳
第09周：LangServe 部署 + Contextual Retrieval ⏳
第10周：DSPy 框架与系统优化 + Agentic RAG ⏳
第11周：ChatBI 自然语言查询系统 ⏳