Agent 工程体系 · 第 43/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

OpenAI Agents SDK 会话工程：历史、Session、流式、取消和恢复

在 Agent 应用中，“会话”不是一个聊天窗口变量，而是一组需要被正确排序、持久化、裁剪、恢复和并发控制的状态。

一次普通的 Runner.run() 可能包含多个模型回合：

模型生成文本；
模型请求调用工具；
工具返回结果；
模型继续推理；
发生 handoff，切换到另一个 Agent；
触发 Guardrail 或人工审批；
最终输出答案。

因此，真正需要保存的并不只有用户消息和最终答案，还包括工具调用、工具结果、handoff 边界、审批状态，以及在中断时继续运行所需的内部状态。

OpenAI Agents SDK 默认使用 Responses API 作为 OpenAI 模型的底层接口，但在其上层负责 Agent loop、工具执行、handoff、Guardrail、Session 和 Trace 等运行时能力。(openai.github.io)

本文把以下几个容易混淆的概念分开：

历史：下一次模型调用实际可见的输入项；
Session：由 SDK 管理历史持久化的客户端记忆层；
流式：运行过程中的事件传输机制；
取消：停止当前运行的控制操作；
恢复：从 RunState 或 Session 中继续一个未完成的流程。

一、先建立正确的状态模型

1.1 历史不是“字符串列表”

在简单聊天中，历史通常被想象成：

[
    {"role": "user", "content": "你好"},
    {"role": "assistant", "content": "你好，有什么可以帮你？"},
]

但 Agent 运行的历史更接近 Responses API 的输入项序列：

[
  user message,
  assistant message,
  tool call,
  tool output,
  handoff call,
  handoff output,
  approval request,
  approval response,
  reasoning item,
  ...
]

可以形式化为：

H_t = [x_1, x_2, \ldots, x_n]

其中：

$H_t$ 表示时刻 $t$ 的会话历史；
$x_i$ 是一个输入项，而不一定是一条普通消息；
输入项之间的顺序具有语义；
工具调用和工具结果必须保持匹配关系；
某些 reasoning item 还可能需要和后续输出保持关联。

因此，下面两种“保存历史”的做法并不等价：

# 只保存最终文本
history.append({
    "role": "assistant",
    "content": result.final_output,
})

# 保存 SDK 生成的完整新项
for item in result.new_items:
    save(item)

第一种方式适合“每轮都是纯文本回答”的最小聊天应用，但不适合依赖工具、handoff、审批或恢复的 Agent。SDK 的结果对象提供了 new_items、raw_responses、last_agent、interruptions 等不同视角；当你需要保留工具、handoff 或审批边界时，应优先使用 new_items，而不是把历史压扁成最终文本。(openai.github.io)

1.2 一次 Agent run 内部是什么关系

设一次运行的输入为 $I_t$ ，当前 Agent 为 $A_t$ ，历史为 $H_t$ 。

模型实际收到的输入可以抽象为：

M_t = \operatorname{Merge}(H_t, I_t)

模型输出 $O_t$ 后，Runner 根据输出类型决定下一步：

\operatorname{Next}(O_t)= \begin{cases} \text{结束并返回最终输出}, & O_t \text{ 是最终文本且没有工具调用}\\ \text{执行工具并追加工具结果}, & O_t \text{ 包含工具调用}\\ \text{切换 Agent 并继续}, & O_t \text{ 包含 handoff}\\ \text{暂停等待审批}, & O_t \text{ 触发人工审批} \end{cases}

对应的运行循环是：

当前 Agent + 当前输入
        │
        ▼
     调用模型
        │
 ┌──────┼────────┬─────────┐
 ▼      ▼        ▼         ▼
最终输出 工具调用 handoff   审批
结束    执行工具  切换Agent 暂停
          │        │         │
          └────────┴─────────┘
                   │
                   ▼
               继续模型调用

Agents SDK 的 Runner 会在模型产生工具调用时执行工具、追加工具结果后重新调用模型；发生 handoff 时则更新当前 Agent 和输入后继续循环。达到 max_turns 时会抛出 MaxTurnsExceeded。(openai.github.io)

这里的关键点是：一次 run 不是一次模型请求。如果模型调用了三个工具，那么一次用户输入可能对应多个模型响应和多个工具轨迹。

二、历史、Session 和 Server-managed Conversation 的区别

2.1 手动历史

最底层的方式是应用自己保存和重建输入：

from agents import Agent, Runner

agent = Agent(
    name="Assistant",
    instructions="回答用户问题。",
)

conversation = []

result = await Runner.run(
    agent,
    conversation + [{"role": "user", "content": "北京今天适合跑步吗？"}],
)

conversation.extend(result.to_input_list())
print(result.final_output)

这里的 to_input_list() 目标是得到下一次模型调用可以使用的输入项列表。

这种方式的优点是完全可控：

可以自己决定保存哪些项；
可以自己实现摘要和裁剪；
可以把历史放进任意数据库；
可以在发送给模型前重排输入。

缺点也很明确：

需要自己防止重复追加；
需要处理工具调用和工具结果；
需要处理 handoff；
需要设计进程重启后的恢复；
需要自己处理并发更新。

手动历史适合需要完全掌控协议和存储结构的系统，但它把很多运行时责任交给了业务代码。

2.2 SDK Session

Session 是 Agents SDK 提供的客户端记忆层。使用 Session 后，Runner 会在每次运行前读取该 Session 的历史，并在运行结束后把本轮新增项写回 Session。(openai.github.io)

最小示例：

from agents import Agent, Runner, SQLiteSession

agent = Agent(
    name="Assistant",
    instructions="回答简洁、准确。",
)

session = SQLiteSession(
    "user_123",
    "conversations.db",
)

result = await Runner.run(
    agent,
    "金门大桥位于哪个城市？",
    session=session,
)
print(result.final_output)

result = await Runner.run(
    agent,
    "它位于哪个州？",
    session=session,
)
print(result.final_output)

第二次调用不需要手动传入第一次的输入和答案。Runner 会执行近似如下的数据流：

Session.get_items(...)
        │
        ▼
历史 H_t + 当前新输入 I_t
        │
        ▼
      Runner
        │
        ├── 模型调用
        ├── 工具调用
        ├── handoff
        └── Guardrail
        │
        ▼
Session.add_items(本轮新增项)

Session 的核心语义可以表示为：

H_{t+1} = H_t \mathbin{+\!\!+} N_t

其中：

$H_t$ 是运行前读取到的历史；
$N_t$ 是本轮产生的新输入、模型输出、工具调用和工具结果；
$+\!\!+$ 表示按顺序追加，而不是集合合并。

SDK 文档明确区分了“读取历史”和“保存本轮新项”：历史会在模型调用前加入输入，但旧历史不会因为被重新排序或过滤而再次作为新输入持久化。(openai.github.io)

2.3 Session 不是 RunState

这两个对象都和状态有关，但解决的问题不同。

对象	主要职责	生命周期
Session	保存跨多次 run 的会话历史	通常跨请求、跨进程
RunState	保存一次未完成 run 的可恢复执行状态	从暂停或取消点继续
Result	暴露一次 run 的输出、事件和状态	当前运行期间及结束后
Trace	记录观测、调试和评估信息	由追踪系统管理

Session 解决的是：

下一次新的用户回合，如何知道之前聊过什么？

RunState 解决的是：

这一次尚未完成的 Agent 流程，如何从中断点继续？

例如，用户问：

删除临时文件。

模型调用了一个需要审批的删除工具。此时会话历史可能已经包含：

用户请求
模型工具调用
审批请求

但“是否已批准”“当前停在哪个 Agent”“下一步应该执行什么”属于这一次 run 的执行状态，应由 RunState 保存。

三、Session 的历史合并、裁剪和持久化

3.1 默认合并规则

使用 Session 时，默认的模型输入近似为：

M_t = H_t \mathbin{+\!\!+} I_t

其中：

$H_t$ ：从 session.get_items(...) 读取的历史；
$I_t$ ：当前调用新传入的输入；
$M_t$ ：本次模型请求的输入。

可以通过 RunConfig.session_input_callback 修改合并逻辑：

from agents import Agent, RunConfig, Runner, SQLiteSession

def keep_recent_history(history, new_input):
    return history[-10:] + new_input

agent = Agent(name="Assistant")
session = SQLiteSession("conversation_123", "conversations.db")

result = await Runner.run(
    agent,
    "只根据最近的上下文回答。",
    session=session,
    run_config=RunConfig(
        session_input_callback=keep_recent_history
    ),
)

这个回调决定本轮发给模型的输入，但不会把旧历史重新保存成新的历史项。也就是说：

存储历史：H_t
模型输入：H_t[-10:] + I_t
本轮新增：N_t
写回存储：H_t + N_t

因此，输入裁剪和历史删除是两个不同动作：

session_input_callback：只影响当前模型看到了什么；
pop_item() 或 clear_session()：真正修改持久化历史。

如果把二者混淆，就会出现一种常见现象：当前回答看起来没有旧上下文，但下一轮旧上下文又重新出现。

3.2 按条目限制历史，不等于按 Token 限制

SDK 支持通过 SessionSettings(limit=N) 限制每次读取的历史项数量：

from agents import (
    Agent,
    RunConfig,
    Runner,
    SessionSettings,
    SQLiteSession,
)

agent = Agent(name="Assistant")
session = SQLiteSession("conversation_123", "conversations.db")

result = await Runner.run(
    agent,
    "总结最近的讨论。",
    session=session,
    run_config=RunConfig(
        session_settings=SessionSettings(limit=50)
    ),
)

limit=50 表示最多获取最近 50 个 Session item，而不是最多 50 个 Token。一个很长的工具输出可能远大于一条普通用户消息，因此条目数限制不能替代 Token 预算。

设模型上下文窗口为 $C$ ，系统提示、工具 schema、当前输入和历史分别占用：

S,\quad T,\quad I,\quad H

要避免超出上下文窗口，需要满足：

S + T + I + H \le C

因此历史预算应为：

H \le C - S - T - I - R

其中 $R$ 是为模型输出和推理保留的空间。

这解释了为什么以下策略可能失败：

history[-50:]

即使只保留 50 项，也可能因为某个工具返回了大段 JSON 而超过窗口。

3.3 三种不同的历史压缩策略

策略一：只取最近项

def recent_only(history, new_input):
    return history[-20:] + new_input

优点是简单、延迟低。

缺点是可能截断工具调用和工具结果的逻辑配对。例如只留下工具结果，却丢掉对应的工具调用，模型可能无法正确理解该结果。

策略二：按完整回合裁剪

假设应用自己给每个用户回合标记边界：

turn 1: user -> assistant
turn 2: user -> assistant -> tool call -> tool output -> assistant
turn 3: user -> assistant

裁剪时保留完整回合，而不是简单按 item 数量截断：

def keep_complete_turns(history, new_input, max_items=30):
    selected = []

    for item in reversed(history):
        selected.append(item)
        if len(selected) >= max_items:
            break

    selected.reverse()
    return selected + new_input

这仍然只是示意。生产实现应识别工具调用与工具结果的关联，而不是只按数组长度切割。

策略三：摘要或 Responses compaction

Agents SDK 提供 OpenAIResponsesCompactionSession，它在底层 Session 之上使用 Responses API 的 compaction 能力压缩长期历史。它可以自动判断是否触发压缩，也可以手动调用 run_compaction()。(openai.github.io)

示例：

from agents import Agent, Runner, SQLiteSession
from agents.memory import OpenAIResponsesCompactionSession

underlying = SQLiteSession(
    "conversation_123",
    "conversations.db",
)

session = OpenAIResponsesCompactionSession(
    session_id="conversation_123",
    underlying_session=underlying,
)

agent = Agent(name="Assistant")

result = await Runner.run(
    agent,
    "继续处理当前任务。",
    session=session,
)
print(result.final_output)

压缩的语义不是“删除任意旧消息”，而是把一段历史替换成更紧凑的表示。可以抽象为：

H' = \operatorname{Compact}(H)

理想情况下，压缩应保留对未来决策有用的信息：

\operatorname{Relevant}(H') \approx \operatorname{Relevant}(H)

但它不保证逐字保留原始历史，也不等价于审计日志。若系统需要完整审计，应同时保存不可变的原始事件记录。

自动压缩还会影响流式生命周期：最后一个可见文本 token 到达后，流式迭代器可能仍在等待压缩、Session 写入或审批状态收尾。文档特别指出，自动 compaction 可能让 stream_events() 在最后一个输出 token 之后继续保持打开。(openai.github.io)

3.4 Session 实现的选择

SDK 提供多种 Session 后端：

SQLiteSession：本地开发和简单应用；
AsyncSQLiteSession：使用 aiosqlite 的异步 SQLite；
RedisSession：多 Worker 或多服务共享；
SQLAlchemySession：已有关系型数据库；
MongoDBSession：已有 MongoDB 或需要横向扩展；
DaprSession：通过 Dapr 使用不同状态存储；
OpenAIConversationsSession：由 OpenAI Conversations API 管理历史；
OpenAIResponsesCompactionSession：在其他 Session 之上提供压缩；
EncryptedSession：在已有 Session 之上提供加密和 TTL。(openai.github.io)

一个重要的边界是：同一次运行中，SDK Session 不能和 conversation_id、previous_response_id 或 auto_previous_response_id 叠加使用。它们代表两套不同的会话延续机制。(openai.github.io)

四、三种会话延续方式如何取舍

4.1 Session：客户端管理历史

session = SQLiteSession("thread_001", "conversations.db")

await Runner.run(agent, "第一轮", session=session)
await Runner.run(agent, "第二轮", session=session)

应用拥有历史存储，SDK 负责读取和写回。

适合：

需要自定义存储；
需要查询、删除、分支和审计；
需要把会话绑定到业务实体；
需要在本地或私有数据库中保存历史。

4.2 `conversation_id`：服务端管理命名会话

服务端会话通常使用一个可共享的 conversation 资源标识。它适合多个请求、服务或组件共同引用同一个会话。

其特点是：

应用保存 conversation_id
       │
       ▼
每次请求引用同一服务端会话
       │
       ▼
服务端负责维护会话历史

这与本地 Session 的责任边界不同：本地 Session 把历史放在应用控制的后端，conversation_id 则把会话资源交给 OpenAI 服务端管理。

4.3 `previous_response_id`：轻量的响应链延续

previous_response_id 表示把下一次 Responses API 请求接在上一条响应之后。Agents SDK 的结果对象提供 last_response_id，可用于下一轮继续该响应链。(openai.github.io)

抽象为：

R_{t+1} = \operatorname{Continue}(R_t, I_{t+1})

它适合简单的连续调用，但应用不应把它误认为完整的本地历史数据库。若需要任意查询、修改、分支或跨系统共享，通常需要显式 Session 或服务端 conversation 资源。

conversation_id 和 previous_response_id 互斥：前者适合命名和共享会话，后者是更轻量的响应链延续。(openai.github.io)

五、流式：不是“把最终答案切成几段”

5.1 `run_streamed()` 的返回值

Agents SDK 提供三个主要 Runner 入口：

await Runner.run(...)
Runner.run_sync(...)
Runner.run_streamed(...)

其中 run_streamed() 返回 RunResultStreaming，通过 result.stream_events() 获取异步事件流。(openai.github.io)

最小的文本流式示例：

import asyncio

from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner


async def main():
    agent = Agent(
        name="Joker",
        instructions="你是一个有帮助的助手。",
    )

    result = Runner.run_streamed(
        agent,
        input="请讲五个笑话。",
    )

    async for event in result.stream_events():
        if (
            event.type == "raw_response_event"
            and isinstance(event.data, ResponseTextDeltaEvent)
        ):
            print(event.data.delta, end="", flush=True)

    print()
    print("is_complete =", result.is_complete)


if __name__ == "__main__":
    asyncio.run(main())

这里的 ResponseTextDeltaEvent 是底层 Responses API 的文本增量事件。原始事件适合把文本尽快显示给用户；更高层的事件则适合观察 Agent、工具和 handoff 的运行过程。(openai.github.io)

5.2 三层事件视角

可以把流式事件分为三层：

第一层：原始模型事件

例如：

response.created
response.output_text.delta
response.completed

优点是延迟低、信息接近模型协议。

缺点是业务代码需要理解 Responses API 事件结构。

第二层：Run item 事件

这类事件反映 Agent 运行项，例如：

message_output_created
tool_called
tool_output
handoff_requested
handoff_occurred

它更适合 UI 展示：

助手正在思考
正在查询订单
工具返回结果
正在交给退款专家

第三层：Agent 事件

这类事件强调当前 Agent 的生命周期，例如当前 Agent 改变、Agent 运行开始或结束。

工程上通常会同时使用：

原始文本事件：实时显示回答；
高层工具事件：显示进度；
Result 对象：在流结束后获取完整结果。

5.3 流结束不等于最后一个 token

下面这种写法是不完整的：

async for event in result.stream_events():
    if is_text_delta(event):
        send_to_browser(event)

# 误以为这里不需要进一步处理

正确理解是：

最后一个文本 token
        │
        ▼
可能仍有：
- 工具状态收尾
- Session 持久化
- 审批状态更新
- 自动 compaction
- Trace 结束
        │
        ▼
stream_events() 迭代器结束
        │
        ▼
run 才真正完成

SDK 文档明确要求持续消费 stream_events()，直到异步迭代器结束；只有到此时，流式运行才算完成，result.is_complete 才能反映最终状态。(openai.github.io)

因此，Web 服务不能在收到最后一个文本 token 后立即释放运行资源或认为 Session 已经写入完成。

六、取消：立即停止和当前回合完成后停止

6.1 立即取消

result.cancel()

默认会尽快停止当前流式运行。

这适合：

用户点击“停止生成”；
HTTP 客户端断开连接；
服务端超时；
用户已经不再需要当前答案。

但“取消”不是事务回滚。若模型请求或工具调用已经到达外部系统，取消本地 Runner 不一定能撤销外部副作用。

例如：

模型请求发送成功
        │
        ▼
模型要求调用 charge_card
        │
        ▼
工具已经向支付服务发起扣款
        │
        ▼
用户点击取消

此时取消 Agent 运行不能自动把支付操作回滚。具有副作用的工具必须设计幂等键、状态查询和补偿逻辑。

6.2 `after_turn` 取消

如果希望当前回合完整结束后再停止：

result.cancel(mode="after_turn")

这里的“回合”不是整个用户请求，也不一定等于最终输出。一次 Agent run 可能包含多个模型回合和工具回合。

可以有如下过程：

Turn 1：模型请求查询订单
Turn 2：工具返回订单
Turn 3：模型准备生成答案

如果在工具调用后使用 after_turn，运行可能在当前工具回合结束后停止，但整个用户问题还没有最终回答。

SDK 文档因此建议：如果 cancel(mode="after_turn") 在工具回合后停止，不要立即把结果当作“新用户回合”追加；应使用规范化输入继续当前未完成的用户回合。(openai.github.io)

6.3 一个可运行的取消示例

import asyncio

from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner


async def main():
    agent = Agent(
        name="Assistant",
        instructions="写一篇很长的技术说明。",
    )

    result = Runner.run_streamed(
        agent,
        "解释分布式系统中的一致性问题。",
    )

    async def consume():
        async for event in result.stream_events():
            if (
                event.type == "raw_response_event"
                and isinstance(event.data, ResponseTextDeltaEvent)
            ):
                print(event.data.delta, end="", flush=True)

    consumer = asyncio.create_task(consume())

    await asyncio.sleep(2)

    # 方案 A：立即停止
    result.cancel()

    try:
        await consumer
    except asyncio.CancelledError:
        print("\nconsumer task cancelled")

    print("\nis_complete =", result.is_complete)


if __name__ == "__main__":
    asyncio.run(main())

实际服务中，取消通常由 HTTP 断开、前端按钮或超时任务触发。应保证：

先调用 result.cancel()；
再等待 stream_events() 完成或捕获取消异常；
不要在流尚未排空时直接读取最终状态；
不要把已显示给用户的部分文本当作完整答案写入最终消息。

6.4 “取消后重新发送同一个问题”为什么可能重复

假设用户发来：

帮我生成报告。

模型已经调用了工具，随后客户端断开。服务端重连后直接执行：

await Runner.run(agent, "帮我生成报告。", session=session)

如果 Session 已经保存了原始用户输入和部分工具轨迹，这可能导致：

旧的用户输入
旧的工具调用
新的一次相同用户输入
新的工具调用

结果可能是重复执行工具。

更稳妥的恢复路径是：

若只是完整回合后的下一轮：使用 Session 开始新的用户输入；
若当前 run 尚未完成：从 RunState 恢复；
若必须手动继续：使用 to_input_list(mode="normalized")，并根据当前 Agent 继续未完成流程；
对外部副作用工具使用幂等键。

七、恢复：`RunState` 是一次 run 的暂停边界

7.1 什么是 RunState

RunState 是可序列化的 Agent 运行快照，包含恢复一次运行所需的信息，例如：

当前 Agent；
已产生的模型响应；
已生成的运行项；
工具调用和工具结果；
审批状态；
使用量；
可选的服务端会话标识；
等待继续处理的输入。

SDK 将 RunState 定义为适用于 human-in-the-loop 的持久化暂停/恢复边界。(openai.github.io)

它不是简单的消息数组：

消息历史 = “发生过什么”
RunState = “发生过什么 + 当前执行到哪里 + 下一步怎么接着跑”

7.2 工具审批暂停和恢复

下面是完整的审批恢复结构：

from agents import Agent, Runner

agent = Agent(
    name="Assistant",
    instructions="需要时使用工具。",
)

result = await Runner.run(
    agent,
    "删除不再需要的临时文件。",
)

if result.interruptions:
    state = result.to_state()

    for interruption in result.interruptions:
        state.approve(interruption)

    result = await Runner.run(
        agent,
        state,
    )

print(result.final_output)

状态变化如下：

用户请求
  │
  ▼
模型生成工具调用
  │
  ▼
工具需要审批
  │
  ▼
result.interruptions
  │
  ▼
result.to_state()
  │
  ▼
approve / reject
  │
  ▼
Runner.run(agent, state)
  │
  ▼
继续原来的 Agent 流程

这里不能把审批暂停当成普通的新用户输入。因为待审批的工具调用已经属于当前 run 的一部分；恢复时应继续该状态，而不是再次发送原始问题。

7.3 流式审批恢复

流式运行中，必须先排空事件流，再检查中断：

from agents import Agent, Runner

agent = Agent(
    name="Assistant",
    instructions="危险操作需要审批。",
)

result = Runner.run_streamed(
    agent,
    "删除临时文件。",
)

async for event in result.stream_events():
    # 在这里处理文本、工具事件和进度事件
    pass

if result.interruptions:
    state = result.to_state()

    for interruption in result.interruptions:
        state.approve(interruption)

    resumed = Runner.run_streamed(agent, state)

    async for event in resumed.stream_events():
        pass

SDK 文档明确指出：流式运行遇到工具审批时，stream_events() 会先结束，然后中断信息才可从 result.interruptions 读取；应把结果转成 RunState 后恢复。(openai.github.io)

7.4 序列化到数据库或队列

可以把状态转成 JSON：

state = result.to_state()
payload = state.to_json()

进程重启后恢复：

from agents import Runner

state = await RunState.from_json(
    initial_agent=agent,
    state_json=payload,
)

result = await Runner.run(
    agent,
    state,
)

如果通过字符串保存：

payload = state.to_string()

state = await RunState.from_string(
    initial_agent=agent,
    state_string=payload,
)

恢复时需要注意 Agent map 和 context 的重建。RunState.from_json() / from_string() 支持 context_override、context_deserializer 和 strict_context；如果上下文对象不是天然可 JSON 序列化的，就必须明确指定如何恢复，不能假设 Python 对象会自动完整重建。(openai.github.io)

一个典型的持久化表可以是：

CREATE TABLE agent_run_states (
    run_id TEXT PRIMARY KEY,
    session_id TEXT NOT NULL,
    state_json TEXT NOT NULL,
    status TEXT NOT NULL,
    version INTEGER NOT NULL,
    updated_at TIMESTAMP NOT NULL
);

更新时要使用版本号避免两个 Worker 同时恢复同一个状态：

UPDATE agent_run_states
SET state_json = :state_json,
    status = :status,
    version = version + 1,
    updated_at = CURRENT_TIMESTAMP
WHERE run_id = :run_id
  AND version = :expected_version;

如果影响行数为 0，说明状态已被其他 Worker 修改，当前恢复操作不能继续覆盖。

八、在暂停期间追加新的用户输入

有时 Agent 暂停等待审批，但用户又补充了一句话：

另外，生成报告后保存到项目目录。

此时可以将新输入暂存到 RunState：

state = result.to_state()

state.add_input(
    "另外，生成报告后保存到项目目录。"
)

for interruption in state.get_interruptions():
    state.approve(interruption)

result = await Runner.run(
    agent,
    state,
)

add_input() 不会立即发起模型请求，而是把输入放进待处理队列。恢复时，Runner 会在下一次模型调用前接纳这些输入。

多次调用会保持顺序：

state.add_input("先检查文件列表。")
state.add_input("再删除临时文件。")

模型下一次看到的新增输入顺序仍然是：

先检查文件列表。
再删除临时文件。

SDK 文档还指出，暂存输入会被序列化到 RunState，因此经过 to_json() / from_json() 或字符串序列化后仍可恢复。(openai.github.io)

但不是所有状态都能追加输入。以下情况通常应结束当前 run，再开启新的用户回合：

状态已经是终态；
没有剩余模型回合；
模型响应已经接受但仍在等待本地处理；
当前中断对应的工具结果可能直接结束整个运行。

九、取消后的两种恢复路径

9.1 当前 run 尚未完成：恢复 RunState

当 after_turn 取消让当前流程停在中间状态时：

result.cancel(mode="after_turn")

async for _event in result.stream_events():
    pass

state = result.to_state()

resumed = await Runner.run(
    result.last_agent,
    state,
)

核心思想是：

取消 ≠ 新回合
取消 = 当前 run 暂停在某个合法边界

9.2 当前回合已经完成：开始新回合

如果运行已经完成，用户下一次提问就是新的输入：

result = await Runner.run(
    agent,
    "继续解释刚才的第二个例子。",
    session=session,
)

此时 Session 会提供之前已经持久化的历史。

区分两者的判断依据不是“浏览器有没有收到完整文本”，而是：

stream_events() 是否已经结束；
result.is_complete 是否为真；
是否存在 interruptions；
RunState 是否还有未完成的模型步骤。

十、Session 并发：同一个会话不能随意并行写

Session 的历史本质上是一个有序日志：

H = [x_1, x_2, \ldots, x_n]

如果两个请求同时对同一个 Session 写入：

请求 A：用户说“查订单”
请求 B：用户说“取消订单”

可能产生任意顺序：

查订单 -> 取消订单

或：

取消订单 -> 查订单

如果“取消订单”依赖“查订单”的结果，那么顺序就不再是展示问题，而是业务正确性问题。

最简单的会话级并发控制是为每个 session_id 加锁：

import asyncio
from collections import defaultdict

session_locks = defaultdict(asyncio.Lock)

async def run_turn(session_id, session, agent, user_input):
    async with session_locks[session_id]:
        return await Runner.run(
            agent,
            user_input,
            session=session,
        )

这段代码只适用于单进程。多 Worker 部署必须使用分布式锁、数据库行锁、队列串行化或具有并发控制能力的 Session 后端。

此外，还要把“流式输出锁”和“Session 写入锁”分开考虑：

用户可以看到 A 请求的输出；
A 请求结束前，不能让 B 请求覆盖同一会话的历史；
客户端断开时，A 请求要被取消或进入后台；
Worker 崩溃时，B 请求需要知道 A 是否留下了可恢复的 RunState。

十一、错误路径和重复执行

11.1 模型请求失败

from agents import Agent, Runner

try:
    result = await Runner.run(
        agent,
        "查询订单状态。",
        session=session,
    )
except Exception as exc:
    logger.exception("agent run failed", extra={
        "session_id": session_id,
    })
    raise

失败后不能简单地假设 Session 一定没有写入任何内容。具体写入边界取决于运行阶段、后端和重试路径，因此应检查：

items = await session.get_items()

重点关注：

用户输入是否已持久化；
工具调用是否已持久化；
工具结果是否已持久化；
是否存在重复项；
是否留下了需要恢复的状态。

11.2 `conversation_locked`

服务端管理会话时，多个请求可能同时访问同一 conversation。SDK 对 conversation_locked 会进行带退避的自动重试，并在重试前回退内部会话跟踪器中的输入，避免把同一批准备好的项重复追加。(openai.github.io)

但这不是业务层无限重试保证。生产系统仍应：

限制重试次数；
记录 request ID、run ID 和 session ID；
区分锁冲突与模型请求失败；
避免多个前端请求同时操作同一逻辑回合。

11.3 工具副作用和不安全重放

即使 SDK 在自己的状态中努力保持一个 InputItem 的唯一出现，也不等于外部服务只会执行一次。文档明确区分了 SDK 内部的一次性输入保证和 Provider 或外部工具的交付保证；如果请求可能已经到达 Provider，允许不安全重放可能导致服务端工作重复。(openai.github.io)

对副作用工具，应显式设计：

async def create_refund(order_id: str, idempotency_key: str):
    ...

幂等键可以由：

K = \operatorname{hash}(session\_id, run\_id, tool\_call\_id)

生成。

这样即使因为网络超时、Worker 重启或恢复逻辑导致相同工具调用再次抵达业务服务，也能根据 $K$ 返回已处理结果，而不是重复扣款、退款或删除。

十二、一个端到端的服务端骨架

下面的示例把 Session、流式、取消和 RunState 持久化边界放在一起：

import asyncio
import json
from dataclasses import dataclass

from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner, SQLiteSession


@dataclass
class RunRecord:
    run_id: str
    session_id: str
    state_json: str | None = None
    status: str = "running"


agent = Agent(
    name="SupportAgent",
    instructions=(
        "你是客服 Agent。需要外部操作时使用工具；"
        "回答前先确认事实，不要猜测。"
    ),
)


async def stream_turn(
    *,
    run_id: str,
    session_id: str,
    user_input: str,
    stop_event: asyncio.Event,
):
    session = SQLiteSession(
        session_id,
        "conversations.db",
    )

    record = RunRecord(
        run_id=run_id,
        session_id=session_id,
    )

    result = Runner.run_streamed(
        agent,
        user_input,
        session=session,
    )

    async def consume_events():
        async for event in result.stream_events():
            if (
                event.type == "raw_response_event"
                and isinstance(event.data, ResponseTextDeltaEvent)
            ):
                yield event.data.delta

    consumer = consume_events()

    try:
        async for text in consumer:
            # 实际服务中应写入 SSE、WebSocket 或消息队列
            print(text, end="", flush=True)

            if stop_event.is_set():
                result.cancel(mode="after_turn")

        # 必须继续消费到迭代器结束，等待 Session 等收尾逻辑
        if result.interruptions:
            state = result.to_state()
            record.state_json = state.to_json()
            record.status = "waiting_for_approval"

        elif result.is_complete:
            record.status = "completed"

        return record

    except asyncio.CancelledError:
        result.cancel()
        record.status = "cancelled"

        # 是否保存 RunState 取决于当前状态是否可恢复
        try:
            await consumer.aclose()
        except Exception:
            pass

        raise

这个骨架省略了具体 Web 框架，但体现了几个重要边界：

run_streamed() 返回的是运行对象，不是普通字符串；
输出文本只是事件流的一部分；
必须消费到流结束，不能只看到最后一个 token；
审批中断要保存 RunState；
客户端取消不代表外部工具副作用已经回滚；
运行记录和 Session 历史需要分开保存。

实际生产实现还应加入：

run_id 和 tool_call_id；
Session 级并发控制；
状态版本号；
工具幂等键；
超时和重试策略；
Trace 关联；
敏感数据脱敏；
断线重连后的恢复策略。

十三、常见误解和对应失败表现

误解一：Session 就是“自动摘要”

不是。

普通 Session 主要负责历史读取和写回。它不会自动知道哪些信息重要，也不会自动把十万字对话变成高质量摘要。需要控制上下文时，应使用回调、手动摘要、compaction 或独立的长期记忆系统。

误解二：保存 `final_output` 就保存了 Agent 记忆

不是。

如果中间发生了工具调用、handoff 或审批，仅保存 final_output 会丢失 Agent 继续工作所需的结构化轨迹。

误解三：流式收到最后一个 token 就结束了

不是。

Session 持久化、审批状态和 compaction 可能在最后一个 token 后继续执行。真正的完成边界是 stream_events() 迭代器结束。(openai.github.io)

误解四：取消等于回滚

不是。

取消主要停止 SDK 侧的运行。已经发出的模型请求和已经执行的外部工具不会因为 Python 协程取消而自动撤销。

误解五：恢复就是重新发送原问题

通常不是。

重新发送原问题可能造成重复用户输入、重复工具调用和重复副作用。未完成的 run 应优先从 RunState 恢复。

误解六：`SessionSettings(limit=50)` 能保证不超 Token

不能。

它限制的是历史 item 数，不是 token 数。工具输出、长文本和多模态输入仍可能使上下文超限。

误解七：一个 Session 可以被多个请求随意并行使用

不应这样假设。

并行写入会改变历史顺序；在多 Worker 环境下，还可能造成重复追加、版本覆盖和恢复竞争。

十四、如何诊断一次异常会话

遇到“模型记错上下文”“工具重复执行”或“恢复后回答跳跃”时，应按以下顺序检查。

第一步：检查 Session 原始项

items = await session.get_items()

for index, item in enumerate(items):
    print(index, item)

不要只打印最终文本，重点查看：

用户输入是否重复；
工具调用是否有匹配结果；
handoff 是否成对出现；
审批请求是否残留；
裁剪是否留下了不完整的工具链。

第二步：检查本轮新增项

for item in result.new_items:
    print(type(item).__name__, item)

这可以区分：

问题在历史读取；
问题在本轮 Agent loop；
问题在 Session 写回；
问题在恢复后再次追加。

第三步：检查当前 Agent

print("last_agent =", result.last_agent)

发生 handoff 后，继续下一轮时通常应关注最后运行的 Agent，而不是无条件重新使用最初 Agent。流式结果中的 current_agent 还会随着 handoff 更新。(openai.github.io)

第四步：检查是否把审批当成终态

if result.interruptions:
    print("run is waiting for approval")

有 interruptions 时，运行不是普通完成，也不能立即新建一个相同用户回合。

第五步：检查 RunState 是否可恢复

state = result.to_state()
print(state.pending_input)
print(state.get_interruptions())

如果状态已经终止、没有剩余模型回合，或当前工具结果可能直接结束运行，就不应继续向该状态追加输入。(openai.github.io)

十五、会话工程的核心边界

一个可靠的 Agents SDK 会话系统，至少要维护四条边界：

历史边界

历史是有序的 Responses input item 序列，不是最终答案字符串。

持久化边界

Session 保存跨 run 的会话历史；保存时应保留本轮新增项，而不是把裁剪后的旧历史重新追加。

流式边界

用户可见的最后一个 token 不是运行完成边界；必须等 stream_events() 结束。

恢复边界

RunState 用于恢复一次未完成 run；新的用户消息只有在合法状态下才能通过 add_input() 暂存，并在恢复前被接纳。

可以把完整生命周期概括为：

新用户输入
    │
    ▼
读取 Session 历史
    │
    ▼
Runner 执行 Agent loop
    │
    ├── 流式输出事件
    ├── 工具调用
    ├── handoff
    ├── Guardrail
    └── 审批中断
            │
            ▼
       保存 RunState
            │
      ┌─────┴─────┐
      ▼           ▼
   审批后恢复    用户取消
      │           │
      └─────┬─────┘
            ▼
       继续或结束 run
            │
            ▼
       持久化本轮历史

当应用只需要多轮聊天时，Session 已经足够；当应用需要人工审批、断点续跑、进程重启恢复或长时间任务时，必须把 Session 和 RunState 同时纳入架构；当应用还涉及流式和外部副作用时，则需要额外处理流结束、取消语义、幂等执行和并发顺序。

这也是 Agent 会话工程与普通聊天记录的根本区别：聊天记录回答“用户和模型说过什么”，会话工程还必须回答“当前运行到哪里、哪些操作已经发生、下一步能否安全继续”。

系列导航与关联阅读

官方资料

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。