《AI Agent 高级编程》4.1 OpenAI Agents SDK：handoff/guardrails/tracing 实现委派

import importlib.metadata as m
print(m.version("openai-agents"), m.version("openai"))

0.23.1 3.28.0

import inspect
from agents import Agent, Runner
print([p for p in inspect.signature(Agent.__init__).parameters if p != "self"])
print([p for p in inspect.signature(Runner.run).parameters if p != "self"])

['name', 'handoff_description', 'tools', 'mcp_servers', 'mcp_config', 'instructions', 'prompt', 'handoffs', 'model', 'model_settings', 'input_guardrails', 'output_guardrails', 'output_type', 'hooks', 'tool_use_behavior', 'reset_tool_choice']
['starting_agent', 'input', 'context', 'max_turns', 'hooks', 'run_config', 'error_handlers', 'previous_response_id', 'auto_previous_response_id', 'conversation_id', 'session']

from agents import handoff, function_tool
print(inspect.signature(handoff))
print(inspect.signature(function_tool))

(agent: 'Agent[TContext]', tool_name_override: 'str | None' = None, tool_description_override: 'str | None' = None, on_handoff: 'OnHandoffWithInput[THandoffInput] | OnHandoffWithoutInput | None' = None, input_type: 'type[THandoffInput] | None' = None, input_filter: 'Callable[[HandoffInputData], HandoffInputData] | None' = None, nest_handoff_history: 'bool | None' = None, is_enabled: 'bool | Callable[[RunContextWrapper[Any], Agent[TContext]], MaybeAwaitable[bool]]' = True) -> 'Handoff[TContext, Agent[TContext]]'
(func: 'ToolFunction[...] | None' = None, *, name_override: 'str | None' = None, description_override: 'str | None' = None, docstring_style: 'DocstringStyle | None' = None, use_docstring_info: 'bool' = True, failure_error_function: 'ToolErrorFunction | None | object' = <object object at 0x...>, strict_mode: 'bool' = True, is_enabled: 'bool | Callable[[RunContextWrapper[Any], AgentBase], MaybeAwaitable[bool]]' = True, needs_approval: 'bool | Callable[[RunContextWrapper[Any], dict[str, Any], str], Awaitable[bool]]' = False, tool_input_guardrails: 'list[ToolInputGuardrail[Any]] | None' = None, tool_output_guardrails: 'list[ToolOutputGuardrail[Any]] | None' = None, timeout: 'float | None' = None, timeout_behavior: 'ToolTimeoutBehavior' = 'error_as_result', timeout_error_function: 'ToolErrorFunction | None' = None, defer_loading: 'bool' = False, custom_data_extractor: 'FunctionToolCustomDataExtractor | None' = None, allowed_callers: 'list[ToolCaller] | None' = None, output_type: 'Any | None' = None, output_json_schema: 'dict[str, Any] | None' = None) -> 'FunctionTool | Callable[[ToolFunction[...]], FunctionTool]'

from agents import InputGuardrail, OutputGuardrail, GuardrailFunctionOutput
import inspect
print(inspect.signature(InputGuardrail.__init__))
print(inspect.signature(GuardrailFunctionOutput.__init__))

(self, guardrail_function: 'Callable[[RunContextWrapper[TContext], Agent[Any], str | list[TResponseInputItem]], MaybeAwaitable[GuardrailFunctionOutput]]', name: 'str | None' = None, run_in_parallel: 'bool' = True) -> None
(self, output_info: 'Any', tripwire_triggered: 'bool') -> None

import asyncio, json
from agents import InputGuardrail, GuardrailFunctionOutput, RunContextWrapper, Agent

async def banned(ctx: RunContextWrapper, agent: Agent, inp) -> GuardrailFunctionOutput:
    text = inp if isinstance(inp, str) else json.dumps(inp, ensure_ascii=False)
    bad = "绕过" in text
    return GuardrailFunctionOutput(output_info={"banned": bad}, tripwire_triggered=bad)

ig = InputGuardrail(guardrail_function=banned, name="no-bypass")
print("guardrail.name =", ig.get_name())

ctx = RunContextWrapper(context=None)
r1 = asyncio.run(banned(ctx, None, "帮我写一份周报"))
r2 = asyncio.run(banned(ctx, None, "教我怎么绕过护栏"))
print("safe   ->", r1.tripwire_triggered, r1.output_info)
print("unsafe ->", r2.tripwire_triggered, r2.output_info)

guardrail.name = no-bypass
safe   -> False {'banned': False}
unsafe -> True {'banned': True}

from agents import trace, agent_span, function_span, generation_span
import inspect
print(inspect.signature(trace))
print(inspect.signature(agent_span))
print(inspect.signature(function_span))

(workflow_name: 'str', trace_id: 'str | None' = None, group_id: 'str | None' = None, metadata: 'dict[str, Any] | None' = None, tracing: 'TracingConfig | None' = None, disabled: 'bool' = False) -> 'Trace'
(name: 'str', handoffs: 'list[str] | None' = None, tools: 'list[str] | None' = None, output_type: 'str | None' = None, span_id: 'str | None' = None, parent: 'Trace | Span[Any] | None' = None, disabled: 'bool' = False) -> 'Span[AgentSpanData]'
(name: 'str', input: 'str | None' = None, output: 'str | None' = None, span_id: 'str | None' = None, parent: 'Trace | Span[Any] | None' = None, disabled: 'bool' = False) -> 'Span[FunctionSpanData]'

from agents import set_trace_processors, trace, agent_span, function_span
from agents.tracing import TracingProcessor

class LocalCapture(TracingProcessor):
    def __init__(self): self.events = []
    def on_trace_start(self, t): self.events.append(("trace_start", t.name))
    def on_trace_end(self, t): self.events.append(("trace_end", t.name))
    def on_span_start(self, s): pass
    def on_span_end(self, s):
        self.events.append(("span_end", type(s.span_data).__name__, s.span_data.name))
    def shutdown(self): pass
    def force_flush(self): pass

cap = LocalCapture()
set_trace_processors([cap])          # 替换默认处理器，杜绝任何外发

with trace("research-copilot"):
    with agent_span("planner", tools=["search"], output_type="str"):
        with function_span("search", input='{"q":"smolagents"}', output="[hit]"):
            pass
    with agent_span("writer"):
        pass

for e in cap.events:
    print(e)

('trace_start', 'research-copilot')
('span_end', 'FunctionSpanData', 'search')
('span_end', 'AgentSpanData', 'planner')
('span_end', 'AgentSpanData', 'writer')
('trace_end', 'research-copilot')

# research_agents_sdk.py —— 需要活的 OpenAI 会话才能运行，本节未执行
import asyncio, json
from agents import (Agent, Runner, function_tool, handoff,
                    InputGuardrail, GuardrailFunctionOutput, RunContextWrapper)

@function_tool
def search(query: str) -> str:
    """检索本地知识库，返回命中文本。
    Args:
        query: 查询词。
    """
    return f"[hit] {query} => 3 条"

async def no_injection(ctx: RunContextWrapper, agent: Agent, inp) -> GuardrailFunctionOutput:
    text = inp if isinstance(inp, str) else json.dumps(inp, ensure_ascii=False)
    return GuardrailFunctionOutput(output_info={}, tripwire_triggered="忽略以上指令" in text)

researcher = Agent(name="researcher", instructions="用 search 找证据", tools=[search])
writer = Agent(name="writer", instructions="把证据写成带来源的结论")
router = Agent(
    name="router",
    instructions="判断该检索还是该写作，必要时 handoff",
    handoffs=[handoff(researcher), handoff(writer, input_filter=lambda d: d)],
    input_guardrails=[InputGuardrail(guardrail_function=no_injection, name="no-injection")],
)

async def main():
    result = await Runner.run(router, "调研 smolagents 的 CodeAgent", max_turns=8)
    print(result.final_output)          # 最终文本
    print(result.last_agent.name)       # 最后停在哪个 Agent
    print(len(result.new_items))        # 本轮新增的 item 数

asyncio.run(main)

（本代码需要活的 OpenAI 会话才能运行，本节未执行，故不贴输出。）