结构化结果、Handoff 与会话生命周期：练习答案

参考实现

完整参考：

import asyncio,tempfile,os
from agents import SQLiteSession
async def main():
    with tempfile.TemporaryDirectory() as d:
        path=os.path.join(d,"x.db")
        a,b=SQLiteSession("a",path),SQLiteSession("b",path)
        await a.add_items([{"role":"user","content":"1"},{"role":"assistant","content":"2"}])
        await b.add_items([{"role":"user","content":"3"}])
        assert len(await a.get_items()) == 2
        await a.clear_session()
        assert await a.get_items() == []
        assert len(await b.get_items()) == 1
        print("isolated deletion")
asyncio.run(main())

删除范围按会话键确定。跨租户键的生成与鉴权必须在调用存储前完成。

返回正文

交接和坏结构的真实 Runner 答案

独立文件运行。只捕获 SDK 的 ModelBehaviorError，不将缺密钥或其他异常算结构验证成功。

import json
from pydantic import BaseModel
from agents import Agent, Runner, RunConfig, Model, ModelResponse, handoff
from agents.usage import Usage
from openai.types.responses import ResponseFunctionToolCall, ResponseOutputMessage, ResponseOutputText

class Answer(BaseModel):
    summary: str
    source: str

seen=[]
missing_source=False
class ScriptedModel(Model):
    async def get_response(self, system_instructions, input, model_settings, tools, output_schema, handoffs, tracing, **kwargs):
        seen.append({'instructions':system_instructions,'input':input})
        if handoffs:
            item=ResponseFunctionToolCall(id='h1',call_id='c1',name=handoffs[0].tool_name,arguments='{}',type='function_call')
        else:
            assert 'secret' not in json.dumps(input)
            item=ResponseOutputMessage(id='m1',type='message',role='assistant',status='completed',
                content=[ResponseOutputText(type='output_text',annotations=[],text=json.dumps({'summary':'refund policy'} if missing_source else {'summary':'refund policy','source':'p1'}))])
        return ModelResponse(output=[item],usage=Usage(),response_id=None)
    async def stream_response(self,*args,**kwargs):
        raise NotImplementedError('this lab uses Runner.run_sync')
        yield

def filter_history(data):
    # 白名单交接摘要，清除原历史与本轮工具材料；实际摘要应由应用授权数据生成。
    return data.clone(input_history='Approved request: refund policy',pre_handoff_items=())
from agents.exceptions import ModelBehaviorError
for invalid in [False,True]:
    missing_source=invalid;seen.clear();model=ScriptedModel()
    specialist=Agent(name='Support',model=model,output_type=Answer)
    triage=Agent(name='Triage',model=model,handoffs=[handoff(specialist,input_filter=filter_history)])
    try:
        result=Runner.run_sync(triage,'secret one; secret two',max_turns=4,run_config=RunConfig(tracing_disabled=True))
    except ModelBehaviorError:
        assert invalid
    else:
        assert not invalid and result.last_agent.name=='Support'
        assert result.final_output.source=='p1'
        assert any(x.type=='handoff_output_item' for x in result.new_items)
    assert len(seen)==2 and 'secret' not in json.dumps(seen[1]['input'])
print('Support handoff, whitelist and missing source rejection passed')