结构化结果、Handoff 与会话生命周期
最终结果的 schema 可以限定形状，不能保证事实。handoff 可以改变当前执行 Agent，不能提升用户权限；SQLiteSession 可以保存历史，不能替代业务任务检查点。这三种状态要分别观察。
定义结构与交接
复用上一章环境，完整离线实验：
from pydantic import BaseModel, ValidationError
from agents import Agent
class Answer(BaseModel):
count: int
source: str
specialist = Agent(name="Inventory", instructions="Return verified stock only.", output_type=Answer)
triage = Agent(name="Triage", instructions="Inventory questions go to Inventory.", handoffs=[specialist])
assert triage.handoffs[0] is specialist
assert Answer.model_validate({"count":7,"source":"stock:p1"}).count == 7
try: Answer.model_validate({"count":"many","source":"stock:p1"})
except ValidationError: print("bad structure rejected")
else: raise AssertionError("invalid count accepted")把 specialist 放入 handoffs 只登记可交接对象；真实选择需要 Runner 调用模型。交接默认可能携带历史，敏感内容应在应用构造输入时最小化，并按 SDK 的输入过滤机制裁剪。Agent.as_tool 返回子 Agent 结果后由父 Agent继续；handoff 是接收者继续控制后续流程，两者的收尾责任不同。
动态 instructions 可以根据本地角色提供用语和业务上下文，但身份与权限判断仍在工具层。输出 source 字段必须对应真实工具证据，不能只检查它是字符串。长上下文保留来源、未解决问题和任务状态，避免摘要把 pending 变成 completed。
会话存储的真实离线实验
独立保存运行，不调用模型。SQLiteSession 不传 db_path 时使用内存；本例显式使用临时文件，验证重建对象后仍读取到数据，并清理资源。
import asyncio, tempfile, os
from agents import SQLiteSession
async def main():
with tempfile.TemporaryDirectory() as directory:
path = os.path.join(directory,"session.db")
first = SQLiteSession("tenant-a:user-7",path)
await first.add_items([{"role":"user","content":"只读查询 p1"}])
second = SQLiteSession("tenant-a:user-7",path)
assert len(await second.get_items()) == 1
other = SQLiteSession("tenant-b:user-7",path)
assert await other.get_items() == []
await second.clear_session()
assert await first.get_items() == []
print("persisted and cleared")
asyncio.run(main())session ID 是存储键，不是权限。服务端验证用户身份后生成键，清除会话还需处理 trace、缓存、导出与备份的保留规则。多个并发请求写同会话可能出现语义顺序冲突，应用需串行化或明确分支。
旧版代码常手动拼 result.to_input_list() 传下一轮；维护时先检查是否重复追加历史，再比较 session 自动管理。不要同时把完整旧历史作为新输入又传 session，否则上下文可能重复。保存会话不等于保存外部效果与审批状态，任务恢复需要业务账本。
Runner实际转移控制权与历史过滤
handoffs登记目标并不执行转移。模型必须输出handoff工具调用，Runner识别后把当前Agent换成目标，再以过滤后的历史调用目标模型。最终输出类型属于最后运行的Agent。
下面按本课固定SDK版本实现Model接口，用确定性协议响应穿过真实Runner。第一轮返回handoff调用，第二轮返回JSON；输出schema由Runner验证成Answer，不是预先手工构造Answer假装执行成功。
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=[]
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='{"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=())
model=ScriptedModel()
specialist=Agent(name='Billing',instructions='billing specialist',model=model,output_type=Answer)
triage=Agent(name='Triage',instructions='route request',model=model,
handoffs=[handoff(specialist,input_filter=filter_history)])
result=Runner.run_sync(triage,'secret internal note; refund question',run_config=RunConfig(tracing_disabled=True),max_turns=4)
assert result.last_agent.name=='Billing'
assert isinstance(result.final_output,Answer) and result.final_output.source=='p1'
assert len(seen)==2 and 'Approved request: refund policy' in json.dumps(seen[1]['input'])
assert any(item.type=='handoff_output_item' for item in result.new_items)
print(result.last_agent.name,result.final_output.model_dump())输出Billing及结构化字段。seen保存模型真正收到的输入，过滤后不存在secret；handoff_output_item证明发生了转移。input_history只改变原输入，pre_handoff_items是此前运行产物，new_items包含本轮转移记录；应分别决定保留范围。例中保留不含敏感内容的转移工具事件，清除旧历史。真实应用需对每类工具结果做最小化白名单，不能只删除关键词。
真实模型使用相同两个Agent和handoff对象，将model替换为 os.environ["OPENAI_MODEL"]，保留output_type与filter；再执行相同Runner入口。远程模型是否选择正确接收方是另一项评估，本轮没有调用。handoff是控制权转移，若希望调用专家后回到主Agent，应使用专家作为工具的组合。
练习
交接变体：将专家名改为Support，再验证last_agent与handoff事件；把原输入增加另一段秘密，目标模型仍只能见到白名单摘要。将目标模型JSON缺少source，观察Runner拒绝无效最终结构。
在临时SQLite数据库中向session A加入两条消息，session B加入一条；清空A后，B仍有一条。使用真实SQLiteSession运行。