19 · AI 服务端集成：LLM 调用 / 流式 / 函数调用

你在前端调过 LLM，现在要从后端把模型能力封装成稳定、可测、可控成本的 API。本篇讲 LLM 调用的服务端工程化。

速览卡

项	内容
一句话定位	把"调 LLM"从随手脚本变成生产级后端能力
核心内容	LLM 调用封装、流式响应（SSE）、函数调用（tools）、超时/重试/限流、密钥与审计
读完能回答	后端为什么需要再包一层？流式怎么发给前端？模型工具调用怎么做才稳？
相关板块	C09 · 服务端语言上手 ｜ C12 · 错误处理日志配置 ｜ G · AI 域

---

一、为什么后端要包一层 LLM

前端直连模型有这些痛点，后端统一收口解决：

痛点	后端收口
密钥泄露	模型密钥只在服务端
参数散落	Prompt/模型/参数统一管理
成本失控	统一限流、配额、审计
无缓存/重试	服务端做语义缓存与重试
前端多端	一个 API 服务所有客户端

原则：前端永远不拿模型密钥。所有 AI 能力都走后端 API。

---

二、LLM 调用封装

import httpx
from pydantic import BaseModel, Field


class ChatRequest(BaseModel):
    messages: list[dict]
    temperature: float = Field(0.7, ge=0, le=2)
    stream: bool = False


async def call_llm(messages, temperature=0.7, timeout=30):
    async with httpx.AsyncClient(timeout=timeout) as client:
        resp = await client.post(
            os.environ["LLM_BASE_URL"] + "/chat/completions",
            headers={"Authorization": f"Bearer {os.environ['LLM_API_KEY']}"},
            json={"model": os.environ["LLM_MODEL"], "messages": messages, "temperature": temperature},
        )
        resp.raise_for_status()
        return resp.json()["choices"][0]["message"]["content"]

要点：超时必设（模型可能很慢）、错误要按 C12 转成统一错误体、密钥从环境变量读。

---

三、流式响应：SSE 给前端"打字机"

from fastapi.responses import StreamingResponse


@app.post("/chat")
async def chat(req: ChatRequest):
    async def event_stream():
        async with httpx.AsyncClient(timeout=60) as client:
            async with client.stream("POST", url, headers=..., json=...) as resp:
                async for line in resp.aiter_lines():
                    if line.startswith("data: "):
                        yield f"data: {line[6:]}\n\n"

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

SSE 类比：前端 EventSource/fetch 读流，服务端 text/event-stream 逐块吐。比 WebSocket 更适合"单向持续输出"（网络域 N 规划中，暂以 C09 异步 的 asyncio 心智为准）。

---

四、函数调用（tools）：让模型能"干活"

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查询城市天气",
        "parameters": {"type": "object", "properties": {"city": {"type": "string"}}},
    },
}]

调用循环：

① 用户提问 → ② 带 tools 调模型 → ③ 模型返回 tool_calls
④ 执行真实函数 → ⑤ 把结果喂回模型 → ⑥ 模型给出最终回答

安全红线：模型只是"提出"调用，执行必须由你的代码校验。别让模型直接执行任意命令/SQL。

---

五、超时 / 重试 / 限流

机制	做法
超时	模型调用设超时，超时返回可重试错误
重试	网络错误/5xx 重试，业务错误（超长）不重试
限流	按用户/密钥限调用次数，超限 429
熔断	模型服务连续失败时快速失败

for attempt in range(3):
    try:
        return await call_llm(...)
    except httpx.ConnectError:
        if attempt == 2:
            raise
        await asyncio.sleep(0.5 * (attempt + 1))

限流与熔断见 C05 与 C21。

---

六、审计与可观测

• 每次调用记录：用户、模型、token 数、耗时、成本
• 敏感输入输出脱敏后进日志
• 与 C08 的指标打通（延迟/错误率/成本）

AI 服务的"日志"尤其重要：模型是非确定的，出问题要能回放"当时给了什么输入、模型回了什么"。

---

自检清单

• [ ] 能说清后端包一层 LLM 的五个理由
• [ ] 会封装带超时、密钥、错误处理的 LLM 调用
• [ ] 会用 SSE 实现流式输出
• [ ] 会配 tools 并实现"执行 → 喂回 → 最终回答"的循环
• [ ] 有超时/重试/限流/审计，模型调用可回放

常见误区

• 误区：前端直连模型就行。 密钥泄露 + 成本失控 + 无法统一治理，必须后端收口。
• 误区：流式就是 WebSocket。 单向持续输出用 SSE 更简单，不需要双向通道。
• 误区：模型能安全执行工具。 模型只提"建议"，执行和校验永远在你的代码里。
• 误区：模型调用不需要超时。 不设超时的调用可能挂死你的请求。

动手练习

练习 A：Chat API

实现 POST /chat（非流式），带 Pydantic 校验、超时、统一错误体。

练习 B：SSE 流式

把 /chat 改成 SSE 流式，前端用 fetch 逐块渲染打字机效果。

练习 C：工具调用

给模型配一个 get_weather 工具，跑通"提问 → 调工具 → 喂回 → 回答"全流程，并在日志记录工具调用。

---

相关文档

本区

• 09 · 服务端语言上手 —— FastAPI 异步与 SSE
• 12 · 错误处理日志配置 —— 错误体与审计日志
• 21 · AI 服务可靠性 —— 成本/限流/语义缓存

其他域

• 01 · LLM-API 实操入门 —— 模型调用入门
• 03 · Agent 与协议生态 —— 函数调用的原理