护栏执行时机、流式事件与运行追踪:练习答案
参考实现
完整参考:
python
import asyncio
from agents import Agent,Runner,RunConfig,input_guardrail,GuardrailFunctionOutput,InputGuardrailTripwireTriggered
@input_guardrail(run_in_parallel=False)
async def guard(ctx,agent,input):
return GuardrailFunctionOutput(output_info=None,tripwire_triggered=isinstance(input,str) and len(input)>=8)
async def main():
try: await Runner.run(Agent(name="A",input_guardrails=[guard]),"12345678",run_config=RunConfig(tracing_disabled=True))
except InputGuardrailTripwireTriggered: print("boundary blocked")
else: raise AssertionError("boundary accepted")
asyncio.run(main())只验证拒绝路径;允许路径会调用真实模型,必须另有凭证、费用和集成验证。
输出护栏与本地轨迹答案
正文完整块保存为 guardrail_trace.py,本答案另存文件运行,不上传追踪。
python
from guardrail_trace import run,writes
for final,expected in [('unsafe final',True),('public receipt',False)]:
writes.clear()
denied,events=run(final)
assert denied is expected and writes==['receipt-1']
assert {'function','handoff','guardrail'} <= {e.get('type') for e in events}
assert 'internal-secret' not in str(events)
print('same prior effect, different final acceptance, safe local tracing')