OpenAI Agents SDK #13：每次多轮对话都要手写 `.to_input_list()`？Sessions 帮你彻底告别这个坑

# 第一轮
result1 = await Runner.run(agent, "你好")
# 第二轮，手动拼接历史
result2 = await Runner.run(agent, result1.to_input_list() + [{"role": "user", "content": "继续"}])
# 第三轮，再拼一次……
result3 = await Runner.run(agent, result2.to_input_list() + [{"role": "user", "content": "你刚才说的那个，展开讲讲"}])

工作原理：Runner 内部发生了什么

快速上手：SQLiteSession

import asyncio
from agents import Agent, Runner
from agents.extensions.memory import SQLiteSession

async def main():
    agent = Agent(
        name="chat_agent",
        instructions="你是一个乐于助人的助手。"
    )

# 用同一个 session_id 跨轮共享历史
    session = SQLiteSession(
        session_id="user_123",
        db_path="./chat_history.db"  # 也可以传 ":memory:" 用内存数据库
    )

# 第一轮——直接传字符串，不需要 to_input_list()
    result1 = await Runner.run(agent, "我叫张三，记住我的名字", session=session)
    print(result1.final_output)

# 第二轮——历史自动携带，模型能记住上文
    result2 = await Runner.run(agent, "我叫什么名字？", session=session)
    print(result2.final_output)  # 会正确回答「张三」

asyncio.run(main())

Session 后端全景：10 种选择

高级控制：三个调优手段

1. SessionSettings(limit=N) — 限制历史长度

from agents.extensions.memory import SQLiteSession, SessionSettings
from agents import RunConfig

session = SQLiteSession("user_123", "./chat.db")

# 只携带最近 10 条历史
config = RunConfig(session_settings=SessionSettings(limit=10))

result = await Runner.run(agent, "继续上次的话题", session=session, run_config=config)

2. session_input_callback — 自定义历史合并逻辑

from agents import RunConfig
from agents.items import TResponseInputItem

def my_history_merger(
    history: list[TResponseInputItem],
    new_input: list[TResponseInputItem]
) -> list[TResponseInputItem]:
    # 示例：只保留最近 5 轮，并在历史前插入系统摘要
    recent_history = history[-10:] if len(history) > 10 else history
    summary_message = [{
        "role": "system",
        "content": "以下是对话历史摘要：用户正在咨询产品 A 的退款问题。"
    }]
    return summary_message + recent_history + new_input

config = RunConfig(session_input_callback=my_history_merger)
result = await Runner.run(agent, "我刚才问了什么？", session=session, run_config=config)

3. pop_item() — 撤销上一条消息

# 用户认为上一轮回答有误，需要重新回答
last_item = await session.pop_item()
print(f"已撤销: {last_item}")

# 重新提问（不携带被撤销的那条）
result = await Runner.run(agent, "刚才的问题我换个说法", session=session)

自定义 Session：实现 SessionABC

from typing import Optional
from agents.extensions.memory import SessionABC
from agents.items import TResponseInputItem

class RedisClusterSession(SessionABC):
    def __init__(self, session_id: str, redis_client):
        self.session_id = session_id
        self.redis = redis_client
        self._key = f"session:{session_id}:items"

async def get_items(self, limit: Optional[int] = None) -> list[TResponseInputItem]:
        """从 Redis 拉取历史，limit 控制返回条数"""
        raw_items = await self.redis.lrange(self._key, 0, -1)
        items = [json.loads(i) for i in raw_items]
        return items[-limit:] if limit else items

async def add_items(self, items: list[TResponseInputItem]) -> None:
        """追加新条目到 Redis list"""
        pipeline = self.redis.pipeline()
        for item in items:
            pipeline.rpush(self._key, json.dumps(item))
        await pipeline.execute()

async def pop_item(self) -> Optional[TResponseInputItem]:
        """删除并返回最后一条"""
        raw = await self.redis.rpop(self._key)
        return json.loads(raw) if raw else None

async def clear_session(self) -> None:
        """清空整个会话"""
        await self.redis.delete(self._key)

与 Human in the Loop 结合

from agents import Runner, RunConfig, RunState

# 第一阶段：Agent 跑到某个检查点，需要人类确认
result = await Runner.run(
    agent,
    "帮我给 production 数据库执行这段 SQL",
    session=session
)

# 检查是否触发了人工审批
if result.interrupted:
    # 将 run_state 序列化，等待人类审批
    pending_state: RunState = result.run_state
    # 存到数据库、消息队列等……
    await save_pending_approval(pending_state)
    return  # 先返回，等审批

# 第二阶段：审批通过后，从 RunState 恢复
pending_state = await load_pending_approval()
final_result = await Runner.run(
    agent,
    input=None,  # 从 RunState 恢复，无需重传输入
    session=session,
    run_state=pending_state
)

实践建议

小结