AI 工作流专栏 10：流式输出 SSE 与函数调用全解

data: 你

data: 好

data: ，世界

data: [DONE]

# stream_basic.py：流式调用，逐字打印
import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),                 # 用 DeepSeek 演示
    base_url="https://api.deepseek.com/v1",                # 兼容 OpenAI 协议
)

# 关键：stream=True 让返回值变成一个"生成器"，要遍历才拿得到数据
stream = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "用 100 字介绍杭州西湖。"}],
    stream=True,                                           # 就这一行
    stream_options={"include_usage": True},                # 让最后一个 chunk 带 token 用量
)

full_text = ""
usage = None
for chunk in stream:                                       # 逐个 chunk 遍历
    # 每个 chunk 是模型生成的一小段（可能是一个字、一个词）
    delta = chunk.choices[0].delta.content if chunk.choices[0].delta.content else ""
    if delta:
        full_text += delta
        print(delta, end="", flush=True)                   # 立即打印，不换行
    # 最后一个 chunk 没有 choices，只有 usage
    if chunk.usage:
        usage = chunk.usage

print(f"\n\n--- 完整结果 ---\n{full_text}")
print(f"token 用量：输入 {usage.prompt_tokens} + 输出 {usage.completion_tokens}")

ChatCompletionChunk(
    choices=[
        Choice(
            index=0,
            delta=ChoiceDelta(content="杭", role="assistant"),  # 这次新增的内容
            finish_reason=None,                                  # 还没结束
        )
    ],
    usage=None,  # 只有最后一个 chunk 才有 usage
)

# stream_adapter.py：流式适配器（简化版）
import os
from typing import Iterator, Optional
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()


def chat_stream(
    messages: list[dict],
    model: str = "deepseek-chat",
    api_key: Optional[str] = None,
    base_url: Optional[str] = None,
    temperature: float = 0.7,
) -> Iterator[str]:
    """流式对话：yield 出每个增量片段，调用方负责拼接。"""
    client = OpenAI(
        api_key=api_key or os.getenv("DEEPSEEK_API_KEY"),
        base_url=base_url or "https://api.deepseek.com/v1",
    )
    stream = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=temperature,
        stream=True,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:                       # 过滤掉空片段
            yield delta                 # 产出给调用方


# 用法
if __name__ == "__main__":
    for piece in chat_stream([{"role": "user", "content": "讲个程序员冷笑话"}]):
        print(piece, end="", flush=True)

pip install fastapi "uvicorn[standard]" openai python-dotenv

# server.py：FastAPI SSE 服务端
import os
import json
import asyncio
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

app = FastAPI()

# 允许前端跨域调用（开发时前端跑在别的端口）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],       # 生产环境请改成你的前端域名
    allow_methods=["*"],
    allow_headers=["*"],
)

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
)


class ChatRequest(BaseModel):
    message: str


def sse_format(text: str) -> str:
    """把文本包成标准 SSE 事件格式：data: xxx\n\n"""
    return f"data: {json.dumps({'content': text}, ensure_ascii=False)}\n\n"


@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    """SSE 流式聊天接口"""

    def event_generator():
        """生成器：边从 LLM 收，边按 SSE 格式吐"""
        try:
            stream = client.chat.completions.create(
                model="deepseek-chat",
                messages=[
                    {"role": "system", "content": "你是一位友好的助手，回答简洁。"},
                    {"role": "user", "content": req.message},
                ],
                stream=True,
            )
            for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    yield sse_format(delta)          # 每个增量包成一个 SSE 事件

            # 正常结束，推一个结束标记
            yield "data: [DONE]\n\n"
        except Exception as e:
            # 中途出错也要通过 SSE 告诉前端，而不是默默断开
            yield sse_format(f"\n\n[生成出错：{e}]")
            yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",              # SSE 的 MIME 类型，关键！
        headers={
            "Cache-Control": "no-cache",             # 禁用缓存
            "X-Accel-Buffering": "no",               # 告诉 nginx 别缓冲（见后文第五节）
            "Connection": "keep-alive",
        },
    )


# 启动：uvicorn server:app --reload --port 8000

uvicorn server:app --reload --port 8000

<!-- index.html：前端 SSE 消费（纯原生 JS，无需任何库） -->
<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8"><title>流式聊天 Demo</title></head>
<body>
  <div id="output" style="white-space: pre-wrap; font-size: 16px;"></div>
  <button id="ask">问一下杭州</button>

  <script>
    document.getElementById('ask').addEventListener('click', async () => {
      const output = document.getElementById('output');
      output.textContent = '';   // 清空旧内容

      // 用 fetch 发 POST，拿到的是一个流式 response
      const response = await fetch('http://localhost:8000/chat/stream', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ message: '用 100 字介绍杭州西湖' }),
      });

      // 拿到 response.body 这个 ReadableStream，逐块读取
      const reader = response.body.getReader();
      const decoder = new TextDecoder('utf-8');
      let buffer = '';

      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        // SSE 事件以 \n\n 分隔，可能一次读到多个事件，要按 \n\n 切开
        const parts = buffer.split('\n\n');
        buffer = parts.pop();      // 最后一段可能不完整，留着下次拼

        for (const part of parts) {
          const line = part.trim();
          if (!line.startsWith('data: ')) continue;
          const data = line.slice(6);     // 去掉 "data: " 前缀
          if (data === '[DONE]') { output.textContent += '\n[结束]'; return; }
          try {
            const obj = JSON.parse(data);
            output.textContent += obj.content;   // 追加显示，形成打字机效果
          } catch (e) { /* 忽略非 JSON 行 */ }
        }
      }
    });
  </script>
</body>
</html>

location /chat/stream {
    proxy_pass http://backend;
    proxy_buffering off;          # 关键：关闭缓冲
    proxy_cache off;              # 关闭缓存
    proxy_http_version 1.1;       # SSE 需要 HTTP/1.1 长连接
    proxy_set_header Connection "";
    proxy_read_timeout 300s;      # 流式响应可能很久，别让 Nginx 提前超时
    # 顺带禁用 gzip，避免压缩也导致攒包
    gzip off;
}

from fastapi import Request

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest, request: Request):
    async def event_generator():
        stream = client.chat.completions.create(
            model="deepseek-chat", messages=[{"role": "user", "content": req.message}],
            stream=True,
        )
        for chunk in stream:
            # 检查客户端是否已断开
            if await request.is_disconnected():
                print("客户端已断开，停止生成")
                break                      # 跳出循环，后端停止调 LLM
            delta = chunk.choices[0].delta.content
            if delta:
                yield sse_format(delta)
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

try:
    for chunk in stream:
        ...
except Exception as e:
    yield sse_format(f"\n\n[生成中断：{str(e)[:100]}]")
finally:
    yield "data: [DONE]\n\n"     # 无论成功失败，都要发结束标记

# tools_define.py：定义模型可用的工具
tools = [
    {
        "type": "function",                          # 固定值，表示这是一个函数工具
        "function": {
            "name": "get_weather",                   # 函数名（你的代码里要有同名函数）
            "description": "查询指定城市的实时天气。"
                           "支持中国主要城市。"
                           "当用户询问天气、是否下雨、穿什么衣服时调用。",  # 关键！模型靠它决策
            "parameters": {                          # 参数的 JSON Schema
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名，如 '北京'、'上海'、'杭州'",
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],   # 枚举，限定取值
                        "description": "温度单位，默认摄氏度",
                    },
                },
                "required": ["city"],                # city 是必填，unit 可选
            },
        },
    },
]

# function_calling_demo.py：完整 Function Calling 循环
import os
import json
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
)

# ---- 第 1 步：定义工具 schema ----
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的实时天气。当用户询问天气、是否下雨、穿衣建议时调用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名，如 '北京'、'上海'、'杭州'",
                    },
                },
                "required": ["city"],
            },
        },
    }
]


# ---- 第 2 步：真正实现这个函数（这里用 mock，生产里换成真实天气 API） ----
def get_weather(city: str) -> str:
    """实际项目中这里会调用天气 API（如和风天气、心知天气）。
    为了演示可跑，我们返回写死的数据。"""
    mock_data = {
        "北京": "晴，25°C，西北风 3 级",
        "上海": "多云，28°C，东南风 2 级",
        "杭州": "小雨，22°C，东风 4 级",
    }
    return mock_data.get(city, f"暂无 {city} 的天气数据")


# 用函数名做映射，方便根据模型返回的函数名找到真实函数
available_functions = {"get_weather": get_weather}


# ---- 第 3 步：开始对话 ----
messages = [
    {"role": "system", "content": "你是一个贴心的生活助手，回答要基于真实的天气数据。"},
    {"role": "user", "content": "我明天去杭州出差，那边天气怎么样？需要带伞吗？"},
]

# ---- 第 4 步：第一次调模型，看它要不要用工具 ----
print("【第 1 次调用：模型决策】")
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    tools=tools,                 # 把工具清单传给模型
    tool_choice="auto",          # auto = 让模型自己决定要不要调；也可强制 "none" / 指定函数
)
msg = response.choices[0].message

# ---- 第 5 步：检查模型是否要求调用工具 ----
if msg.tool_calls:
    # 重要：把模型这条"assistant"消息（含 tool_calls）原样加进 messages
    messages.append(msg)

    for tool_call in msg.tool_calls:
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)   # 参数是 JSON 字符串，要解析
        print(f"  模型想调用：{func_name}({func_args})")

        # 执行真实函数
        func = available_functions.get(func_name)
        if func:
            result = func(**func_args)
        else:
            result = f"未知函数：{func_name}"
        print(f"  函数返回：{result}")

        # ---- 第 6 步：把函数结果作为 role=tool 消息喂回 ----
        messages.append({
            "role": "tool",                       # 关键：role 必须是 "tool"
            "tool_call_id": tool_call.tool_call_id,  # 必须对上刚才那个 tool_call 的 id
            "content": str(result),               # 函数返回值转成字符串
        })

    # ---- 第 7 步：第二次调模型，让它基于工具结果生成最终回答 ----
    print("\n【第 2 次调用：模型生成最终回答】")
    final_response = client.chat.completions.create(
        model="deepseek-chat",
        messages=messages,
        tools=tools,      # 也可以再带上，允许它继续调
    )
    print(final_response.choices[0].message.content)
else:
    # 模型没要求调工具，直接就是最终回答
    print(msg.content)

【第 1 次调用：模型决策】
  模型想调用：get_weather({'city': '杭州'})
  函数返回：小雨，22°C，东风 4 级

【第 2 次调用：模型生成最终回答】
杭州明天是小雨，气温 22°C 左右，东风 4 级。建议您带上雨伞，穿一件薄外套……

# parallel_function_call.py：并行执行多个函数调用
import os
import json
from concurrent.futures import ThreadPoolExecutor
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com/v1")

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查询指定城市的实时天气。",
        "parameters": {
            "type": "object",
            "properties": {"city": {"type": "string", "description": "城市名"}},
            "required": ["city"],
        },
    },
}]

def get_weather(city: str) -> str:
    data = {"北京": "晴 25°C", "上海": "多云 28°C", "杭州": "雨 22°C"}
    return data.get(city, "未知")

messages = [
    {"role": "user", "content": "对比北京、上海、杭州三个城市的天气，哪里最适合明天出游？"},
]

resp = client.chat.completions.create(
    model="deepseek-chat", messages=messages, tools=tools, tool_choice="auto",
)
msg = resp.choices[0].message

if msg.tool_calls:
    messages.append(msg)

    # 用线程池并行执行所有 tool_calls（IO 密集型，线程池合适）
    def run_one(tc):
        args = json.loads(tc.function.arguments)
        result = get_weather(**args)
        return {"role": "tool", "tool_call_id": tc.tool_call_id, "content": str(result)}

    with ThreadPoolExecutor(max_workers=4) as pool:
        tool_messages = list(pool.map(run_one, msg.tool_calls))

    messages.extend(tool_messages)          # 一次性把所有结果喂回

    # 第二次调用，生成对比回答
    final = client.chat.completions.create(model="deepseek-chat", messages=messages, tools=tools)
    print(final.choices[0].message.content)