链式调用与状态管理

课程简介

多步骤链式调用、中间结果传递、错误处理。

🎬 本课程视频：Building Systems with ChatGPT API — ChatGPT API 系统构建

多步链式调用与状态管理

一、从单步到多步

真实世界的 LLM 应用很少「一问一答」就结束。大多数业务需要多个步骤的链式处理：先理解用户意图，再检索相关信息，然后分析推理，接着生成回答，最后还要做输出合规性检查。

多步链式调用让这些步骤可以被结构化地管理和追踪。

二、三种基本调用模式

2.1 顺序链

顺序链是最基础和常用的模式。前一步的输出作为后一步的输入，像流水线一样依次执行。

def sequential_pipeline(user_input):
    '''顺序处理管道'''
    # Step 1: 理解意图
    intent = classify_intent(user_input)

    # Step 2: 检索信息
    context = retrieve_info(user_input, intent)

    # Step 3: 生成回答
    answer = generate_answer(user_input, context, intent)

    # Step 4: 安全检查
    safe_answer = safety_check(answer)

    return safe_answer

适用场景：
- 有明确依赖关系的多步骤流程
- 每个步骤需要单独调优
- 需要在中间步骤进行干预或修正

状态字典实现：

class SequentialPipeline:
    '''使用统一状态字典的顺序管道'''

    def __init__(self):
        self.steps = []

    def add_step(self, name, func):
        self.steps.append({"name": name, "func": func})

    def run(self, initial_input):
        state = {
            "input": initial_input,
            "step_results": {},
            "errors": []
        }

        for step in self.steps:
            try:
                # 执行当前步骤
                result = step["func"](state)
                # 保存结果到状态字典
                state["step_results"][step["name"]] = result
                state["current_output"] = result
            except Exception as e:
                state["errors"].append({
                    "step": step["name"],
                    "error": str(e)
                })
                # 根据策略决定是否继续
                if not self._should_continue(state):
                    break

        return state

2.2 路由链

路由链根据输入的内容或中间结果选择不同的下游路径。

def router_pipeline(user_input):
    '''路由管道'''
    # 1. 分类
    category = classify(user_input)

    # 2. 根据类别路由
    if category == "technical":
        return tech_pipeline(user_input)
    elif category == "billing":
        return billing_pipeline(user_input)
    elif category == "general":
        return general_pipeline(user_input)
    else:
        return fallback_pipeline(user_input)

路由条件类型：
- 基于输入内容的分类
- 基于中间结果的判断
- 基于用户属性的个性化路由
- 基于当前系统负载的容量路由

2.3 并行链

并行链让多个独立的任务同时执行，提升效率。

import asyncio

async def parallel_pipeline(user_input):
    '''并行处理管道'''
    # 同时执行多个独立分析
    task1 = analyze_sentiment(user_input)
    task2 = extract_entities(user_input)
    task3 = check_safety(user_input)

    # 等待所有任务完成
    sentiment, entities, safety = await asyncio.gather(
        task1, task2, task3
    )

    # 汇总结果
    return {
        "sentiment": sentiment,
        "entities": entities,
        "safety_check": safety
    }

三、中间状态管理

3.1 统一状态字典

中间状态管理是多步链式调用的核心挑战。推荐使用统一状态字典：

class PipelineState:
    '''管道的统一状态管理'''

    def __init__(self, initial_input):
        self.state = {
            "input": initial_input,          # 原始输入
            "context": {},                    # 累积的上下文
            "step_outputs": {},              # 每一步的输出
            "metadata": {                     # 元数据
                "start_time": time.time(),
                "step_order": [],
                "token_usage": 0
            },
            "errors": []                     # 错误记录
        }

    def get(self, key, default=None):
        return self.state.get(key, default)

    def set(self, key, value):
        self.state[key] = value

    def record_step(self, step_name, output):
        '''记录步骤执行结果'''
        self.state["step_outputs"][step_name] = output
        self.state["metadata"]["step_order"].append(step_name)

    def get_step_output(self, step_name):
        '''获取指定步骤的输出'''
        return self.state["step_outputs"].get(step_name)

3.2 状态管理的最佳实践

1. 显式读写：每步明确声明读什么、写什么
2. 不可变关键数据：重要数据写入后不修改，需要变更时创建新版本
3. 定期快照：关键节点时记录状态快照，便于回溯
4. 大小控制：避免状态字典过大（如塞入整段对话历史）

四、错误处理

4.1 每步独立捕获

class RobustPipeline:
    '''具有错误处理能力的管道'''

    def execute_step(self, step_func, state, retry_count=2):
        '''执行一个步骤，带重试和错误处理'''
        for attempt in range(retry_count + 1):
            try:
                return step_func(state)
            except TemporaryError as e:
                if attempt < retry_count:
                    # 临时错误，等待后重试
                    time.sleep(2 ** attempt)  # 指数退避
                    continue
                else:
                    # 重试耗尽，执行降级策略
                    return self._degrade_step(step_func, state, e)
            except FatalError as e:
                # 致命错误，立即终止
                state["errors"].append({"step": step_func.__name__, "error": str(e)})
                raise

4.2 降级策略

def _degrade_step(self, step_func, state, error):
    '''步骤执行失败时的降级策略'''
    step_name = step_func.__name__

    # 1. 尝试简化版本
    if hasattr(step_func, "fallback"):
        return step_func.fallback(state)

    # 2. 返回默认值
    return {"status": "degraded", "error": str(error), "default_value": None}

五、流式输出

5.1 边生成边推送

对于耗时较长的多步流程，流式输出可以大幅改善用户体验：

async def streaming_pipeline(user_input):
    '''流式输出管道'''
    # 第一步：分类（快速）
    category = await classify(user_input)
    yield {"type": "status", "content": f"识别为{category}问题"}

    # 第二步：检索（较慢）
    context = await retrieve(user_input, category)
    yield {"type": "status", "content": "信息检索完成"}

    # 第三步：生成（流式输出）
    async for chunk in generate_stream(user_input, context):
        yield {"type": "token", "content": chunk}

六、实际案例：智能客服管道

class CustomerServicePipeline:
    '''智能客服多步管道'''

    async def handle(self, user_message):
        state = PipelineState(user_message)

        # Step 1: 安全过滤
        safety = await self.safety_check(state)
        state.record_step("safety_check", safety)
        if not safety.passed:
            return self.safe_reply()

        # Step 2: 意图识别
        intent = await self.classify_intent(state)
        state.record_step("intent", intent)

        # Step 3: 信息检索
        context = await self.retrieve_knowledge(state, intent)
        state.record_step("retrieval", context)

        # Step 4: 并行质量检查
        quality_check = await self.check_quality(state)
        state.record_step("quality_check", quality_check)

        # Step 5: 生成回答
        answer = await self.generate(state)
        state.record_step("generation", answer)

        return answer

七、总结

多步链式调用是 LLM 生产系统的核心设计模式。顺序链适合有依赖关系的流水线任务，路由链实现条件分支，并行链提升多角度分析的效率。统一状态字典精细管理每一步的中间结果，完善的错误处理机制和流式输出能力确保系统的可靠性和用户体验。

六、中间状态管理的最佳实践

6.1 状态设计原则

1. 最小化：只存储必要的数据
2. 类型安全：尽量使用有类型的数据结构
3. 可序列化：确保中间状态可以序列化和反序列化
4. 可追溯：记录状态变更的时间线

6.2 常见状态模式

单例模式：整个系统只有一个共享状态字典

分片模式：不同组件使用不同的状态空间

事件溯源：不修改状态，而是追加状态变更事件

6.3 状态序列化与持久化

import pickle
from datetime import datetime

def save_state(state, path):
    # 添加时间戳
    state["_saved_at"] = datetime.now().isoformat()
    with open(path, "wb") as f:
        pickle.dump(state, f)

def load_state(path):
    with open(path, "rb") as f:
        return pickle.load(f)

七、生产级链式调用的考虑

1. 错误处理：每个步骤都需要 try-catch
2. 超时控制：每个 LLM 调用设置超时
3. 重试策略：临时失败自动重试
4. 日志记录：记录每个步骤的输入和输出
5. 监控告警：监控链路完整性和延迟

八、总结

关键要点回顾：
- 顺序链适用于固定流程
- 路由链实现条件分支
- 并行链提升独立任务的效率
- 统一状态字典是管理中间数据的核心
- 生产级部署需要错误处理、超时和监控

六、中间状态管理的最佳实践

6.1 状态设计原则

1. 最小化：只存储必要的数据
2. 类型安全：尽量使用有类型的数据结构
3. 可序列化：确保中间状态可以序列化和反序列化
4. 可追溯：记录状态变更的时间线

6.2 常见状态模式

单例模式：整个系统只有一个共享状态字典

分片模式：不同组件使用不同的状态空间

事件溯源：不修改状态，而是追加状态变更事件

6.3 状态序列化与持久化

import pickle
from datetime import datetime

def save_state(state, path):
    # 添加时间戳
    state["_saved_at"] = datetime.now().isoformat()
    with open(path, "wb") as f:
        pickle.dump(state, f)

def load_state(path):
    with open(path, "rb") as f:
        return pickle.load(f)

七、生产级链式调用的考虑

1. 错误处理：每个步骤都需要 try-catch
2. 超时控制：每个 LLM 调用设置超时
3. 重试策略：临时失败自动重试
4. 日志记录：记录每个步骤的输入和输出
5. 监控告警：监控链路完整性和延迟

八、总结

关键要点回顾：
- 顺序链适用于固定流程
- 路由链实现条件分支
- 并行链提升独立任务的效率
- 统一状态字典是管理中间数据的核心
- 生产级部署需要错误处理、超时和监控

六、中间状态管理的最佳实践

6.1 状态设计原则

1. 最小化：只存储必要的数据
2. 类型安全：尽量使用有类型的数据结构
3. 可序列化：确保中间状态可以序列化和反序列化
4. 可追溯：记录状态变更的时间线

6.2 常见状态模式

单例模式：整个系统只有一个共享状态字典

分片模式：不同组件使用不同的状态空间

事件溯源：不修改状态，而是追加状态变更事件

6.3 状态序列化与持久化

import pickle
from datetime import datetime

def save_state(state, path):
    # 添加时间戳
    state["_saved_at"] = datetime.now().isoformat()
    with open(path, "wb") as f:
        pickle.dump(state, f)

def load_state(path):
    with open(path, "rb") as f:
        return pickle.load(f)

七、生产级链式调用的考虑

1. 错误处理：每个步骤都需要 try-catch
2. 超时控制：每个 LLM 调用设置超时
3. 重试策略：临时失败自动重试
4. 日志记录：记录每个步骤的输入和输出
5. 监控告警：监控链路完整性和延迟

八、总结

关键要点回顾：
- 顺序链适用于固定流程
- 路由链实现条件分支
- 并行链提升独立任务的效率
- 统一状态字典是管理中间数据的核心
- 生产级部署需要错误处理、超时和监控

延伸阅读

• 📺 B 站播放列表：Building Systems with ChatGPT API — ChatGPT API 系统构建
• 📚 更多学习资源，请访问 deeplearning.ai 官网