AI 应用开发实战（1）：FastAPI + LLM API 从零搭建第一个 AI 应用，TaoToken 统一 Key 接入与流式输出 SSE 完整教程

1. 从零搭一个能流式输出的 AI 应用，到底难在哪

FastAPI + LLM API 这套组合，是很多人做 AI 应用开发的第一站。它能做什么？简单说，就是让浏览器里输入一句话，后端转发给大模型，再把模型一个字一个字吐出来的内容实时推回页面，形成打字机效果。适合谁？适合已经会一点 Python、想跑通第一个端到端 AI 应用的开发者，也适合想把多个模型统一到一个 Key 下管理的团队。

但真正动手时，卡点往往不在 FastAPI 本身，而在三件事：第一，模型供应商换来换去，Key 和 base_url 到处散落，改一处漏一处；第二，流式输出 SSE 写出来了，浏览器却要等全部生成完才显示，体验和同步请求没区别；第三，上下文越聊越长，token 超限报错，或者响应越来越慢。

这篇就按「统一 Key 接入 + 流式输出 SSE」这条主线，把 FastAPI 后端从零搭起来。我会给出可复制的 config.toml 与 settings.json 配置骨架、依赖清单、路由与流式响应代码，再用 curl 和浏览器双端验证，最后把常见报错逐个排掉。你跟着敲一遍，就能得到一个能对话、能流式打字的最小可用应用。

2. TaoToken 前置：把 Key 和通道统一起来

在写代码之前，先把「接入层」定下来。我试过把不同模型的 Key 直接写进业务代码，结果换模型时改了五六个文件，还漏了一个测试脚本。后来改成统一走一个 API 通道，业务代码只认一个 base_url 和一个 Key，切换模型只改配置。

TaoToken 在这里扮演的就是这个统一入口：它提供兼容 OpenAI SDK 的接口，你拿一个 Key，就能在同一个通道里调用不同模型。对 FastAPI 项目来说，好处是 llm.py 里只初始化一个 client，模型名从配置读，不用为每个供应商写一套适配。

你需要先拿到两样东西：API Key 和 base_url。Key 在控制台的 API Keys 页面创建，base_url 用https://taotoken.net/api。这两个值后面会写进.env或config.toml，不要硬编码进代码。

注意：Key 属于敏感信息，.env和config.toml都要加进.gitignore，别提交到仓库。

如果你还没创建 Key，可以先去控制台建一个，再回来继续。模型对话的在线体验入口也可以先点开看看返回格式，方便对照后面的 curl 结果。

3. 可复制配置：config.toml 与 settings.json 骨架

项目结构先定下来，后面所有文件都往这个目录里放：

ai-chat-app/ ├── main.py # FastAPI 入口 ├── llm.py # LLM API 封装层 ├── config.py # 配置加载 ├── config.toml # 主配置 ├── settings.json # 运行时可调参数 ├── requirements.txt ├── .env # 只放 Key，不提交 ├── .gitignore └── static/ └── index.html # 前端页面

依赖清单，直接复制：

fastapi==0.115.0 uvicorn[standard]==0.30.0 python-dotenv==1.0.0 openai==1.50.0 tomli==2.0.1

安装：

python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install -r requirements.txt

config.toml放不常变的接入配置：

[llm] base_url = "https://taotoken.net/api" model = "gpt-4o-mini" temperature = 0.7 max_tokens = 2048 timeout = 60 [app] site_name = "AI Chat" max_history_tokens = 4000

settings.json放运行时可调的参数，方便不改代码就调整：

{ "stream": true, "temperature": 0.7, "max_tokens": 2048, "system_prompt": "你是一个有用的 AI 助手，回答尽量简洁准确。" }

.env只放 Key：

LLM_API_KEY=sk-your-taotoken-key

配置加载层config.py，把 toml、json、env 三处合并：

import os import json import tomli from dotenv import load_dotenv load_dotenv() def load_config(): with open("config.toml", "rb") as f: cfg = tomli.load(f) with open("settings.json", "r", encoding="utf-8") as f: runtime = json.load(f) cfg["llm"]["api_key"] = os.getenv("LLM_API_KEY") cfg["runtime"] = runtime return cfg CONFIG = load_config()

这样设计的原因：接入信息（base_url、model）和运行时行为（temperature、system_prompt）分开，前者改配置重启，后者可以热更新。业务代码只读CONFIG，不关心值从哪来。

4. 封装 LLM 调用层与 FastAPI 流式路由

4.1 llm.py：同步与流式两个函数

不要在路由里直接写 API 调用，抽一层出来，换模型只改这一个文件：

from openai import OpenAI from config import CONFIG client = OpenAI( api_key=CONFIG["llm"]["api_key"], base_url=CONFIG["llm"]["base_url"], timeout=CONFIG["llm"]["timeout"], ) MODEL = CONFIG["llm"]["model"] def chat_sync(messages, temperature=None, max_tokens=None): resp = client.chat.completions.create( model=MODEL, messages=messages, temperature=temperature or CONFIG["llm"]["temperature"], max_tokens=max_tokens or CONFIG["llm"]["max_tokens"], ) return resp.choices[0].message.content def chat_stream(messages, temperature=None, max_tokens=None): resp = client.chat.completions.create( model=MODEL, messages=messages, temperature=temperature or CONFIG["llm"]["temperature"], max_tokens=max_tokens or CONFIG["llm"]["max_tokens"], stream=True, ) for chunk in resp: delta = chunk.choices[0].delta if chunk.choices else None if delta and delta.content: yield delta.content

chat_sync适合批量处理、后台任务；chat_stream适合对话场景，用户能立刻看到第一个字。

4.2 main.py：SSE 流式路由

SSE 的关键是media_type="text/event-stream"，以及每条消息以data:开头、以两个换行结尾。同时要加X-Accel-Buffering: no，否则 Nginx 会缓冲，流式就变成一次性返回。

import json from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse from fastapi.staticfiles import StaticFiles from config import CONFIG from llm import chat_sync, chat_stream app = FastAPI(title=CONFIG["app"]["site_name"]) app.mount("/static", StaticFiles(directory="static"), name="static") sessions = {} MAX_TOKENS = CONFIG["app"]["max_history_tokens"] def get_session(sid): if sid not in sessions: sessions[sid] = [{"role": "system", "content": CONFIG["runtime"]["system_prompt"]}] return sessions[sid] def trim_history(messages, max_tokens=MAX_TOKENS): total, kept = 0, [] for msg in reversed(messages): t = int(len(msg["content"]) / 1.5) if total + t > max_tokens: break total += t kept.append(msg) kept.reverse() if kept and kept[0]["role"] != "system": kept.insert(0, {"role": "system", "content": CONFIG["runtime"]["system_prompt"]}) return kept @app.get("/", response_class=HTMLResponse) async def index(): with open("static/index.html", "r", encoding="utf-8") as f: return f.read() @app.post("/api/chat") async def chat(request: Request): body = await request.json() sid = body.get("session_id", "default") text = body.get("message", "").strip() if not text: return JSONResponse({"error": "消息不能为空"}, status_code=400) history = get_session(sid) history.append({"role": "user", "content": text}) reply = chat_sync(trim_history(history)) history.append({"role": "assistant", "content": reply}) return JSONResponse({"reply": reply}) @app.post("/api/chat/stream") async def chat_stream_endpoint(request: Request): body = await request.json() sid = body.get("session_id", "default") text = body.get("message", "").strip() if not text: return JSONResponse({"error": "消息不能为空"}, status_code=400) history = get_session(sid) history.append({"role": "user", "content": text}) messages = trim_history(history) async def generate(): full = "" for chunk in chat_stream(messages): full += chunk yield f"data: {json.dumps({'delta': chunk}, ensure_ascii=False)}\n\n" history.append({"role": "assistant", "content": full}) yield "data: [DONE]\n\n" return StreamingResponse( generate(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", }, ) @app.post("/api/clear") async def clear(request: Request): body = await request.json() sessions[body.get("session_id", "default")] = [] return JSONResponse({"status": "ok"})

这里把 chunk 用json.dumps包了一层，是为了避免模型输出里带换行时破坏 SSE 格式。前端解析时取delta字段即可。

4.3 前端读取 SSE

前端用fetch+ReadableStream读，比EventSource更灵活，因为可以发 POST：

async function sendMessage() { const input = document.getElementById('message-input'); const text = input.value.trim(); if (!text) return; input.value = ''; const box = document.getElementById('chat-box'); const userDiv = document.createElement('div'); userDiv.className = 'message user'; userDiv.textContent = text; box.appendChild(userDiv); const aiDiv = document.createElement('div'); aiDiv.className = 'message assistant'; box.appendChild(aiDiv); const resp = await fetch('/api/chat/stream', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ session_id: 'web', message: text }), }); const reader = resp.body.getReader(); const decoder = new TextDecoder(); let buffer = ''; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); const parts = buffer.split('\n\n'); buffer = parts.pop(); for (const part of parts) { if (!part.startsWith('data: ')) continue; const data = part.slice(6); if (data === '[DONE]') continue; try { aiDiv.textContent += JSON.parse(data).delta; } catch (e) {} box.scrollTop = box.scrollHeight; } } }

5. 验证请求：curl 与浏览器双端跑通

先启动服务：

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

5.1 curl 验证非流式

curl -X POST http://localhost:8000/api/chat \ -H "Content-Type: application/json" \ -d '{"session_id":"test","message":"用一句话解释什么是 SSE"}'

预期返回：

{"reply":"SSE 是服务器向浏览器单向推送事件的 HTTP 技术，适合流式输出。"}

5.2 curl 验证流式

curl -N -X POST http://localhost:8000/api/chat/stream \ -H "Content-Type: application/json" \ -d '{"session_id":"test","message":"数到五"}'

-N关闭 curl 缓冲，你会看到data: {"delta": "1"}这样的行一条条冒出来，最后是data: [DONE]。如果所有内容一次性出现，说明缓冲没关掉，检查X-Accel-Buffering和 Nginx 的proxy_buffering off。

5.3 浏览器验证

打开http://localhost:8000，输入问题，观察文字是否逐字出现。如果页面白屏，看浏览器控制台有没有 404，多半是static/index.html路径不对。

6. 本篇常见错排查

报错一：openai.AuthenticationError: 401Key 没读到或写错。检查.env里LLM_API_KEY是否以sk-开头，config.py是否在load_dotenv()之后才读环境变量。用python -c "from config import CONFIG; print(CONFIG['llm']['api_key'][:8])"打印前八位确认。

报错二：Connection error或超时base_url 写错。确认是https://taotoken.net/api，不要多加/v1或漏掉协议头。网络层超时可以把config.toml里的timeout调到 120 再试。

报错三：流式不生效，一次性返回三个地方依次查：响应头有没有X-Accel-Buffering: no；Nginx 有没有proxy_buffering off；前端是不是用了resp.text()而不是getReader()。用 curl-N能流式、浏览器不能，基本就是 Nginx 缓冲。

报错四：context_length_exceeded上下文超限。trim_history的max_tokens调小，或者把int(len(content)/1.5)的估算系数调保守。中文实际 token 比字符少，1.5 是偏安全的估算。

报错五：SSE 消息被截断、JSON 解析失败模型输出里带换行，直接拼data: {chunk}会断行。本篇用json.dumps包了一层，前端JSON.parse取delta，就不会断。如果你自己改成了裸拼，记得转义换行。

报错六：多轮对话串会话session_id前端写死了。生产环境应该每个浏览器会话生成唯一 ID，或者用登录态绑定。内存字典sessions在多进程下不共享，上生产换 Redis。

7. 下一步：把统一 Key 用在长期编码与 Agent 场景

跑通这个最小应用后，你会发现接入层统一带来的好处：换模型只改config.toml一行，业务代码零改动。如果你接下来要做的是长期编码助手、Agent 工具调用这类持续消耗 token 的场景，单次按量计费可能不够划算，可以看看 Coding Plan 这类包月方案，把统一 Key 的通道用在更高频的调用上。

需要管理多个 Key、查看用量，去控制台；要新建或轮换 Key，去 API Keys 页面；想先在线试模型返回格式，用模型对话；接入细节和参数说明，查接入文档。Claude Code 相关的接入方式，在 ClaudeCodeAnthropic 页面有单独说明。

把这篇的代码存下来，改掉config.toml里的 model 字段，你就能在同一个 FastAPI 骨架里切换不同模型，而不用重写任何路由。下一步可以在这个基础上加 Prompt 模板、加 RAG 检索，接入层已经帮你省掉了最烦的那部分。