7.3 上下文传递与状态管理

在提示词链中，如何有效地在步骤之间传递信息和管理状态是关键的技术挑战。本节将介绍相关的策略和最佳实践。

7.3.1 上下文传递的挑战

• 信息丢失：每个步骤都是独立的模型调用，如果不主动传递，前序步骤的信息会丢失

• 信息冗余：传递过多信息会导致 Token 浪费和注意力稀释

• 格式不匹配：前后步骤的数据格式需要兼容

7.3.2 传递策略

策略一：全量传递

将前序步骤的完整输出传递给后续步骤。

步骤 2 的输入：
{
  "previous_output": "[步骤 1 的完整输出]",
  "current_task": "[步骤 2 的任务描述]"
}

优点：信息完整 缺点： Token 消耗大，可能包含无关信息

策略二：摘要传递

提取前序步骤输出的关键信息传递。

步骤 2 的输入：
{
  "step1_summary": "[步骤 1 输出的摘要]",
  "current_task": "[步骤 2 的任务描述]"
}

优点：节省 Token ，聚焦关键信息 缺点：可能丢失某些细节

策略三：结构化传递

使用结构化格式传递必要信息。

{
  "metadata": {
    "doc_id": "12345",
    "source": "step1"
  },
  "key_findings": [
    "发现 1",
    "发现 2"
  ],
  "data": {
    "field1": "value1",
    "field2": "value2"
  }
}

优点：结构清晰，易于解析 缺点：需要预先定义格式

7.3.3 状态管理模式

模式一：无状态模式

每个步骤完全独立，所需信息全部通过输入传递。

步骤 N 的所有输入 = f(步骤 N-1 的输出)

适用：简单的线性流程

模式二：累积状态模式

维护一个累积的状态对象，每个步骤可以读取和更新。

result1 = {"summary": "（示例）步骤 1 的输出结果"}

state = {
    "original_input": "...",
    "step1_result": None,
    "step2_result": None,
    "errors": []
}

# 步骤 1 执行后

state["step1_result"] = result1

# 步骤 2 可以访问

step2_input = f"""
原始输入：{state["original_input"]}
步骤 1 结果：{state["step1_result"]}
当前任务：...
"""

print(step2_input.strip())

适用：需要访问历史信息的流程

模式三：外部存储模式

将状态存储在外部系统（数据库、缓存等），步骤通过标识符访问。

class InMemoryStorage:
    def __init__(self):
        self._data = {}

    def save(self, session_id, key, value):
        self._data[(session_id, key)] = value

    def load(self, session_id, key):
        return self._data[(session_id, key)]

storage = InMemoryStorage()
session_id = "demo-session"
result1 = {"summary": "（示例）步骤 1 的输出结果"}

# 步骤 1 完成后

storage.save(session_id, "step1_result", result1)

# 步骤 2 需要时

step1_result = storage.load(session_id, "step1_result")

print(step1_result)

适用：长时间运行或可恢复的流程

7.3.4 实现最佳实践

定义清晰的接口

为每个步骤定义明确的输入输出格式：

# 步骤接口定义

class Step1:
    input_schema = {
        "text": str,
        "config": dict
    }
    output_schema = {
        "entities": list,
        "sentiment": str,
        "confidence": float
    }

使用模板构建提示词

def build_step2_prompt(step1_output, step2_config):
    return f"""

## 上一步结果

{json.dumps(step1_output, ensure_ascii=False)}

## 当前任务

{step2_config['task_description']}

## 输出要求

{step2_config['output_format']}
"""

错误处理

def execute_chain(steps, initial_input):
    current_input = initial_input
    results = []

    for i, step in enumerate(steps):
        try:
            output = step.execute(current_input)
            results.append({
                "step": i + 1,
                "status": "success",
                "output": output
            })
            current_input = output
        except Exception as e:
            results.append({
                "step": i + 1,
                "status": "error",
                "error": str(e)
            })
            # 决定是否继续
            if step.critical:
                break

    return results

7.3.5 上下文压缩技术

当需要传递大量信息时，可以使用压缩技术：

摘要压缩

步骤：总结压缩

输入：[1000 字的详细分析]

输出：[200 字的核心摘要]

将摘要传递给后续步骤

关键信息提取

从以下文本中提取关键信息：
- 最重要的 3 个发现
- 关键数据点
- 主要结论

[原始文本]

结构化压缩

将非结构化文本转换为结构化格式：

输入：大段文字描述
输出：
{
  "main_topic": "主题",
  "key_points": ["要点 1", "要点 2"],
  "action_items": ["行动 1"]
}

想一想

1. 随着链条变长，前面步骤的上下文信息可能丢失或走形。你会用什么策略来保持信息的完整传递？

2. 在多步骤链中，“全量传递上下文”和“只传递摘要”各有什么利弊？什么时候该用哪种？