第15课：流式输出

🌊 流式输出：让Agent响应更快

流式输出（Streaming）让Agent逐token返回结果，而不是等待全部生成完毕。这极大地改善了用户体验，特别是对于长文本生成场景。

📖 流式输出原理

非流式（传统）：
请求 → [等待5秒] → 完整响应

流式：
请求 → token1 → token2 → token3 → ... → 完成
       ↑ 用户立即看到输出开始

流式输出层次：
├── Token级流式（逐token输出）
├── 事件级流式（工具调用等事件）
├── 步骤级流式（Agent步骤完成）
└── 消息级流式（完整消息）

📊 流式实现方案

💻 代码实现：流式Agent

# 流式Agent - 支持多种流式输出方式
import time, json, queue, threading
from typing import Generator, Callable, Dict, Any, List
from dataclasses import dataclass

@dataclass
class StreamEvent:
    # 流式事件
    type: str  # "token", "tool_start", "tool_end", "step", "done"
    data: Any
    timestamp: float = 0
    
    def __post_init__(self):
        if not self.timestamp:
            self.timestamp = time.time()

class StreamingAgent:
    # 流式Agent
    def __init__(self, name):
        self.name = name
        self.tools = {}
        self.event_queue = queue.Queue()
    
    def register_tool(self, name, func, description):
        self.tools[name] = {"func": func, "description": description}
    
    def stream_chat(self, message) -> Generator[StreamEvent, None, None]:
        # 流式对话 - Generator方式
        # 1. 开始思考
        yield StreamEvent("step", {"phase": "thinking", "message": "正在思考..."})
        time.sleep(0.1)
        
        # 2. 模拟流式token生成
        response = self._generate_response(message)
        for char in response:
            yield StreamEvent("token", {"text": char})
            time.sleep(0.02)  # 模拟生成延迟
        
        # 3. 检查是否需要工具调用
        if self._needs_tool(message):
            yield StreamEvent("step", {"phase": "tool_calling"})
            tool_result = self._call_tool(message)
            yield StreamEvent("tool_end", {"result": tool_result})
        
        # 4. 完成
        yield StreamEvent("done", {"message": "生成完成"})
    
    def _generate_response(self, message):
        # 模拟LLM生成
        responses = {
            "你好": "你好！我是AI助手，很高兴为你服务。",
            "Python": "Python是一种高级编程语言，以简洁优雅著称。",
        }
        for key, resp in responses.items():
            if key in message:
                return resp
        return f"关于'{message}'，让我为你详细解答。"
    
    def _needs_tool(self, message):
        return any(kw in message for kw in ["搜索", "计算", "天气"])
    
    def _call_tool(self, message):
        for name, tool in self.tools.items():
            if any(kw in message for kw in tool["description"].split()):
                return tool["func"](message)
        return "工具调用完成"

class StreamCollector:
    # 流式输出收集器
    def __init__(self):
        self.tokens = []
        self.events = []
    
    def collect(self, stream: Generator[StreamEvent, None, None]) -> Dict:
        for event in stream:
            self.events.append(event)
            if event.type == "token":
                self.tokens.append(event.data["text"])
            elif event.type == "step":
                print(f"  📌 {event.data.get('message', event.data.get('phase', ''))}")
            elif event.type == "tool_end":
                print(f"  🔧 工具结果: {str(event.data['result'])[:60]}")
            elif event.type == "done":
                pass
        
        full_text = "".join(self.tokens)
        return {
            "text": full_text,
            "token_count": len(self.tokens),
            "event_count": len(self.events),
            "duration": self.events[-1].timestamp - self.events[0].timestamp if self.events else 0,
        }

# SSE格式模拟
def stream_to_sse(stream: Generator[StreamEvent, None, None]):
    # 将流式事件转为SSE格式
    for event in stream:
        sse_data = json.dumps({"type": event.type, "data": event.data}, ensure_ascii=False)
        yield f"data: {sse_data}\n\n"

# 测试
agent = StreamingAgent("助手")
agent.register_tool("search", lambda q: f"搜索结果：关于{q}", "搜索 查找")
agent.register_tool("calculator", lambda q: "42", "计算 数学")

# 测试流式输出
for msg in ["你好！", "搜索Python教程", "介绍一下Python"]:
    print(f"\n❓ {msg}")
    collector = StreamCollector()
    result = collector.collect(agent.stream_chat(msg))
    print(f"📝 {result['text']}")
    print(f"📊 {result['token_count']} tokens, {result['event_count']} events, {result['duration']:.2f}s")

# SSE格式测试
print("\n📡 SSE格式输出:")
for sse in stream_to_sse(agent.stream_chat("你好")):
    print(sse, end="")

🏋️ 实战练习

深入理解：流式输出核心原理

流式输出的三种协议对比：SSE(服务端推送，LLM场景最佳选择)、WebSocket(双向通信，LLM场景过重)、长轮询(延迟高不推荐)。流式Token处理Pipeline：LLM API - SSE Token流 - 缓冲区 - 逐Token显示，同时检测tool_call结构、Markdown格式、代码块边界。关键处理包括背压控制、错误恢复、格式检测。

进阶实现：流式输出引擎

以下是针对流式输出主题的进阶实现，包含Token流+工具调用检测+格式渲染+回调等核心功能。代码经过实机运行验证。

# StreamEngine - 流式输出进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class StreamEngine:
    # 流式输出进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = StreamEngine({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

常见问题FAQ

流式输出最佳实践

1. 理解原理再实践 - 先搞清楚为什么再动手实现
2. 渐进式复杂化 - 先让最简版本跑通再逐步优化
3. 错误处理优先 - 假设一切都会失败提前做好准备
4. 可观测性从Day1 - 不要等出问题才加监控
5. 文档即代码 - 好的文档和好的代码一样重要
6. 持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：流式输出的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。