Skip to content

护栏执行时机、流式事件与运行追踪:练习答案 ​

参考实现 ​

完整参考:

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')

用心记录每一步成长