第 6 章：上下文与会话记忆

系列教程：OpenAI Agents SDK 从入门到实战

本章目标：搞懂 RunContext（依赖注入）和 Session（跨轮记忆），让你的 Agent 既能访问业务数据，又能记住聊过的话。

为什么需要上下文和记忆？

Agent 默认是"金鱼记忆"。每次调用 Runner.run()，它都从零开始，既不记得之前聊过什么，也拿不到你应用里的任何数据。

这带来两个问题：

拿不到业务数据：工具函数需要访问数据库连接、当前用户信息、API 客户端等，但 Agent 不知道这些东西在哪
记不住对话历史：问完"我叫小明"，再问"我叫什么"，AI 一脸茫然

SDK 用两套机制分别解决这两个问题：

问题	解决方案	类比
工具函数需要外部依赖	RunContext	像 FastAPI 的依赖注入
跨轮对话需要记忆	Session	像浏览器的 cookie/session

它们各司其职，互不冲突，也可以一起用。

RunContext：给 Agent 注入运行时依赖

问题场景

假设你在做一个航空客服系统，工具函数需要知道当前乘客是谁、航班号是什么。你不能把这些硬编码进去，因为每个用户不一样。

笨办法是用全局变量，但那太丑了，也不安全。RunContext 的思路是：定义一个上下文对象，传给 Runner.run()，工具函数里自动就能拿到。

定义 Context 类型

Context 可以是任何类型，最常用的是 Pydantic BaseModel 或 dataclass：

from pydantic import BaseModel

class CustomerContext(BaseModel):
    """客服系统的运行时上下文"""
    customer_id: str
    customer_name: str
    vip_level: int = 0
    db_connection: object = None  # 数据库连接之类的也可以塞进来

    class Config:
        arbitrary_types_allowed = True  # 允许非标准类型

用 dataclass 也行：

from dataclasses import dataclass

@dataclass
class CustomerContext:
    customer_id: str
    customer_name: str
    vip_level: int = 0

在工具函数中访问 Context

关键角色是 RunContextWrapper。工具函数的第一个参数如果声明为 RunContextWrapper[YourContext]，SDK 会自动注入：

from agents import RunContextWrapper, function_tool

@function_tool
async def check_order(
    context: RunContextWrapper[CustomerContext],  # SDK 自动注入，不算工具参数
    order_id: str,                                # 这个才是 LLM 填的参数
) -> str:
    """查询订单状态"""
    # 通过 context.context 访问你的自定义上下文
    customer = context.context
    # 这里可以用 customer.db_connection 查数据库
    return f"客户 {customer.customer_name} 的订单 {order_id} 状态：已发货"

注意：context 参数对 LLM 是不可见的，LLM 不知道它的存在，也不需要填它。LLM 只看到 order_id 这个参数。

Agent 的泛型标注

用 Agent[YourContext] 告诉类型检查器，这个 Agent 用的是哪种 Context：

from agents import Agent

agent = Agent[CustomerContext](
    name="客服助手",
    instructions="你是一个客服助手，帮助客户查询订单。",
    tools=[check_order],
)

这个泛型标注主要是给 IDE 和 mypy 用的，让它们能正确做类型推断。不写也能运行，但写了会更安全。

运行时传入 Context

from agents import Runner

# 创建上下文实例，填入当前用户的信息
ctx = CustomerContext(
    customer_id="C001",
    customer_name="张三",
    vip_level=2,
)

# 通过 context 参数传给 Runner.run()
result = await Runner.run(agent, "帮我查一下订单 ORD-12345", context=ctx)

整个流程就是：你创建 Context -> 传给 Runner -> SDK 包成 RunContextWrapper -> 工具函数里通过 context.context 拿到。

完整可运行示例：航空客服

import asyncio
import os

from openai import AsyncOpenAI
from pydantic import BaseModel

from agents import Agent, Runner, RunContextWrapper, OpenAIChatCompletionsModel, set_tracing_disabled, function_tool

# --- 模型配置 ---
set_tracing_disabled(True)
client = AsyncOpenAI(
    base_url=os.getenv("MODEL_BASE_URL", "http://localhost:8317/v1"),
    api_key=os.getenv("MODEL_API_KEY", "sk-12345678"),
)
model = OpenAIChatCompletionsModel(
    model=os.getenv("MODEL_NAME", "gpt-5.2"),
    openai_client=client,
)

# ========== 1. 定义上下文 ==========
class AirlineContext(BaseModel):
    """航空客服的运行时上下文"""
    passenger_name: str | None = None
    confirmation_number: str | None = None
    seat_number: str | None = None
    flight_number: str | None = None

# ========== 2. 定义工具，通过 context 访问数据 ==========
@function_tool
async def get_booking_info(
    context: RunContextWrapper[AirlineContext],
    confirmation_number: str,
) -> str:
    """根据确认号查询预订信息"""
    # 把确认号存到上下文里，后续工具可以复用
    context.context.confirmation_number = confirmation_number
    # 实际项目中这里会查数据库
    return (
        f"预订信息：确认号 {confirmation_number}，"
        f"乘客 {context.context.passenger_name or '未知'}，"
        f"航班 CA1234，座位 18A"
    )

@function_tool
async def change_seat(
    context: RunContextWrapper[AirlineContext],
    new_seat: str,
) -> str:
    """更换座位"""
    old_seat = context.context.seat_number or "18A"
    context.context.seat_number = new_seat
    return f"座位已从 {old_seat} 更换为 {new_seat}"

# ========== 3. 创建 Agent ==========
agent = Agent[AirlineContext](
    name="航空客服",
    instructions=(
        "你是航空公司客服助手。"
        "帮助乘客查询预订信息和更换座位。"
        "回答要简洁专业。"
    ),
    tools=[get_booking_info, change_seat],
    model=model,
)

# ========== 4. 运行 ==========
async def main():
    # 创建上下文，注入当前乘客信息
    ctx = AirlineContext(passenger_name="李明")

    # 第一次调用
    result = await Runner.run(
        agent,
        "帮我查一下确认号 ABC123 的预订信息",
        context=ctx,
    )
    print(f"客服: {result.final_output}")
    print(f"上下文中的确认号: {ctx.confirmation_number}")  # 被工具更新了

    # 第二次调用（注意：这里没有 Session，Agent 不记得上一轮对话）
    result = await Runner.run(
        agent,
        "帮我把座位换到 22B",
        context=ctx,  # 但上下文还在，工具能拿到
    )
    print(f"客服: {result.final_output}")
    print(f"上下文中的座位: {ctx.seat_number}")

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

Context 是跨工具调用共享的 -- 一个工具往里写数据，另一个工具能读到。但它不等于对话记忆，Agent 本身并不知道 Context 里有什么，Context 是给你的代码用的。

Session：跨轮对话记忆

问题场景

没有 Session 的 Agent 就像得了失忆症：

await Runner.run(agent, "我叫小明")
result = await Runner.run(agent, "我叫什么？")
print(result.final_output)  # "你还没告诉我你叫什么呢"

每次 Runner.run() 都是全新的一轮，AI 看不到之前的对话。

SQLiteSession：最简单的本地持久化

SDK 内置了 SQLiteSession，用 SQLite 存储对话历史：

from agents import SQLiteSession

# session_id：对话的唯一标识，不同 ID 就是不同的对话
# db_path：数据库文件路径，默认 ":memory:"（内存，程序退出就没了）
session = SQLiteSession("chat-001", db_path="conversations.db")

然后把它传给 Runner.run()：

result = await Runner.run(agent, "我叫小明", session=session)
result = await Runner.run(agent, "我叫什么？", session=session)
print(result.final_output)  # "你叫小明"

就这样。传了同一个 session，SDK 会自动在每次 run 之前加载历史，run 结束后保存新内容。

Session 的工作原理

Runner.run(agent, input="你好", session=session)

内部发生了什么：
1. session.get_items()    -- 从数据库加载历史对话
2. 把历史 + 当前输入拼在一起发给模型
3. 模型返回结果
4. session.add_items()    -- 把本轮的输入和输出存起来

下次 run 的时候，第 1 步就能拿到之前所有记录。你不用操心任何细节。

完整示例：有记忆的多轮对话

import asyncio
import os

from openai import AsyncOpenAI

from agents import Agent, Runner, SQLiteSession, OpenAIChatCompletionsModel, set_tracing_disabled

# --- 模型配置 ---
set_tracing_disabled(True)
client = AsyncOpenAI(
    base_url=os.getenv("MODEL_BASE_URL", "http://localhost:8317/v1"),
    api_key=os.getenv("MODEL_API_KEY", "sk-12345678"),
)
model = OpenAIChatCompletionsModel(
    model=os.getenv("MODEL_NAME", "gpt-5.2"),
    openai_client=client,
)

agent = Agent(
    name="记忆助手",
    instructions="你是一个有记忆的AI助手，回答要简洁。",
    model=model,
)

async def main():
    # 创建 Session，数据存到文件里
    session = SQLiteSession("demo-chat", db_path="memory.db")

    # 第一轮：告诉 AI 一些信息
    result = await Runner.run(
        agent,
        "我叫小明，我是一名 Python 程序员，住在北京",
        session=session,
    )
    print(f"AI: {result.final_output}\n")

    # 第二轮：AI 应该记住名字
    result = await Runner.run(agent, "我叫什么？", session=session)
    print(f"AI: {result.final_output}\n")

    # 第三轮：AI 应该记住职业
    result = await Runner.run(agent, "我的职业是什么？", session=session)
    print(f"AI: {result.final_output}\n")

    # 第四轮：AI 应该记住城市
    result = await Runner.run(agent, "我住在哪里？", session=session)
    print(f"AI: {result.final_output}\n")

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

预期输出类似：

AI: 你好小明！Python 程序员在北京，不错的组合。有什么我能帮你的？

AI: 你叫小明。

AI: 你是一名 Python 程序员。

AI: 你住在北京。

而且因为用了文件数据库，哪怕程序退出了再重新运行（只要 session_id 一样），AI 照样能记住。

手动管理历史 vs Session

在 Session 出现之前，多轮对话靠的是 result.to_input_list() 手动拼接：

旧方式：手动拼接

import asyncio
import os

from openai import AsyncOpenAI

from agents import Agent, Runner, OpenAIChatCompletionsModel, set_tracing_disabled

# --- 模型配置（省略详细注释，参考第1章）---
set_tracing_disabled(True)
client = AsyncOpenAI(
    base_url=os.getenv("MODEL_BASE_URL", "http://localhost:8317/v1"),
    api_key=os.getenv("MODEL_API_KEY", "sk-12345678"),
)
model = OpenAIChatCompletionsModel(
    model=os.getenv("MODEL_NAME", "gpt-5.2"),
    openai_client=client,
)

agent = Agent(name="助手", instructions="回答要简洁。", model=model)

async def main():
    # 第一轮
    result = await Runner.run(agent, "我叫小明")
    print(f"AI: {result.final_output}")

    # 手动把历史拼成新的输入
    input_items = result.to_input_list()
    input_items.append({"role": "user", "content": "我叫什么？"})

    # 第二轮，带上完整历史
    result = await Runner.run(agent, input_items)
    print(f"AI: {result.final_output}")

    # 第三轮，继续拼接...
    input_items = result.to_input_list()
    input_items.append({"role": "user", "content": "再说一遍？"})
    result = await Runner.run(agent, input_items)
    print(f"AI: {result.final_output}")

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

新方式：Session

import asyncio
import os

from openai import AsyncOpenAI

from agents import Agent, Runner, SQLiteSession, OpenAIChatCompletionsModel, set_tracing_disabled

# --- 模型配置（省略详细注释，参考第1章）---
set_tracing_disabled(True)
client = AsyncOpenAI(
    base_url=os.getenv("MODEL_BASE_URL", "http://localhost:8317/v1"),
    api_key=os.getenv("MODEL_API_KEY", "sk-12345678"),
)
model = OpenAIChatCompletionsModel(
    model=os.getenv("MODEL_NAME", "gpt-5.2"),
    openai_client=client,
)

agent = Agent(name="助手", instructions="回答要简洁。", model=model)

async def main():
    session = SQLiteSession("my-chat")

    result = await Runner.run(agent, "我叫小明", session=session)
    print(f"AI: {result.final_output}")

    result = await Runner.run(agent, "我叫什么？", session=session)
    print(f"AI: {result.final_output}")

    result = await Runner.run(agent, "再说一遍？", session=session)
    print(f"AI: {result.final_output}")

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

对比

维度	手动 `to_input_list()`	Session
代码量	每轮都要拼接，容易写错	传个 session 就完事
持久化	没有，程序关了就丢了	可以存文件，跨进程保留
多用户	自己管理多个列表	不同 session_id 自动隔离
灵活性	完全自己控制	SDK 自动管理，也支持手动操作

to_input_list() 不是没用了，在一些需要精细控制输入的场景（比如你只想带部分历史），它依然有价值。但日常的多轮对话，Session 更省心。

完整实战：带 Context 和 Session 的聊天助手

把 RunContext 和 Session 组合起来，做一个既能访问业务数据、又有记忆的客服助手：

import asyncio
import os
from dataclasses import dataclass, field

from openai import AsyncOpenAI

from agents import Agent, Runner, RunContextWrapper, SQLiteSession, OpenAIChatCompletionsModel, set_tracing_disabled, function_tool

# --- 模型配置 ---
set_tracing_disabled(True)
client = AsyncOpenAI(
    base_url=os.getenv("MODEL_BASE_URL", "http://localhost:8317/v1"),
    api_key=os.getenv("MODEL_API_KEY", "sk-12345678"),
)
model = OpenAIChatCompletionsModel(
    model=os.getenv("MODEL_NAME", "gpt-5.2"),
    openai_client=client,
)

# ========== 上下文：存放业务数据 ==========
@dataclass
class ShopContext:
    """网店客服上下文"""
    user_id: str
    username: str
    # 模拟的订单数据
    orders: dict = field(default_factory=lambda: {
        "ORD-001": {"item": "机械键盘", "status": "已发货", "price": 399},
        "ORD-002": {"item": "无线鼠标", "status": "待发货", "price": 129},
    })

# ========== 工具：通过 context 访问业务数据 ==========
@function_tool
async def query_order(
    context: RunContextWrapper[ShopContext],
    order_id: str,
) -> str:
    """查询订单详情"""
    orders = context.context.orders
    if order_id in orders:
        order = orders[order_id]
        return (
            f"订单 {order_id}：{order['item']}，"
            f"价格 {order['price']} 元，状态：{order['status']}"
        )
    return f"未找到订单 {order_id}"

@function_tool
async def list_orders(
    context: RunContextWrapper[ShopContext],
) -> str:
    """列出当前用户的所有订单"""
    orders = context.context.orders
    if not orders:
        return "暂无订单"
    lines = []
    for oid, info in orders.items():
        lines.append(f"  {oid}: {info['item']} - {info['status']}")
    return f"{context.context.username} 的订单：\n" + "\n".join(lines)

# ========== Agent ==========
agent = Agent[ShopContext](
    name="网店客服",
    instructions=(
        "你是网店客服助手。可以帮用户查询订单信息。"
        "回答要简洁友好，使用中文。"
    ),
    tools=[query_order, list_orders],
    model=model,
)

# ========== 运行 ==========
async def main():
    # 上下文：当前用户的业务数据
    ctx = ShopContext(user_id="U001", username="小明")

    # Session：让 AI 记住对话历史
    session = SQLiteSession("shop-chat-001", db_path="shop.db")

    print("网店客服已上线，输入 q 退出\n")

    while True:
        user_input = input("你: ")
        if user_input.strip().lower() in ("q", "quit", "exit"):
            print("再见！")
            break

        result = await Runner.run(
            agent,
            user_input,
            context=ctx,      # 注入业务数据
            session=session,   # 保持对话记忆
        )
        print(f"客服: {result.final_output}\n")

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

这个示例里： - RunContext 让工具函数能访问到当前用户的订单数据 - Session 让 AI 记住之前聊过的内容

两者各管各的，互相不影响。

小结

本章讲了两件事：

RunContext（依赖注入）： - 定义一个 Context 类（BaseModel 或 dataclass） - 工具函数声明 RunContextWrapper[YourContext] 参数，SDK 自动注入 - Agent 用 Agent[YourContext] 泛型标注 - 运行时通过 Runner.run(..., context=ctx) 传入

Session（对话记忆）： - SQLiteSession(session_id, db_path=...) 创建会话 - Runner.run(..., session=session) 自动管理对话历史 - 同一个 session_id = 同一段对话，不同 session_id = 不同对话 - 文件数据库支持跨进程、跨重启的持久化

手动历史管理： - result.to_input_list() 是旧方式，需要精细控制时仍然有用 - 日常多轮对话直接用 Session 更省事

一句话概括：Context 是给你的代码传数据，Session 是给 AI 传记忆。

下一步预告

Agent 现在既能拿到业务数据，又有了对话记忆。但目前 Agent 的回复都是自由文本，如果我们需要它输出结构化的 JSON 数据呢？

下一章我们学习结构化输出（Structured Output），让 Agent 按照你定义的格式精确输出数据。

← 第5章流式输出第7章结构化输出 →