OpenAI Agents SDK #10：99% 的开发者都搞错了——Context 到底传没传给 LLM？

RunContextWrapper：本地 Context 的载体

wrapper.usage 特别有用——它聚合了单次运行里所有 Agent 消耗的 token 总量，省去你自己累加的麻烦。

代码示例一：基础用法

from dataclasses import dataclass
from agents import Agent, Runner, function_tool
from agents.run_context import RunContextWrapper

# 1. 定义你的业务状态对象
@dataclass
class UserInfo:
    name: str
    uid: int
    is_premium: bool = False

# 2. 工具函数接收 RunContextWrapper[UserInfo]，获取本地 Context
@function_tool
def greet_user(wrapper: RunContextWrapper[UserInfo]) -> str:
    """向当前登录用户打招呼"""
    user = wrapper.context          # 拿到 UserInfo 实例
    tier = "高级会员" if user.is_premium else "普通用户"
    return f"你好，{user.name}！你当前是{tier}（UID: {user.uid}）"

# 3. Agent 标注泛型类型，绑定 UserInfo
agent: Agent[UserInfo] = Agent(
    name="PersonalAssistant",
    instructions="你是一个个人助理，可以通过工具获取用户信息来提供服务。",
    tools=[greet_user],
)

# 4. 运行时把 UserInfo 实例传给 context 参数
user_info = UserInfo(name="张三", uid=10086, is_premium=True)

result = Runner.run_sync(
    starting_agent=agent,
    input="请帮我打个招呼",
    context=user_info,              # 整次运行内所有工具都能访问
)

print(result.final_output)
# 示例输出：你好，张三！你当前是高级会员（UID: 10086）

# 5. 查看 token 消耗（wrapper.usage 聚合了全部 Agent）
# result.usage.total_tokens 可直接查看

ToolContext：工具级元数据的精准入口

代码示例二：多 Agent Handoff 下的 Context 共享

from dataclasses import dataclass, field
from typing import List
from agents import Agent, Runner, function_tool, handoff
from agents.run_context import RunContextWrapper

@dataclass
class CustomerServiceContext:
    customer_id: str
    visited_agents: List[str] = field(default_factory=list)  # 追踪流转路径
    resolved: bool = False

# 前台分流 Agent 工具：记录首次接触
@function_tool
def log_triage(wrapper: RunContextWrapper[CustomerServiceContext]) -> str:
    """记录分流信息"""
    ctx = wrapper.context
    ctx.visited_agents.append("triage")   # 直接修改 Context，下游 Agent 能看到
    return f"客户 {ctx.customer_id} 已记录，准备分流"

# 退款 Agent 工具：读取并更新 Context
@function_tool
def process_refund(wrapper: RunContextWrapper[CustomerServiceContext]) -> str:
    """处理退款"""
    ctx = wrapper.context
    ctx.visited_agents.append("refund")   # 追加到路径链
    ctx.resolved = True
    return f"退款已处理，服务路径：{' → '.join(ctx.visited_agents)}"

# 定义退款专项 Agent
refund_agent: Agent[CustomerServiceContext] = Agent(
    name="RefundAgent",
    instructions="你专门处理退款请求，使用 process_refund 工具完成操作。",
    tools=[process_refund],
)

# 前台分流 Agent，完成分流后 Handoff 给退款 Agent
triage_agent: Agent[CustomerServiceContext] = Agent(
    name="TriageAgent",
    instructions="你是前台客服，先记录分流信息，然后把退款请求转给退款专员。",
    tools=[log_triage],
    handoffs=[refund_agent],             # 声明可交接的目标 Agent
)

# 执行：Context 在 TriageAgent → RefundAgent 间自动流转
ctx = CustomerServiceContext(customer_id="C-2025-001")
result = Runner.run_sync(
    starting_agent=triage_agent,
    input="我要申请退款",
    context=ctx,
)

# RefundAgent 处理完后，ctx 里记录了完整路径
print(ctx.visited_agents)   # ['triage', 'refund']
print(ctx.resolved)         # True

Handoff 下的进阶：input_filter 与 on_handoff

from agents.handoffs import HandoffInputData
from agents import handoff

def slim_history_filter(data: HandoffInputData) -> HandoffInputData:
    """
    只保留最近 3 条消息，避免退款 Agent 接收过多无关上下文。
    data.run_context 里有 RunContextWrapper，可以读写 Context。
    """
    recent_items = list(data.new_items)[-3:]
    return data.clone(new_items=tuple(recent_items))

refund_handoff = handoff(
    agent=refund_agent,
    input_filter=slim_history_filter,       # 过滤传入历史
    on_handoff=lambda ctx, _: print(        # 交接时触发回调
        f"即将移交，当前路径：{ctx.context.visited_agents}"
    ),
)

「本地 Context 和 LLM-visible Context」的边界在哪里

1. instructions（静态或动态函数生成）
2. Runner.run(input=...) 传入的初始消息
3. 函数工具的返回值（工具把 Context 里的数据格式化成字符串返回）
4. 检索结果注入

护栏中的 Context 复用

@input_guardrail
async def policy_check(
    ctx: RunContextWrapper[CustomerServiceContext],   # 拿到同一个 Context
    agent: Agent,
    input: str,
) -> GuardrailFunctionOutput:
    # 嵌套调用验证 Agent，复用当前 Context
    result = await Runner.run(
        validation_agent,
        input=input,
        context=ctx.context,   # 透传，避免重新构建
    )
    triggered = "违规" in result.final_output
    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=triggered,
    )

3 条可立即落地的实践建议

预告 #11