护栏执行时机、流式事件与运行追踪
输入护栏如果与 Agent 并行运行，模型或工具可能在护栏完成之前已经开始。对于必须先拒绝的请求，使用阻塞护栏；无论输入还是输出护栏，都不能替代工具执行前授权。输出护栏检查最终输出，不能撤销之前的副作用。
不调用模型验证阻塞护栏
复用固定版本环境，完整实验。长度规则只演示 SDK tripwire，不是提示注入检测器。
import asyncio
from agents import (Agent, Runner, RunConfig, input_guardrail,
GuardrailFunctionOutput, InputGuardrailTripwireTriggered)
@input_guardrail(run_in_parallel=False)
async def size_limit(ctx, agent, input):
return GuardrailFunctionOutput(output_info={"rule":"length"},
tripwire_triggered=isinstance(input,str) and len(input)>8)
a = Agent(name="Bounded", input_guardrails=[size_limit])
async def main():
try:
await Runner.run(a,"x"*9,run_config=RunConfig(tracing_disabled=True))
except InputGuardrailTripwireTriggered:
print("blocked before model")
else:
raise AssertionError("guardrail not triggered")
asyncio.run(main())本例故意不配置密钥；阻塞护栏必须在模型调用前抛出指定异常。如果捕获任意 Exception，会把缺密钥错误误计为通过。真实内容安全需独立分类与人工策略，不能用字数或敏感词代表全部安全。
流式事件不是最终结果
在线服务中使用 Runner.run_streamed，遍历 stream_events()；原始响应事件承载文本增量，run item 事件承载工具等语义，agent updated 表示控制者变化。客户端可以展示增量，但最终发布等运行完成并校验 final_output。断连后任务是否继续由服务端策略决定；取消等待不保证外部工具撤销。
下面是在线入口，环境和费用条件同第一章，单独保存运行；本次未执行真实模型流。
import asyncio,os
from agents import Agent,Runner,RunConfig
async def main():
run = Runner.run_streamed(Agent(name="Brief",model=os.environ["OPENAI_MODEL"]),
"用一句话解释幂等。",max_turns=2,
run_config=RunConfig(tracing_disabled=True))
async for event in run.stream_events():
print(event.type)
print(run.final_output)
asyncio.run(main())Tracing 与部署责任
SDK trace 能关联 Agent、模型、工具、handoff 与护栏。真实数据可能进入追踪系统，采集前最小化和脱敏；本书实验禁用上传，不意味着生产不需要观测。业务 trace 还需 task ID、模型/提示版本、效果状态、用量、错误和人工决定。不要把本地 context 或全量工具结果默认序列化进日志。
审批绑定动作摘要与有效期，工具执行前重新授权。部署设置请求 deadline、并发池、任务队列、幂等与灰度；用固定样例测试拒答、越权、重复工具、预算耗尽和断连恢复。SDK 小版本演进可能增加审批/工具护栏能力，升级先读对应 release 与签名，不把当前在线文档的新参数直接复制到0.7.0。
用真实输出护栏与本地处理器观察时序
保存为 guardrail_trace.py，在本课 0.7.0 环境运行，无密钥。模型回放依次要求 handoff、执行工具、给出最终文本。最终接收者的 output_guardrail 检查最终文本；第一遍故意拒绝，仍能观察工具已经执行。第二遍合法返回用于对照。
import json
from agents import (Agent, Runner, RunConfig, Model, ModelResponse, function_tool,
handoff, output_guardrail, GuardrailFunctionOutput, OutputGuardrailTripwireTriggered,
set_trace_processors)
from agents.tracing import TracingProcessor
from agents.usage import Usage
from openai.types.responses import ResponseFunctionToolCall, ResponseOutputMessage, ResponseOutputText
class LocalTrace(TracingProcessor):
def __init__(self):self.events=[]
def on_trace_start(self,trace):self.events.append({'event':'run_start','trace_id':trace.trace_id})
def on_trace_end(self,trace):self.events.append({'event':'run_end','trace_id':trace.trace_id})
def on_span_start(self,span):pass
def on_span_end(self,span):
# 不序列化 span.export()；只保存受控关联字段和类型。
self.events.append({'event':'span_end','trace_id':span.trace_id,
'span_id':span.span_id,'parent_id':span.parent_id,
'type':span.span_data.type})
def shutdown(self):pass
def force_flush(self):pass
writes=[]
@function_tool
def record_receipt() -> str:
"""Record an authorized sample receipt."""
writes.append('receipt-1')
return 'internal-secret: receipt recorded'
@output_guardrail
async def final_policy(ctx,agent,output):
return GuardrailFunctionOutput(output_info={'rule':'public-format'},tripwire_triggered=output!='public receipt')
class ProtocolModel(Model):
def __init__(self,final):self.final=final;self.calls=0
async def get_response(self,system_instructions,input,model_settings,tools,output_schema,handoffs,tracing,**kwargs):
self.calls+=1
if handoffs:
item=ResponseFunctionToolCall(id='h1',call_id='handoff-1',name=handoffs[0].tool_name,arguments='{}',type='function_call')
elif self.calls==2:
item=ResponseFunctionToolCall(id='t1',call_id='tool-1',name='record_receipt',arguments='{}',type='function_call')
else:
item=ResponseOutputMessage(id='m1',type='message',role='assistant',status='completed',content=[ResponseOutputText(type='output_text',annotations=[],text=self.final)])
return ModelResponse(output=[item],usage=Usage(),response_id=None)
async def stream_response(self,*args,**kwargs):
raise NotImplementedError('non-streaming lab')
yield
def run(final):
local=LocalTrace()
set_trace_processors([local]) # 替换默认后台上传器；add_trace_processor 会保留上传器。
model=ProtocolModel(final)
worker=Agent(name='Receipts',model=model,tools=[record_receipt],output_guardrails=[final_policy])
router=Agent(name='Router',model=model,handoffs=[handoff(worker)])
denied=False
try:
result=Runner.run_sync(router,'internal-secret request',max_turns=4,
run_config=RunConfig(trace_include_sensitive_data=False,workflow_name='local-receipts'))
assert result.final_output=='public receipt'
except OutputGuardrailTripwireTriggered:denied=True
finally:local.force_flush()
kinds={e.get('type') for e in local.events}
assert {'agent','function','handoff','guardrail'} <= kinds
assert local.events[0]['event']=='run_start' and local.events[-1]['event']=='run_end'
assert len({e['trace_id'] for e in local.events})==1
assert 'internal-secret' not in json.dumps(local.events)
return denied,local.events
if __name__=='__main__':
denied,events=run('unsafe final')
assert denied and writes==['receipt-1'] # 拒绝最终输出不能撤销工具。
writes.clear()
denied,_=run('public receipt')
assert not denied and writes==['receipt-1']
print('output tripwire, prior effect, local correlated spans and safe projection passed')function、handoff、guardrail、agent span 共享一个 trace_id；parent_id 让工具关联到父阶段。回放 Model 没有供应商网络请求，因此不会生成真实远程 generation 耗时，用它无法评估 token 成本。LocalTrace 只保留类型与关联标识，未保存请求、输出、异常全文或本地 context；这些 ID 在真实系统仍需访问控制与保留期限。
set_trace_processors 替换默认处理器，add_trace_processor 则在默认上传之外再加一个。SDK 默认启用 tracing 且可能采集敏感输入输出，本例同时替换上传器并设 trace_include_sensitive_data=False，再做字段白名单；单独关闭敏感字段不是“所有元数据都已脱敏”的保证。处理器设置是进程全局的，应在服务启动时统一配置，不能像小实验一样每个并发请求改一次。本地记录器用于单进程验证，生产处理器需要并发安全、有界队列和 flush/shutdown 策略。
拒绝 unsafe final 只阻止最终发布，writes 中已经有 receipt-1。凡必须在副作用前拒绝的业务限制，都应放在工具执行前授权/审批；用输出护栏做退款权限检查已经太晚。官方护栏执行流程和Tracing 处理器分别解释这两个边界，实验把它们放到同一执行链观察。
练习
输出与追踪变体：分别运行合法/非法最终文本，核对 tripwire 与工具次数；本地轨迹必须包含 handoff/function/guardrail 且不含 internal-secret。解释为什么异常不是撤销操作。
将拒绝边界从长度>8改为>=8，输入恰好8个字符应触发同一指定异常，并保持tracing_disabled，不接触模型。