第 2 课

链式调用与状态管理

Building Systems with ChatGPT API — ChatGPT API 系统构建

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

链式调用与状态管理

课程简介

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

🎬 本课程视频：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 状态管理的最佳实践

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

四、错误处理

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 状态设计原则

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

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)

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

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

八、总结

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

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

6.1 状态设计原则

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

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)

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

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

八、总结

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

6.1 状态设计原则

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

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)

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

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

链式调用与状态管理

链式调用与状态管理

课程简介

多步链式调用与状态管理

一、从单步到多步

二、三种基本调用模式

2.1 顺序链

2.2 路由链

2.3 并行链

三、中间状态管理

3.1 统一状态字典

3.2 状态管理的最佳实践

四、错误处理

4.1 每步独立捕获

4.2 降级策略

五、流式输出

5.1 边生成边推送

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

七、总结

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

6.1 状态设计原则

6.2 常见状态模式

6.3 状态序列化与持久化

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

八、总结

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

6.1 状态设计原则

6.2 常见状态模式

6.3 状态序列化与持久化

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

八、总结

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

6.1 状态设计原则

6.2 常见状态模式

6.3 状态序列化与持久化

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

八、总结

延伸阅读