Python大模型API工程化实战：流式输出、结构化输出与上下文管理

1. 从“能跑通”到“敢上生产”：为什么第二篇要聊这些

很多人第一次用 Python 调大模型接口，跑通一个client.chat.completions.create就觉得自己会了。我当初也这么想，直到把一个内部问答工具交给同事用，才发现问题全在后面：接口偶尔超时、长对话越聊越贵、模型返回的 JSON 解析不了、并发一上来就限流。这些坑，第一篇讲基础调用时根本不会遇到，但只要你打算把东西真正用起来，就绕不开。

这篇是“Python GPT 秘籍”的第二篇，核心关键词是Python、大模型 API、流式输出、结构化输出、上下文管理、重试与限流、成本控制。它不讲怎么注册账号、怎么装库，那些第一篇已经说过了。这里要解决的是：当你已经能调通接口之后，怎么让代码从“玩具”变成“工具”。适合两类人看——一类是写过几个 Demo 但一上量就翻车的开发者，另一类是想把大模型能力接进自己项目、却不知道工程上该注意什么的工程师。

我自己的经验是，大模型应用 80% 的坑不在模型本身，而在你围绕它写的那些“胶水代码”。流式怎么处理、上下文怎么裁剪、失败怎么重试、输出怎么约束，这些才是决定你的项目能不能稳定跑下去的关键。下面按我实际踩坑的顺序，一块一块拆开讲。

2. 流式输出：别只盯着“打字机效果”，先想清楚数据怎么落地

2.1 流式到底解决了什么问题，以及它不解决什么

流式输出（streaming）最直观的好处是首字延迟低，用户不用干等十几秒才看到第一个字。但很多人把它当成纯粹的“体验优化”，这就想窄了。在真实项目里，流式还承担一个隐性职责：让长回答的生成过程可被中断、可被观测。比如用户看到一半发现方向不对，可以直接点停止，你就能提前掐断请求，省下后面的 token 费用。

但流式不解决的问题也很明确：它不会让你的总耗时变短，反而因为分块传输，整体完成时间可能略长一点点。它也不负责帮你拼装完整结果——每个 chunk 是碎片，你得自己攒。我见过有人以为开了 stream 就能自动拿到完整 JSON，结果拿到一堆半截字符串，解析全崩。

2.2 一个能直接抄的流式处理骨架

下面这段是我现在项目里常用的结构，核心思路是：边收边拼、边拼边判断、异常时保留已收内容。

import openai client = openai.OpenAI(api_key="你的key", base_url="你的兼容地址") def stream_chat(messages, model="gpt-4o-mini"): full_content = [] try: stream = client.chat.completions.create( model=model, messages=messages, stream=True, timeout=30, ) for chunk in stream: delta = chunk.choices[0].delta if delta and delta.content: full_content.append(delta.content) yield delta.content except Exception as e: # 关键：异常时把已经收到的内容交出去，而不是丢掉 partial = "".join(full_content) raise RuntimeError(f"流式中断，已收到 {len(partial)} 字：{e}") from e finally: # 这里可以做落库、计费等收尾 pass

这段代码有几个细节值得说。第一，full_content用列表攒而不是字符串拼接，因为字符串在循环里+=会产生大量临时对象，长回答时性能差距很明显。第二，finally里做收尾，保证无论正常结束还是异常退出，计费和日志都能落下来。第三，异常时把已收内容带出去，用户至少能看到半截答案，而不是一片空白。

2.3 流式场景下最容易忽略的三个坑

第一个坑是超时设置。流式请求如果设了很短的 timeout，长回答会在中途被掐断。我的做法是把 timeout 设成“首包超时”和“整体超时”两个概念，很多 SDK 支持分别配置，首包给 10 秒，整体给 120 秒。

第二个坑是前端缓冲。后端明明是一块块推的，前端却等全部收完才渲染，用户还是干等。这通常是中间层（比如某些网关）开了缓冲。排查方法是看响应头有没有X-Accel-Buffering: no之类的设置，没有就手动关掉。

第三个坑是流式下的错误处理。非流式请求失败会直接抛异常，流式请求可能已经推了一半才失败。这时候你不能简单重试，否则用户会看到重复内容。我的策略是：如果已推送内容超过一定比例（比如 30%），就不重试，直接告知用户“回答中断，请重试”；如果刚开始就失败，才允许静默重试。

提示：流式接口的计费和非流式一样按 token 算，但如果你中途掐断，已经生成的部分照样计费。所以“停止生成”能省钱，但省的是没生成的那部分，不是全部。

3. 结构化输出：让模型乖乖吐 JSON 的几种手段和它们的边界

3.1 为什么“请返回 JSON”这句话经常不管用

你肯定写过这种 prompt：“请以 JSON 格式返回，包含 name、age 两个字段。”大部分时候模型会照做，但总有那么几次，它给你包一层 ```json 代码块，或者在 JSON 前后加一句“好的，以下是结果：”。你的json.loads直接报错。

这不是模型不听话，而是它本质上在“续写文本”，不是在“执行函数”。你让它返回 JSON，它理解成“生成看起来像 JSON 的文本”，而文本里多几个字对它来说完全合理。所以要稳定拿到结构化数据，得用更强的约束手段。

3.2 三种约束手段的对比与选型

我现在的默认选择是：能用结构化输出就用结构化输出，退而求其次用函数调用，JSON Mode 只在临时脚本里用。原因很简单，前两者在服务端就帮你保证了格式合法性，你少写一堆 try-except。

3.3 用函数调用拿结构化数据的完整示例

tools = [{ "type": "function", "function": { "name": "extract_user_info", "description": "从文本中提取用户信息", "parameters": { "type": "object", "properties": { "name": {"type": "string", "description": "用户姓名"}, "age": {"type": "integer", "description": "用户年龄"}, "city": {"type": "string", "description": "所在城市"} }, "required": ["name", "age"] } } }] resp = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "张三今年 28 岁，住在杭州"}], tools=tools, tool_choice={"type": "function", "function": {"name": "extract_user_info"}} ) import json args = resp.choices[0].message.tool_calls[0].function.arguments data = json.loads(args) print(data) # {'name': '张三', 'age': 28, 'city': '杭州'}

注意tool_choice这里强制指定了函数，这样模型不会跑去闲聊。如果你有多个函数，可以让模型自己选，但那样就多了一层不确定性。生产环境里，我倾向于一个请求只干一件事，函数定义得越窄，输出越稳。

3.4 结构化输出翻车时的兜底策略

即便用了 schema，也不是 100% 保险。模型可能返回null，可能把数字写成字符串，可能漏掉必填字段。所以我在解析层永远包一层校验：

def safe_parse(raw, required_fields): try: data = json.loads(raw) except json.JSONDecodeError: return None, "JSON 解析失败" missing = [f for f in required_fields if f not in data or data[f] is None] if missing: return None, f"缺少字段：{missing}" return data, None

拿到None和错误信息后，我的处理是带错误信息重试一次，把“上次你漏了 age 字段”写进新一轮 prompt。实测下来，这种“纠错重试”能把成功率从 90% 拉到 99% 以上。但要注意重试次数上限，我一般设 2 次，再多就是浪费钱。

4. 上下文管理：长对话不是“全塞进去”就完事

4.1 上下文窗口的真实成本结构

很多人对上下文的理解停留在“模型能装多少字”。但工程上你要算的是三笔账：token 费用、延迟、以及信息密度。一个 10 轮对话，如果每轮都把历史全带上，第 10 轮的输入 token 可能是第 1 轮的 10 倍。费用是平方级增长的，延迟也跟着涨。

更麻烦的是，历史里大量内容是寒暄和重复，真正有用的信息可能就几句。全塞进去不仅贵，还会稀释模型的注意力，让它抓不住重点。我做过对比，同样的问题，把无关历史裁掉后，回答质量反而更高。

4.2 三种裁剪策略，按场景选

第一种是滑动窗口，只保留最近 N 轮。实现最简单，适合客服问答这类“只关心最近几轮”的场景。缺点是早期的重要信息会丢。

第二种是摘要压缩，把早期对话用模型总结成一段话，替代原始历史。适合长程对话，比如一个持续几天的项目助手。代价是要多调一次模型做摘要，有额外成本。

第三种是关键信息抽取，把对话里的事实（用户偏好、已确认的结论）抽成结构化字段，单独存起来，每轮只带这些字段。适合信息密度高、需要长期记忆的场景，比如个人助理。

我实际项目里最常用的是滑动窗口 + 关键信息抽取的组合：最近 5 轮原文保留，更早的内容抽成几个关键字段挂在 system prompt 里。这样既保住了近期上下文，又不会让 token 无限膨胀。

4.3 一个可复用的上下文管理器

class ContextManager: def __init__(self, max_turns=5, max_tokens=3000): self.history = [] self.facts = {} self.max_turns = max_turns self.max_tokens = max_tokens def add(self, role, content): self.history.append({"role": role, "content": content}) self._trim() def _trim(self): # 先按轮数裁 if len(self.history) > self.max_turns * 2: self.history = self.history[-self.max_turns * 2:] # 再按 token 粗估裁（中文约 1 字 1 token，英文约 4 字符 1 token） while self._estimate_tokens() > self.max_tokens and len(self.history) > 2: self.history.pop(0) def _estimate_tokens(self): return sum(len(m["content"]) for m in self.history) def build_messages(self, system_prompt): fact_text = "\n".join(f"{k}: {v}" for k, v in self.facts.items()) sys = system_prompt + (f"\n已知信息：\n{fact_text}" if fact_text else "") return [{"role": "system", "content": sys}] + self.history

这个类不复杂，但能解决大部分长对话的膨胀问题。_estimate_tokens用的是粗估，够用就行，真要精确可以用 tiktoken 之类的库，但会增加依赖，看你的项目取舍。

注意：裁剪历史时千万别把 system prompt 裁掉，也别把用户的最后一条消息裁掉。我见过有人无脑pop(0)，结果把系统设定删了，模型行为直接跑偏。

5. 重试、限流与降级：让接口抖动不至于拖垮整个服务

5.1 哪些错误该重试，哪些不该

不是所有失败都值得重试。我的分类是这样的：

• 该重试：超时、连接错误、429 限流、500/502/503 服务端错误。
• 不该重试：401 鉴权失败、400 参数错误、内容审核拦截。这些重试多少次都一样，纯浪费。

区分清楚这一点，能省下大量无谓的等待。我早期就是无脑重试，结果一个参数写错，程序卡在那里重试了半分钟才报错，排查时一脸懵。

5.2 指数退避 + 抖动的重试实现

import time import random def retry_with_backoff(func, max_retries=3, base_delay=1.0): for attempt in range(max_retries + 1): try: return func() except Exception as e: err = str(e).lower() retryable = any(k in err for k in ["timeout", "429", "500", "502", "503", "connection"]) if not retryable or attempt == max_retries: raise # 指数退避 + 随机抖动，避免多个请求同时重试打爆服务 delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5) time.sleep(delay)

这里的**抖动（jitter）**很关键。如果你有 100 个并发请求同时被限流，它们会在同一时刻一起重试，再次打爆服务。加上随机抖动，重试时间就散开了。这个技巧在分布式系统里很常见，放到大模型调用上同样管用。

5.3 限流下的并发控制

光有重试还不够，你得从源头控制并发数。我用的是信号量：

import threading semaphore = threading.Semaphore(5) # 最多 5 个并发 def limited_call(func): with semaphore: return func()

并发数设多少合适？我的经验是从 3 到 5 起步，观察 429 出现的频率再调。不同服务商的限流策略不一样，有的按 RPM（每分钟请求数），有的按 TPM（每分钟 token 数）。如果你不确定，宁可保守一点，因为 429 之后的重试等待比降并发更浪费时间。

5.4 降级方案：模型挂了，服务不能挂

再稳的接口也有抽风的时候。我的降级策略分三层：

第一层，换模型。主模型超时，自动切到更小更快的模型，虽然质量降一点，但至少能返回结果。

第二层，返回缓存。如果是重复问题，直接返回上次的结果，并标注“可能不是最新”。

第三层，友好报错。实在不行，给用户一句“服务繁忙，请稍后再试”，而不是抛一个堆栈出去。

这三层不是每个项目都要全上，但至少要有第三层。我见过太多项目，接口一挂，前端直接白屏，用户体验极差。

6. 成本控制：那些账单出来才后悔的事

6.1 先搞清楚钱花在哪

大模型调用计费主要看输入 token 和输出 token，输出通常比输入贵。所以省钱的两个方向是：减少输入和减少输出。减少输入靠上下文裁剪，前面讲过了。减少输出靠约束回答长度，比如在 prompt 里明确“用一句话回答”“不超过 50 字”。

但这里有个反直觉的点：有时候让模型多输出反而更省钱。比如你让它直接给答案，它可能绕一大圈；你让它先列要点再总结，它反而更快收敛。这个得靠实测，没有通用规律。

6.2 缓存能省下的钱超出你想象

如果你的应用有大量重复问题（比如 FAQ 类），缓存命中率可能到 50% 以上。缓存怎么做？简单点用问题文本的哈希做 key，复杂点用向量相似度匹配。我一般先用精确匹配，命中率不够再上语义匹配。

import hashlib cache = {} def get_cache_key(messages): text = "|".join(m["content"] for m in messages) return hashlib.md5(text.encode()).hexdigest() def cached_chat(messages, **kwargs): key = get_cache_key(messages) if key in cache: return cache[key] result = client.chat.completions.create(messages=messages, **kwargs) cache[key] = result return result

这个缓存很粗糙，但对付 FAQ 场景够用。要注意的是缓存要设过期时间，否则模型更新了你还返回老答案。我一般设 24 小时。

6.3 监控与告警：别等月底账单吓一跳

我现在每个项目都会记两个指标：每次调用的 token 数和累计费用。用简单的日志就能做：

def log_usage(resp, model): usage = resp.usage cost = usage.prompt_tokens * 0.000001 + usage.completion_tokens * 0.000002 # 示例单价 print(f"[{model}] 输入 {usage.prompt_tokens} 输出 {usage.completion_tokens} 费用约 {cost:.6f}")

单价按你实际用的模型填。有了这个日志，你就能看出哪些功能最烧钱，哪些 prompt 效率低。我靠这个发现过一个功能，因为 prompt 写得太啰嗦，每次多花 30% 的输入 token，改完直接省下一大笔。

提示：不同模型的单价差异很大，小模型可能只有大模型的十分之一。如果一个任务小模型能胜任，就别用大模型。我习惯先用小模型跑一遍，效果不行再升级。

7. 我在实际项目里踩过的几个真实坑

第一个坑是流式和非流式混用。有个功能我图省事，前端用流式，后端统计却按非流式解析，结果统计出来的字数永远是 0。排查了半天才发现，流式返回的是 chunk，根本没有完整的message.content。后来统一在流式结束时用攒好的内容做统计，问题才解决。

第二个坑是重试导致重复扣费。有次网络抖动，请求其实已经到达服务端并生成了结果，但客户端没收到响应就重试了，结果同一次调用被计费两次。后来我在请求里加了唯一 ID，服务端做幂等处理，才避免重复。如果你的服务商支持幂等键，一定要用上。

第三个坑是上下文裁剪把关键信息裁没了。有个多轮任务，用户第一轮说了预算范围，后面几轮都在聊细节。我的滑动窗口把第一轮裁掉后，模型开始推荐超预算的方案。后来我把“预算”这类关键字段抽出来单独存，才彻底解决。这也印证了前面说的：纯滑动窗口不够，得配合关键信息抽取。

第四个坑是JSON 解析没做兜底。有次模型返回的 JSON 里带了个尾随逗号，json.loads直接崩，整个请求失败。后来我加了个预处理，用正则把尾随逗号去掉，再解析。虽然不优雅，但很实用。更稳妥的做法是用支持宽松解析的库，或者干脆用结构化输出从源头避免。

这些坑的共同点是：它们都不在文档里，只有真正跑起来才会遇到。所以我的建议是，别等上线才发现问题，在开发阶段就故意制造一些异常（断网、超时、返回脏数据），看看你的代码扛不扛得住。

8. 把工程细节做扎实，比追新模型更值钱

模型每隔几个月就更新一代，但你这套围绕它的工程代码，可能要用上一两年。流式怎么处理、结构化输出怎么约束、上下文怎么裁剪、失败怎么重试、成本怎么控制，这些能力不会因为模型换代就失效。我见过太多人把精力全花在“试新模型”上，结果自己的调用层脆弱得不堪一击，换个模型就全线崩。

反过来，把这一层做扎实的人，换模型只需要改一个配置项。这才是真正的效率。如果你正在把大模型接进自己的项目，建议先把这篇里的几个模块落地：一个带异常处理的流式函数、一个带校验的结构化解析、一个会裁剪的上下文管理器、一个带抖动的重试封装。这四样东西搭起来，你的项目就从“能跑”变成“敢用”了。

后续如果还想深入，可以往两个方向走：一是做一套完整的评测集，每次换模型或改 prompt 都跑一遍，用数据说话；二是把调用层抽象成统一接口，让不同服务商可以无缝切换。这两个方向我都在做，等有成熟经验了再写第三篇。