OpenAI Agents Python SDK 源码剖析报告

graph TD
    A[用户输入] --> B[Runner 运行器]
    B --> C[Agent 代理]
    C --> D[Model 模型调用]
    C --> E[Tools 工具集]
    C --> F[Handoffs 代理切换]
    C --> G[Guardrails 安全防护]
    
    D --> H[OpenAI Responses API]
    D --> I[OpenAI Chat Completions API] 
    D --> J[LiteLLM 其他模型]
    
    E --> K[Function Tools 函数工具]
    E --> L[MCP Tools MCP工具]
    E --> M[Built-in Tools 内置工具]
    
    F --> N[Target Agent 目标代理]
    
    G --> O[Input Guardrails 输入防护]
    G --> P[Output Guardrails 输出防护]
    
    C --> Q[Session 会话管理]
    Q --> R[SQLite Session SQL会话]
    Q --> S[Redis Session Redis会话]
    Q --> T[Custom Session 自定义会话]
    
    B --> U[Tracing 追踪]
    U --> V[内置追踪处理器]
    U --> W[外部追踪服务]
    
    style A fill:#e1f5fe
    style B fill:#f3e5f5
    style C fill:#e8f5e8
    style D fill:#fff3e0
    style E fill:#fce4ec
    style F fill:#f1f8e9
    style G fill:#fff8e1

sequenceDiagram
    participant User as 用户
    participant Runner as Runner运行器
    participant Agent as Agent代理
    participant Model as Model模型
    participant Tools as Tools工具
    participant Session as Session会话
    participant Tracing as Tracing追踪
    
    User->>Runner: 输入查询
    Runner->>Session: 获取会话历史
    Session-->>Runner: 返回历史消息
    
    Runner->>Tracing: 开始追踪
    Runner->>Agent: 启动代理
    
    loop 代理执行循环
        Agent->>Model: 调用模型
        Model-->>Agent: 返回响应
        
        alt 需要工具调用
            Agent->>Tools: 执行工具
            Tools-->>Agent: 返回工具结果
        else 需要代理切换
            Agent->>Agent: 切换到新代理
        else 生成最终输出
            Agent-->>Runner: 返回最终结果
        end
    end
    
    Runner->>Session: 保存会话
    Runner->>Tracing: 结束追踪
    Runner-->>User: 返回结果

class Runner:
    @classmethod
    async def run(
        cls,
        starting_agent: Agent[TContext],
        input: str | list[TResponseInputItem],
        *,
        context: TContext | None = None,
        max_turns: int = DEFAULT_MAX_TURNS,
        hooks: RunHooks[TContext] | None = None,
        run_config: RunConfig | None = None,
        # ... 其他参数
    ) -> RunResult:
        """执行代理工作流的异步方法"""
        
    @classmethod
    def run_sync(cls, ...) -> RunResult:
        """执行代理工作流的同步方法"""
        
    @classmethod 
    def run_streamed(cls, ...) -> RunResultStreaming:
        """执行代理工作流的流式方法"""

@dataclass
class Agent(AgentBase, Generic[TContext]):
    """AI代理配置类，包含指令、工具、防护措施、代理切换等配置"""
    
    name: str                                    # 代理名称
    instructions: str | Callable | None = None  # 代理指令（系统提示）
    tools: list[Tool] = field(default_factory=list)        # 可用工具列表
    handoffs: list[Agent | Handoff] = field(default_factory=list)  # 可切换的代理
    model: str | Model | None = None             # 使用的模型
    model_settings: ModelSettings                # 模型设置
    input_guardrails: list[InputGuardrail]      # 输入防护
    output_guardrails: list[OutputGuardrail]    # 输出防护
    output_type: type[Any] | None = None        # 输出类型