OpenAI Agents流式输出实战：用Python+asyncio打造实时笑话生成器（附完整代码）

最近在开发AI应用时，我越来越感受到流式输出的重要性。那种让用户看着文字一个个蹦出来的体验，远比等待几秒钟然后突然出现一大段文字要友好得多。特别是对于需要实时交互的场景，比如聊天机器人、代码助手或者我们今天要讲的笑话生成器，流式输出能让用户感受到AI正在"思考"和"生成"的过程，而不是一个黑盒子。

OpenAI Agents SDK在这方面做得相当出色，它提供了多种级别的流式事件处理机制。你可以选择接收最底层的token级别更新，也可以处理更高级别的"工具调用完成"、"消息生成"等事件。这种灵活性让开发者能够根据应用需求选择合适的粒度，既不会因为事件太多而处理不过来，也不会因为事件太少而显得响应迟钝。

这篇文章我会带你从零开始，构建一个完整的实时笑话生成器。我们会深入探讨asyncio在异步流式处理中的应用，解析不同类型的事件处理策略，并分享我在实际开发中积累的一些调试技巧和性能优化经验。无论你是想为现有应用添加流式输出功能，还是想学习如何构建实时交互的AI应用，这篇文章都能给你实用的指导。

1. 环境准备与基础配置

在开始编码之前，我们需要确保开发环境配置正确。OpenAI Agents SDK对Python版本有一定要求，建议使用Python 3.9或更高版本。我个人的经验是，Python 3.11在异步性能方面有显著提升，如果你的项目对性能要求较高，可以考虑升级。

首先安装必要的依赖包：

pip install openai-agents
pip install python-dotenv
pip install httpx

这里有几个关键点需要注意。openai-agents是核心SDK，它封装了与OpenAI API的交互逻辑。python-dotenv用于管理环境变量，特别是API密钥这样的敏感信息。httpx是一个现代化的HTTP客户端，支持异步请求，在处理流式响应时比传统的requests库更高效。

接下来创建配置文件。我习惯将API密钥放在环境变量中，而不是硬编码在代码里：

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
    DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "gpt-4o-mini")
    
    @classmethod
    def validate(cls):
        if not cls.OPENAI_API_KEY:
            raise ValueError("OPENAI_API_KEY环境变量未设置")
        return True

在项目根目录创建.env文件：

OPENAI_API_KEY=sk-your-actual-api-key-here
DEFAULT_MODEL=gpt-4o-mini

提示：如果你使用的是其他兼容OpenAI API的服务（比如Azure OpenAI或某些国内服务），可以通过修改BASE_URL来适配。不过要注意，不同服务商对流式输出的支持程度可能有所不同。

关于模型选择，我测试过几个不同的模型在流式输出方面的表现：

模型名称	流式响应速度	输出质量	适用场景
gpt-4o-mini	快	良好	实时聊天、简单任务
gpt-4o	中等	优秀	复杂推理、创意生成
gpt-4-turbo	较慢	优秀	需要深度思考的任务

对于笑话生成器这种轻量级应用，gpt-4o-mini是个不错的选择，它在响应速度和成本之间取得了很好的平衡。如果你发现生成的笑话质量不够高，可以切换到gpt-4o，但要注意这可能会增加延迟。

在开始编写核心代码之前，我们先创建一个简单的测试脚本来验证环境配置是否正确：

# test_config.py
import asyncio
from config import Config
from agents import Agent, Runner

async def test_connection():
    """测试与OpenAI API的连接"""
    try:
        Config.validate()
        agent = Agent(
            name="TestAgent",
            instructions="You are a helpful assistant.",
        )
        result = await Runner.run(agent, "Say hello in one word.")
        print(f"连接测试成功: {result.final_output}")
        return True
    except Exception as e:
        print(f"连接测试失败: {e}")
        return False

if __name__ == "__main__":
    asyncio.run(test_connection())

运行这个测试脚本可以确保你的API密钥和网络配置都正确。如果遇到连接问题，常见的排查步骤包括：检查API密钥是否正确、确认网络是否可以访问OpenAI API、验证Python环境是否安装了正确版本的依赖包。

2. 核心架构设计与异步事件处理

构建流式输出应用的核心在于理解事件驱动的异步架构。OpenAI Agents SDK提供了多层次的事件系统，我们需要根据应用需求选择合适的事件处理策略。

2.1 异步编程基础

在深入Agents SDK之前，我们先回顾一下Python的asyncio基础。异步编程的核心是事件循环（Event Loop），它负责调度和执行协程（coroutine）。对于流式输出应用，我们主要使用async for循环来处理异步生成器。

import asyncio
import time

async def process_stream():
    """模拟处理流式数据"""
    async def data_generator():
        """模拟异步数据生成器"""
        for i in range(5):
            await asyncio.sleep(0.5)  # 模拟网络延迟
            yield f"数据块 {i}"
    
    start_time = time.time()
    async for chunk in data_generator():
        elapsed = time.time() - start_time
        print(f"[{elapsed:.2f}s] 收到: {chunk}")
    
    print(f"总耗时: {time.time() - start_time:.2f}秒")

# 运行示例
asyncio.run(process_stream())

这个简单的例子展示了异步生成器的基本模式。在实际的OpenAI Agents应用中，Runner.run_streamed()返回的就是这样一个异步生成器，它会持续产生事件直到整个对话完成。

2.2 事件类型详解

OpenAI Agents SDK主要提供三种类型的事件，每种都有不同的用途：

1. 原始响应事件（RawResponsesStreamEvent）

   • 最底层的事件，对应LLM生成的每个token
   • 包含response.created、response.output_text.delta等子类型
   • 适合需要实时显示每个字符的场景

2. 运行项目事件（RunItemStreamEvent）

   • 更高级别的事件，在某个项目完全生成时触发
   • 包括工具调用、工具输出、消息输出等
   • 适合需要知道"工具调用完成"或"消息生成完成"的场景

3. 代理更新事件（AgentUpdatedStreamEvent）

   • 当代理状态发生变化时触发，比如发生交接（handoff）
   • 包含新旧代理的信息

下面是一个对比表格，帮助你理解不同事件类型的适用场景：

事件类型	触发时机	数据粒度	适用场景	性能影响
RawResponsesStreamEvent	每个token生成时	字符级别	实时打字效果	高（事件数量多）
RunItemStreamEvent	项目完成时	句子/段落级别	进度状态更新	中等
AgentUpdatedStreamEvent	代理切换时	会话级别	多代理协作	低

2.3 笑话生成器的架构设计

基于对事件系统的理解，我们来设计笑话生成器的架构。我们的应用需要：

1. 接收用户输入（比如想要的笑话类型）
2. 调用工具决定生成的笑话数量
3. 流式生成指定数量的笑话
4. 实时显示生成进度
5. 处理可能的错误和异常

# joke_generator/architecture.py
from dataclasses import dataclass
from typing import Optional, List, Callable, Awaitable
from enum import Enum

class EventType(Enum):
    """自定义事件类型枚举"""
    CONNECTION_ESTABLISHED = "connection_established"
    TOOL_CALL_STARTED = "tool_call_started"
    TOOL_CALL_COMPLETED = "tool_call_completed"
    JOKE_STARTED = "joke_started"
    JOKE_CHUNK = "joke_chunk"
    JOKE_COMPLETED = "joke_completed"
    ERROR_OCCURRED = "error_occurred"
    SESSION_ENDED = "session_ended"

@dataclass
class JokeEvent:
    """笑话事件数据类"""
    event_type: EventType
    data: dict
    timestamp: float
    sequence_id: Optional[int] = None
    
class JokeGenerator:
    """笑话生成器核心类"""
    def __init__(self, model: str = "gpt-4o-mini"):
        self.model = model
        self.event_handlers: List[Callable[[JokeEvent], Awaitable[None]]] = []
    
    async def _emit_event(self, event: JokeEvent):
        """触发事件，通知所有注册的处理函数"""
        for handler in self.event_handlers:
            try:
                await handler(event)
            except Exception as e:
                print(f"事件处理函数出错: {e}")
    
    def register_handler(self, handler: Callable[[JokeEvent], Awaitable[None]]):
        """注册事件处理函数"""
        self.event_handlers.append(handler)
    
    # 其他方法将在后续章节实现

这个架构设计的关键在于事件驱动。我们将生成过程分解为多个离散的事件，每个事件都有明确的状态和语义。这样做的好处是：

• 解耦：事件生产者和消费者之间没有直接依赖
• 可扩展：可以轻松添加新的事件类型或处理函数
• 可测试：可以模拟事件流进行单元测试
• 可监控：每个事件都包含时间戳，便于性能分析

在实际开发中，我建议为每个重要的事件类型定义明确的接口和数据结构。这不仅能提高代码的可读性，还能在团队协作时减少误解。

3. 工具函数集成与智能代理配置

工具函数是OpenAI Agents的核心特性之一，它允许LLM调用外部函数来获取信息或执行操作。在我们的笑话生成器中，我们需要一个工具来决定生成多少个笑话。

3.1 创建自定义工具函数

让我们从最简单的工具开始——随机决定笑话数量：

# joke_generator/tools.py
import random
from typing import Optional
from agents import function_tool

@function_tool
def how_many_jokes(
    min_jokes: int = 1,
    max_jokes: int = 10,
    preference: Optional[str] = None
) -> int:
    """
    决定要生成多少个笑话。
    
    Args:
        min_jokes: 最少笑话数量，默认为1
        max_jokes: 最多笑话数量，默认为10
        preference: 用户偏好，可以是'short'、'medium'或'long'
        
    Returns:
        要生成的笑话数量
    """
    if preference == "short":
        return random.randint(min_jokes, min(3, max_jokes))
    elif preference == "medium":
        mid = (min_jokes + max_jokes) // 2
        return random.randint(max(min_jokes, mid-1), min(max_jokes, mid+1))
    elif preference == "long":
        return random.randint(max(min_jokes, max_jokes-2), max_jokes)
    else:
        return random.randint(min_jokes, max_jokes)

这个工具函数有几个值得注意的设计点：

1. 参数验证：通过类型注解和默认值确保参数的有效性
2. 文档字符串：详细的文档字符串会被SDK自动解析，帮助LLM理解工具的用途
3. 错误处理：虽然没有显式的try-catch，但函数内部逻辑避免了可能的错误

在实际应用中，工具函数可能会更复杂。比如，我们可以添加一个从笑话数据库获取主题的工具：

@function_tool
def get_joke_topics(category: str = "general", count: int = 5) -> list:
    """
    获取指定类别的笑话主题列表。
    
    Args:
        category: 笑话类别，如'programming'、'animal'、'food'等
        count: 返回的主题数量
        
    Returns:
        主题字符串列表
    """
    # 这里可以是数据库查询或API调用
    # 为了示例，我们返回硬编码的数据
    topics_by_category = {
        "programming": [
            "为什么程序员讨厌自然？因为太多bug！",
            "Python和Java走进酒吧...",
            "递归笑话：要理解递归笑话，首先要理解递归笑话",
            "为什么开发者总是分不清万圣节和圣诞节？因为Oct 31 == Dec 25",
            "Git commit消息：'修复bug'、'再修复bug'、'真的修复了这次'"
        ],
        "animal": [
            "为什么鸡要过马路？",
            "猫和狗的区别是什么？",
            "企鹅为什么不会飞？",
            "蜗牛对跑车说了什么？",
            "蜜蜂为什么嗡嗡叫？"
        ],
        "food": [
            "为什么面包要去见医生？",
            "番茄为什么变红了？",
            "奶酪对镜子说了什么？",
            "为什么热狗要去上学？",
            "鸡蛋为什么害羞？"
        ]
    }
    
    topics = topics_by_category.get(category.lower(), topics_by_category["general"])
    return topics[:count]

3.2 智能代理配置

有了工具函数，接下来需要配置智能代理。代理的配置决定了它的行为和能力：

# joke_generator/agent.py
from agents import Agent, RunConfig, OpenAIProvider
from .tools import how_many_jokes, get_joke_topics

class JokeAgentFactory:
    """笑话代理工厂类"""
    
    @staticmethod
    def create_basic_agent(
        name: str = "Joker",
        model: str = "gpt-4o-mini",
        temperature: float = 0.8,
        max_tokens: int = 1000
    ) -> Agent:
        """
        创建基础笑话代理。
        
        Args:
            name: 代理名称
            model: 使用的模型
            temperature: 创造性参数（0.0-2.0）
            max_tokens: 最大输出token数
            
        Returns:
            配置好的Agent实例
        """
        return Agent(
            name=name,
            instructions="""
            你是一个专业的笑话生成器。你的任务是生成有趣、恰当的笑话。
            
            工作流程：
            1. 首先调用how_many_jokes工具决定生成多少个笑话
            2. 然后生成指定数量的笑话
            3. 每个笑话应该：
               - 有明确的包袱（punchline）
               - 适合普通观众
               - 避免冒犯性内容
               - 如果可能，包含一点巧妙的双关或反转
            
            输出格式：
            - 每个笑话单独一行
            - 使用数字编号（1. 2. 3. ...）
            - 在笑话之间留一个空行
            - 最后可以加一句友好的结束语
            
            记住：质量比数量更重要！
            """,
            tools=[how_many_jokes],
            model=model,
            model_settings={
                "temperature": temperature,
                "max_tokens": max_tokens,
            }
        )
    
    @staticmethod
    def create_advanced_agent(
        name: str = "AdvancedJoker",
        model: str = "gpt-4o",
        temperature: float = 0.9
    ) -> Agent:
        """
        创建高级笑话代理，支持更多工具。
        """
        return Agent(
            name=name,
            instructions="""
            你是一个高级笑话生成器，拥有更多创作工具。
            
            可选工作流程：
            1. 可以调用get_joke_topics获取灵感
            2. 调用how_many_jokes决定数量
            3. 生成高质量的笑话
            
            特别要求：
            - 可以根据用户请求的类别定制笑话
            - 可以混合不同类型的幽默（双关、反转、夸张等）
            - 确保笑话新鲜、不重复
            - 考虑文化敏感性
            
            创意提示：
            - 技术笑话：结合最新科技趋势
            - 动物笑话：体现动物特性
            - 食物笑话：利用食物特点创造幽默
            """,
            tools=[how_many_jokes, get_joke_topics],
            model=model,
            model_settings={
                "temperature": temperature,
                "max_tokens": 1500,
                "top_p": 0.95,
            }
        )

代理配置中的instructions字段非常重要。它相当于给AI的"工作说明书"，决定了AI如何理解任务和使用工具。我建议：

• 明确工作流程：用编号步骤说明AI应该做什么
• 设定质量标准：说明什么是好的输出
• 指定格式要求：确保输出的一致性
• 添加注意事项：避免常见问题

3.3 运行配置与错误处理

运行配置（RunConfig）控制单次执行的行为：

# joke_generator/runner.py
from agents import Runner, RunConfig, OpenAIProvider
from typing import Optional, Dict, Any
import asyncio

class JokeRunner:
    """笑话运行器，封装流式执行逻辑"""
    
    def __init__(
        self,
        api_key: str,
        base_url: Optional[str] = None,
        default_model: str = "gpt-4o-mini"
    ):
        self.run_config = RunConfig(
            model_provider=OpenAIProvider(
                api_key=api_key,
                base_url=base_url or "https://api.openai.com/v1",
                use_responses=False,
            ),
            model=default_model,
            max_turns=5,  # 防止无限循环
        )
    
    async def run_streaming(
        self,
        agent: Agent,
        user_input: str,
        event_handler: Optional[callable] = None
    ):
        """
        运行流式笑话生成。
        
        Args:
            agent: 要使用的代理
            user_input: 用户输入
            event_handler: 自定义事件处理函数
            
        Yields:
            各种流式事件
        """
        try:
            result = Runner.run_streamed(
                agent,
                input=user_input,
                run_config=self.run_config
            )
            
            async for event in result.stream_events():
                # 应用事件过滤和处理逻辑
                processed_event = self._process_event(event)
                
                if event_handler:
                    await event_handler(processed_event)
                else:
                    yield processed_event
                
                # 检查是否应该提前结束
                if self._should_stop(event):
                    break
                    
        except Exception as e:
            error_event = {
                "type": "error",
                "error": str(e),
                "timestamp": asyncio.get_event_loop().time()
            }
            if event_handler:
                await event_handler(error_event)
            else:
                yield error_event
    
    def _process_event(self, event) -> Dict[str, Any]:
        """处理原始事件，转换为应用层事件"""
        # 这里可以添加自定义处理逻辑
        # 比如过滤某些事件、添加元数据等
        return {
            "type": event.type,
            "data": getattr(event, "data", {}),
            "item": getattr(event, "item", None),
            "timestamp": asyncio.get_event_loop().time()
        }
    
    def _should_stop(self, event) -> bool:
        """判断是否应该提前结束流式处理"""
        # 例如：检测到错误、用户取消、超时等
        return False

错误处理是流式应用中的重要环节。我们需要考虑多种可能的错误情况：

# joke_generator/error_handling.py
from typing import Dict, Any, Optional
import asyncio
from datetime import datetime

class ErrorHandler:
    """统一的错误处理器"""
    
    ERROR_CATEGORIES = {
        "api_error": ["InvalidAPIKey", "RateLimitExceeded", "ModelNotFound"],
        "network_error": ["ConnectionError", "TimeoutError", "SSLError"],
        "processing_error": ["JSONDecodeError", "ValidationError"],
        "user_error": ["InvalidInput", "Cancelled"],
        "unknown_error": []
    }
    
    @classmethod
    async def handle_stream_error(
        cls,
        error: Exception,
        context: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        处理流式处理中的错误。
        
        Args:
            error: 异常对象
            context: 错误发生时的上下文信息
            
        Returns:
            处理后的错误响应
        """
        error_type = type(error).__name__
        error_category = cls._categorize_error(error_type)
        
        response = {
            "success": False,
            "error_type": error_type,
            "error_category": error_category,
            "message": str(error),
            "timestamp": datetime.now().isoformat(),
            "context": {
                "agent_name": context.get("agent_name"),
                "user_input": context.get("user_input", "")[:100],  # 截断长输入
                "retry_count": context.get("retry_count", 0)
            }
        }
        
        # 根据错误类别采取不同措施
        if error_category == "api_error":
            response["suggestion"] = "请检查API密钥和配额"
            response["retryable"] = True
            response["retry_delay"] = 5  # 5秒后重试
            
        elif error_category == "network_error":
            response["suggestion"] = "网络连接不稳定，请检查网络"
            response["retryable"] = True
            response["retry_delay"] = 2
            
        elif error_category == "user_error":
            response["suggestion"] = "请检查输入内容"
            response["retryable"] = False
            
        else:
            response["suggestion"] = "发生未知错误，请联系支持"
            response["retryable"] = False
        
        # 记录错误日志（在实际应用中可能写入文件或数据库）
        await cls._log_error(response)
        
        return response
    
    @classmethod
    def _categorize_error(cls, error_type: str) -> str:
        """将错误分类"""
        for category, error_types in cls.ERROR_CATEGORIES.items():
            if error_type in error_types:
                return category
        return "unknown_error"
    
    @classmethod
    async def _log_error(cls, error_info: Dict[str, Any]):
        """记录错误日志"""
        # 这里可以集成日志系统
        print(f"[ERROR] {error_info}")

4. 完整实现与实战演示

现在我们把所有组件组合起来，创建一个完整的、可运行的笑话生成器应用。

4.1 主应用类实现

# joke_generator/main.py
import asyncio
import json
from typing import Optional, List, Dict, Any
from datetime import datetime
from agents import Agent, Runner

from .agent import JokeAgentFactory
from .runner import JokeRunner
from .error_handling import ErrorHandler

class RealTimeJokeGenerator:
    """
    实时笑话生成器主类。
    
    功能特性：
    - 支持多种笑话类型（编程、动物、食物等）
    - 可配置的笑话数量和质量
    - 完整的流式输出体验
    - 错误处理和重试机制
    - 性能监控和日志记录
    """
    
    def __init__(
        self,
        api_key: str,
        base_url: Optional[str] = None,
        default_model: str = "gpt-4o-mini",
        enable_logging: bool = True
    ):
        """
        初始化笑话生成器。
        
        Args:
            api_key: OpenAI API密钥
            base_url: API基础URL（用于自定义部署）
            default_model: 默认使用的模型
            enable_logging: 是否启用日志记录
        """
        self.api_key = api_key
        self.base_url = base_url
        self.default_model = default_model
        self.enable_logging = enable_logging
        
        # 初始化组件
        self.runner = JokeRunner(api_key, base_url, default_model)
        self.agent_factory = JokeAgentFactory()
        
        # 状态跟踪
        self.session_id = None
        self.start_time = None
        self.jokes_generated = 0
        self.total_tokens = 0
        
        # 事件处理器列表
        self.event_handlers = []
        
        # 性能统计
        self.stats = {
            "total_sessions": 0,
            "total_jokes": 0,
            "avg_generation_time": 0,
            "success_rate": 1.0
        }
    
    async def generate_jokes(
        self,
        prompt: str,
        joke_type: str = "general",
        num_jokes: Optional[int] = None,
        streaming: bool = True,
        callback = None
    ) -> Dict[str, Any]:
        """
        生成笑话的主要方法。
        
        Args:
            prompt: 用户提示，如"讲几个关于编程的笑话"
            joke_type: 笑话类型（general/programming/animal/food）
            num_jokes: 指定笑话数量（None表示让AI决定）
            streaming: 是否使用流式输出
            callback: 回调函数，用于处理实时输出
            
        Returns:
            包含结果和统计信息的字典
        """
        # 开始新会话
        self._start_session()
        
        try:
            # 准备用户输入
            user_input = self._prepare_input(prompt, joke_type, num_jokes)
            
            # 选择代理
            agent = self._select_agent(joke_type)
            
            # 生成笑话
            if streaming:
                result = await self._generate_streaming(
                    agent, user_input, callback
                )
            else:
                result = await self._generate_non_streaming(agent, user_input)
            
            # 更新统计信息
            self._update_stats(result)
            
            return {
                "success": True,
                "session_id": self.session_id,
                "result": result,
                "stats": self.stats.copy(),
                "generation_time": self._get_elapsed_time()
            }
            
        except Exception as e:
            # 错误处理
            error_result = await ErrorHandler.handle_stream_error(
                e,
                {
                    "agent_name": "JokeGenerator",
                    "user_input": prompt,
                    "session_id": self.session_id
                }
            )
            
            self.stats["success_rate"] = (
                self.stats["total_sessions"] * self.stats["success_rate"]
            ) / (self.stats["total_sessions"] + 1)
            
            return {
                "success": False,
                "session_id": self.session_id,
                "error": error_result,
                "generation_time": self._get_elapsed_time()
            }
    
    async def _generate_streaming(
        self,
        agent: Agent,
        user_input: str,
        callback
    ) -> Dict[str, Any]:
        """
        流式生成笑话。
        
        Args:
            agent: 使用的代理
            user_input: 用户输入
            callback: 回调函数
            
        Returns:
            生成结果
        """
        jokes = []
        current_joke = ""
        in_joke = False
        
        async def handle_event(event):
            nonlocal jokes, current_joke, in_joke
            
            if event["type"] == "run_item_stream_event":
                item = event.get("item")
                if item and item.type == "message_output_item":
                    # 提取文本内容
                    from agents import ItemHelpers
                    text = ItemHelpers.text_message_output(item)
                    
                    # 处理文本，分割笑话
                    for line in text.split('\n'):
                        line = line.strip()
                        if not line:
                            continue
                            
                        # 检测笑话开始（数字编号）
                        if line[0].isdigit() and '.' in line[:3]:
                            if current_joke and in_joke:
                                jokes.append(current_joke.strip())
                                current_joke = ""
                            in_joke = True
                            current_joke = line.split('.', 1)[1].strip() + " "
                        elif in_joke:
                            current_joke += line + " "
                    
                    # 调用回调函数
                    if callback:
                        await callback({
                            "type": "text_chunk",
                            "text": text,
                            "timestamp": event["timestamp"]
                        })
            
            elif event["type"] == "agent_updated_stream_event":
                if callback:
                    await callback({
                        "type": "agent_updated",
                        "new_agent": event["data"].get("new_agent", {}).get("name"),
                        "timestamp": event["timestamp"]
                    })
            
            elif event["type"] == "tool_call_item":
                if callback:
                    await callback({
                        "type": "tool_called",
                        "tool_name": event["data"].get("tool_name"),
                        "timestamp": event["timestamp"]
                    })
        
        # 执行流式运行
        result_stream = await self.runner.run_streaming(
            agent, user_input, handle_event
        )
        
        # 等待流式处理完成
        async for _ in result_stream:
            pass
        
        # 添加最后一个笑话
        if current_joke and in_joke:
            jokes.append(current_joke.strip())
        
        return {
            "jokes": jokes,
            "count": len(jokes),
            "streaming": True
        }
    
    async def _generate_non_streaming(
        self,
        agent: Agent,
        user_input: str
    ) -> Dict[str, Any]:
        """
        非流式生成笑话（一次性返回）。
        
        Args:
            agent: 使用的代理
            user_input: 用户输入
            
        Returns:
            生成结果
        """
        result = await Runner.run(agent, user_input)
        
        # 解析笑话
        jokes = []
        text = result.final_output
        current_joke = ""
        
        for line in text.split('\n'):
            line = line.strip()
            if not line:
                if current_joke:
                    jokes.append(current_joke.strip())
                    current_joke = ""
                continue
                
            if line[0].isdigit() and '.' in line[:3]:
                if current_joke:
                    jokes.append(current_joke.strip())
                current_joke = line.split('.', 1)[1].strip() + " "
            elif current_joke:
                current_joke += line + " "
        
        if current_joke:
            jokes.append(current_joke.strip())
        
        return {
            "jokes": jokes,
            "count": len(jokes),
            "streaming": False,
            "raw_output": text
        }
    
    def _start_session(self):
        """开始新会话"""
        self.session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        self.start_time = asyncio.get_event_loop().time()
        self.jokes_generated = 0
        self.total_tokens = 0
    
    def _prepare_input(
        self,
        prompt: str,
        joke_type: str,
        num_jokes: Optional[int]
    ) -> str:
        """准备用户输入"""
        input_parts = [prompt]
        
        if joke_type != "general":
            input_parts.append(f"主题：{joke_type}笑话")
        
        if num_jokes:
            input_parts.append(f"数量：{num_jokes}个")
        
        return "。".join(input_parts)
    
    def _select_agent(self, joke_type: str) -> Agent:
        """根据笑话类型选择代理"""
        if joke_type == "programming":
            return self.agent_factory.create_advanced_agent(
                name="CodeJoker",
                model="gpt-4o"
            )
        else:
            return self.agent_factory.create_basic_agent(
                name=f"{joke_type.capitalize()}Joker"
            )
    
    def _update_stats(self, result: Dict[str, Any]):
        """更新统计信息"""
        self.stats["total_sessions"] += 1
        self.stats["total_jokes"] += result.get("count", 0)
        
        elapsed = self._get_elapsed_time()
        if elapsed > 0:
            self.stats["avg_generation_time"] = (
                (self.stats["avg_generation_time"] * (self.stats["total_sessions"] - 1) + elapsed)
                / self.stats["total_sessions"]
            )
    
    def _get_elapsed_time(self) -> float:
        """获取经过的时间"""
        if self.start_time:
            return asyncio.get_event_loop().time() - self.start_time
        return 0.0
    
    def get_performance_report(self) -> Dict[str, Any]:
        """获取性能报告"""
        return {
            "summary": self.stats.copy(),
            "current_session": {
                "id": self.session_id,
                "jokes_generated": self.jokes_generated,
                "total_tokens": self.total_tokens,
                "elapsed_time": self._get_elapsed_time()
            },
            "recommendations": self._generate_recommendations()
        }
    
    def _generate_recommendations(self) -> List[str]:
        """生成性能优化建议"""
        recommendations = []
        
        avg_time = self.stats["avg_generation_time"]
        if avg_time > 10.0:
            recommendations.append("生成时间较长，建议使用gpt-4o-mini模型")
        elif avg_time < 2.0:
            recommendations.append("性能良好，可以考虑使用更高质量的模型")
        
        if self.stats["success_rate"] < 0.95:
            recommendations.append("成功率较低，请检查API配置和网络连接")
        
        return recommendations

4.2 使用示例和演示

现在让我们看看如何使用这个笑话生成器。这里有几个不同复杂度的示例：

# examples/basic_usage.py
import asyncio
import os
from joke_generator.main import RealTimeJokeGenerator

async def basic_example():
    """基础使用示例"""
    # 从环境变量获取API密钥
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        print("请设置OPENAI_API_KEY环境变量")
        return
    
    # 创建生成器实例
    generator = RealTimeJokeGenerator(api_key)
    
    # 定义回调函数处理流式输出
    async def streaming_callback(chunk):
        if chunk["type"] == "text_chunk":
            print(chunk["text"], end="", flush=True)
        elif chunk["type"] == "agent_updated":
            print(f"\n[切换到代理: {chunk['new_agent']}]")
        elif chunk["type"] == "tool_called":
            print(f"\n[调用工具: {chunk['tool_name']}]")
    
    print("=== 开始生成编程笑话 ===\n")
    
    # 生成笑话（流式）
    result = await generator.generate_jokes(
        prompt="讲几个有趣的编程笑话",
        joke_type="programming",
        streaming=True,
        callback=streaming_callback
    )
    
    print(f"\n\n=== 生成完成 ===")
    print(f"生成了 {result['result']['count']} 个笑话")
    print(f"耗时: {result['generation_time']:.2f}秒")
    
    # 显示性能报告
    report = generator.get_performance_report()
    print(f"\n平均生成时间: {report['summary']['avg_generation_time']:.2f}秒")

async def advanced_example():
    """高级使用示例：批量生成不同类型笑话"""
    api_key = os.getenv("OPENAI_API_KEY")
    generator = RealTimeJokeGenerator(api_key)
    
    joke_types = ["programming", "animal", "food"]
    results = []
    
    for joke_type in joke_types:
        print(f"\n{'='*40}")
        print(f"生成 {joke_type} 笑话")
        print('='*40)
        
        result = await generator.generate_jokes(
            prompt=f"讲几个关于{joke_type}的笑话",
            joke_type=joke_type,
            num_jokes=3,
            streaming=False  # 批量生成时使用非流式
        )
        
        if result["success"]:
            results.append(result)
            print(f"成功生成 {result['result']['count']} 个笑话")
            
            # 显示生成的笑话
            for i, joke in enumerate(result["result"]["jokes"], 1):
                print(f"{i}. {joke}")
        else:
            print(f"生成失败: {result['error']['message']}")
    
    # 生成汇总报告
    total_jokes = sum(r["result"]["count"] for r in results if r["success"])
    print(f"\n总计生成 {total_jokes} 个笑话")
    
    report = generator.get_performance_report()
    print(f"\n性能报告:")
    print(f"- 总会话数: {report['summary']['total_sessions']}")
    print(f"- 总笑话数: {report['summary']['total_jokes']}")
    print(f"- 平均生成时间: {report['summary']['avg_generation_time']:.2f}秒")
    print(f"- 成功率: {report['summary']['success_rate']:.2%}")
    
    if report["recommendations"]:
        print(f"\n优化建议:")
        for rec in report["recommendations"]:
            print(f"- {rec}")

async def interactive_example():
    """交互式示例：模拟聊天界面"""
    api_key = os.getenv("OPENAI_API_KEY")
    generator = RealTimeJokeGenerator(api_key)
    
    print("欢迎使用实时笑话生成器！")
    print("输入'quit'退出，'stats'查看统计，'help'查看帮助\n")
    
    while True:
        try:
            user_input = input("\n你想听什么笑话？> ").strip()
            
            if user_input.lower() == 'quit':
                print("再见！")
                break
            elif user_input.lower() == 'stats':
                report = generator.get_performance_report()
                print(f"已生成 {report['summary']['total_jokes']} 个笑话")
                print(f"平均响应时间: {report['summary']['avg_generation_time']:.2f}秒")
                continue
            elif user_input.lower() == 'help':
                print("可用命令:")
                print("- quit: 退出程序")
                print("- stats: 查看统计")
                print("- 其他: 生成相应主题的笑话")
                continue
            elif not user_input:
                continue
            
            # 简单解析用户输入
            if "编程" in user_input or "代码" in user_input:
                joke_type = "programming"
            elif "动物" in user_input:
                joke_type = "animal"
            elif "食物" in user_input or "吃" in user_input:
                joke_type = "food"
            else:
                joke_type = "general"
            
            print(f"\n生成{joke_type}笑话中...\n")
            
            async def callback(chunk):
                if chunk["type"] == "text_chunk":
                    print(chunk["text"], end="", flush=True)
            
            result = await generator.generate_jokes(
                prompt=user_input,
                joke_type=joke_type,
                streaming=True,
                callback=callback
            )
            
            print(f"\n\n[生成完成]")
            
        except KeyboardInterrupt:
            print("\n\n程序被中断")
            break
        except Exception as e:
            print(f"\n错误: {e}")

if __name__ == "__main__":
    # 运行基础示例
    print("运行基础示例...")
    asyncio.run(basic_example())
    
    print("\n\n运行高级示例...")
    asyncio.run(advanced_example())
    
    # 取消注释以运行交互式示例
    # print("\n\n运行交互式示例...")
    # asyncio.run(interactive_example())

4.3 性能优化和调试技巧

在实际使用中，你可能会遇到各种性能问题。这里分享一些我在开发过程中积累的优化技巧：

# examples/optimization_tips.py
import asyncio
import time
from typing import List, Dict, Any
import aiohttp
from agents import Agent, Runner

class OptimizationTips:
    """性能优化技巧集合"""
    
    @staticmethod
    async def batch_processing(
        agents: List[Agent],
        inputs: List[str],
        max_concurrent: int = 3
    ) -> List[Dict[str, Any]]:
        """
        批量处理多个请求，提高吞吐量。
        
        Args:
            agents: 代理列表
            inputs: 输入列表
            max_concurrent: 最大并发数
            
        Returns:
            结果列表
        """
        semaphore = asyncio.Semaphore(max_concurrent)
        results = []
        
        async def run_with_semaphore(agent, input_text):
            async with semaphore:
                return await Runner.run(agent, input_text)
        
        tasks = [
            run_with_semaphore(agent, input_text)
            for agent, input_text in zip(agents, inputs)
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results
    
    @staticmethod
    async def connection_pooling():
        """
        使用连接池优化网络性能。
        
        在频繁调用API时，重用HTTP连接可以显著减少延迟。
        """
        import aiohttp
        
        # 创建共享的aiohttp会话
        async with aiohttp.ClientSession() as session:
            # 配置Agent使用自定义会话
            # 注意：这需要根据SDK的具体实现调整
            pass
    
    @staticmethod
    def event_filtering_strategy():
        """
        事件过滤策略，减少不必要的事件处理。
        
        根据应用需求选择合适的事件粒度：
        1. 只关心最终结果：使用非流式API
        2. 需要进度反馈：使用RunItemStreamEvent
        3. 需要打字机效果：使用RawResponsesStreamEvent
        """
        strategies = {
            "minimal": {
                "description": "最小事件集，只处理关键事件",
                "filter": lambda e: e.type in [
                    "run_item_stream_event",
                    "agent_updated_stream_event"
                ]
            },
            "balanced": {
                "description": "平衡集，适合大多数应用",
                "filter": lambda e: e.type != "raw_response_event"
            },
            "verbose": {
                "description": "完整事件集，用于调试",
                "filter": lambda e: True  # 不过滤任何事件
            }
        }
        return strategies
    
    @staticmethod
    async def monitor_performance(generator, duration: int = 60):
        """
        监控生成器性能。
        
        Args:
            generator: 笑话生成器实例
            duration: 监控时长（秒）
        """
        import matplotlib.pyplot as plt
        import numpy as np
        
        response_times = []
        joke_counts = []
        start_time = time.time()
        
        print(f"开始性能监控，时长{duration}秒...")
        
        while time.time() - start_time < duration:
            loop_start = time.time()
            
            result = await generator.generate_jokes(
                prompt="讲一个笑话",
                streaming=False
            )
            
            response_time = time.time() - loop_start
            response_times.append(response_time)
            joke_counts.append(result["result"]["count"])
            
            # 每秒更新一次统计
            if len(response_times) % 5 == 0:
                avg_time = np.mean(response_times[-5:])
                print(f"最近5次平均响应时间: {avg_time:.2f}秒")
            
            await asyncio.sleep(1)  # 控制请求频率
        
        # 生成性能报告
        print(f"\n性能监控完成，共{len(response_times)}次请求")
        print(f"平均响应时间: {np.mean(response_times):.2f}秒")
        print(f"响应时间标准差: {np.std(response_times):.2f}秒")
        print(f"平均每次生成笑话数: {np.mean(joke_counts):.1f}")
        
        # 可视化（可选）
        if len(response_times) > 10:
            try:
                plt.figure(figsize=(10, 6))
                
                plt.subplot(2, 1, 1)
                plt.plot(response_times, 'b-', alpha=0.7)
                plt.axhline(y=np.mean(response_times), color='r', linestyle='--', label=f'平均: {np.mean(response_times):.2f}s')
                plt.xlabel('请求序号')
                plt.ylabel('响应时间 (秒)')
                plt.title('响应时间趋势')
                plt.legend()
                plt.grid(True, alpha=0.3)
                
                plt.subplot(2, 1, 2)
                plt.hist(response_times, bins=20, alpha=0.7, edgecolor='black')
                plt.xlabel('响应时间 (秒)')
                plt.ylabel('频次')
                plt.title('响应时间分布')
                plt.grid(True, alpha=0.3)
                
                plt.tight_layout()
                plt.show()
                
            except ImportError:
                print("未安装matplotlib，跳过可视化")
    
    @staticmethod
    def caching_strategy():
        """
        缓存策略，减少重复请求。
        
        对于常见请求，可以缓存结果以提高性能。
        """
        import hashlib
        import pickle
        from functools import lru_cache
        
        class JokeCache:
            def __init__(self, max_size: int = 100):
                self.cache = {}
                self.max_size = max_size
                self.hits = 0
                self.misses = 0
            
            def _generate_key(self, prompt: str, joke_type: str) -> str:
                """生成缓存键"""
                content = f"{prompt}:{joke_type}".encode('utf-8')
                return hashlib.md5(content).hexdigest()
            
            def get(self, prompt: str, joke_type: str):
                """获取缓存结果"""
                key = self._generate_key(prompt, joke_type)
                if key in self.cache:
                    self.hits += 1
                    return self.cache[key]
                self.misses += 1
                return None
            
            def set(self, prompt: str, joke_type: str, result):
                """设置缓存结果"""
                if len(self.cache) >= self.max_size:
                    # 简单的LRU策略：删除第一个键
                    first_key = next(iter(self.cache))
                    del self.cache[first_key]
                
                key = self._generate_key(prompt, joke_type)
                self.cache[key] = result
            
            def get_stats(self):
                """获取缓存统计"""
                total = self.hits + self.misses
                hit_rate = self.hits / total if total > 0 else 0
                return {
                    "size": len(self.cache),
                    "hits": self.hits,
                    "misses": self.misses,
                    "hit_rate": hit_rate
                }
        
        return JokeCache

4.4 错误处理和恢复策略

在真实的生产环境中，错误处理至关重要。这里提供一些增强的错误处理策略：

# examples/error_recovery.py
import asyncio
import random
from typing import Optional, Dict, Any
from datetime import datetime, timedelta

class EnhancedErrorRecovery:
    """增强的错误恢复策略"""
    
    def __init__(self):
        self.error_history = []
        self.circuit_state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
        self.circuit_opened_at = None
        self.failure_count = 0
        self.success_threshold = 3
        self.failure_threshold = 5
        self.reset_timeout = 60  # 60秒后尝试恢复
    
    async def execute_with_recovery(
        self,
        operation,
        *args,
        max_retries: int = 3,
        backoff_factor: float = 1.5,
        **kwargs
    ):
        """
        带恢复策略的执行方法。
        
        Args:
            operation: 要执行的操作（异步函数）
            max_retries: 最大重试次数
            backoff_factor: 退避因子
            **kwargs: 传递给操作的参数
            
        Returns:
            操作结果
        """
        # 检查熔断器状态
        if self.circuit_state == "OPEN":
            if self._should_try_recovery():
                self.circuit_state = "HALF_OPEN"
                print("[熔断器] 进入半开状态，尝试恢复")
            else:
                raise CircuitBreakerOpenError(
                    f"熔断器已打开，{self._time_until_recovery()}秒后重试"
                )
        
        last_error = None
        
        for attempt in range(max_retries + 1):
            try:
                # 执行操作
                result = await operation(*args, **kwargs)
                
                # 成功：更新熔断器状态
                self._record_success()
                return result
                
            except TemporaryError as e:
                # 临时错误：记录并重试
                last_error = e
                self._record_failure(str(e))
                
                if attempt < max_retries:
                    # 计算退避时间
                    wait_time = backoff_factor ** attempt + random.uniform(0, 1)
                    print(f"[重试] 第{attempt + 1}次失败，{wait_time:.1f}秒后重试: {e}")
                    
                    await asyncio.sleep(wait_time)
                else:
                    print(f"[失败] 达到最大重试次数: {e}")
                    
            except PermanentError as e:
                # 永久错误：不重试
                print(f"[永久错误] 不重试: {e}")
                raise
                
            except Exception as e:
                # 其他错误
                last_error = e
                self._record_failure(str(e))
                
                if attempt == max_retries:
                    print(f"[最终失败] 所有重试均失败: {e}")
        
        # 所有重试都失败
        if last_error:
            raise last_error
    
    def _record_success(self):
        """记录成功，更新熔断器状态"""
        self.failure_count = 0
        
        if self.circuit_state == "HALF_OPEN":
            self.circuit_state = "CLOSED"
            self.circuit_opened_at = None
            print("[熔断器] 恢复成功，关闭熔断器")
    
    def _record_failure(self, error_msg: str):
        """记录失败，更新熔断器状态"""
        self.failure_count += 1
        self.error_history.append({
            "timestamp": datetime.now(),
            "error": error_msg,
            "failure_count": self.failure_count
        })
        
        # 保留最近100个错误
        if len(self.error_history) > 100:
            self.error_history = self.error_history[-100:]
        
        # 检查是否需要打开熔断器
        if (self.circuit_state in ["CLOSED", "HALF_OPEN"] and 
            self.failure_count >= self.failure_threshold):
            self.circuit_state = "OPEN"
            self.circuit_opened_at = datetime.now()
            print(f"[熔断器] 失败次数达到{self.failure_threshold}，打开熔断器")
    
    def _should_try_recovery(self) -> bool:
        """检查是否应该尝试恢复"""
        if self.circuit_state != "OPEN" or not self.circuit_opened_at:
            return False
        
        elapsed = (datetime.now() - self.circuit_opened_at).total_seconds()
        return elapsed >= self.reset_timeout
    
    def _time_until_recovery(self) -> int:
        """计算距离恢复尝试还有多少秒"""
        if not self.circuit_opened_at:
            return 0
        
        elapsed = (datetime.now() - self.circuit_opened_at).total_seconds()
        return max(0, int(self.reset_timeout - elapsed))
    
    def get_status(self) -> Dict[str, Any]:
        """获取当前状态"""
        return {
            "circuit_state": self.circuit_state,
            "failure_count": self.failure_count,
            "error_history_size": len(self.error_history),
            "time_until_recovery": self._time_until_recovery(),
            "recent_errors": self.error_history[-5:] if self.error_history else []
        }

class TemporaryError(Exception):
    """临时错误，可以重试"""
    pass

class PermanentError(Exception):
    """永久错误，不应该重试"""
    pass

class CircuitBreakerOpenError(Exception):
    """熔断器已打开错误"""
    pass

# 使用示例
async def demo_error_recovery():
    """演示错误恢复策略"""
    recovery = EnhancedErrorRecovery()
    
    async def unreliable_operation(succeed: bool = False):
        """模拟不可靠的操作"""
        await asyncio.sleep(0.5)
        
        if random.random() < 0.7 and not succeed:  # 70%失败率
            raise TemporaryError("模拟临时错误")
        
        return "操作成功"
    
    print("测试错误恢复策略...")
    
    for i in range(10):
        print(f"\n尝试 {i + 1}:")
        try:
            result = await recovery.execute_with_recovery(
                unreliable_operation,
                succeed=(i == 8)  # 第9次强制成功
            )
            print(f"结果: {result}")
        except Exception as e:
            print(f"最终错误: {e}")
        
        # 显示状态
        status = recovery.get_status()
        print(f"状态: {status['circuit_state']}, 失败计数: {status['failure_count']}")
        
        await asyncio.sleep(1)

通过这些完整的实现和示例，你应该已经掌握了使用OpenAI Agents SDK构建流式输出应用的核心技术。从环境配置到架构设计，从工具集成到错误处理，每个环节都有其独特的技术要点和最佳实践。

在实际项目中，我建议先从简单的非流式版本开始，确保核心逻辑正确，然后再逐步添加流式输出功能。同时，要密切关注性能指标和错误率，根据实际情况调整配置参数。记住，好的用户体验不仅仅是技术实现，更是对细节的精心打磨。