本地大模型落地核心:Token可控调度与数据主权工程实践
2026/10/3 5:43:15
ChatGPT 不是黑魔法,它只是一套把“文字接龙”做到极致的工程系统。想把它用到生产环境,得先拆成三块积木:
把这三件事串起来,就能解释为什么 ChatGPT 能听懂上下文、会拒绝、偶尔还会“胡说八道”——它其实一直在做“概率最高的下一个词”。
stream=True打开,前端边收边渲染;后端同时起异步任务预生成“下一步提示”,把 CPU 等时间藏在网络延迟里。temperature=0.7保证多样,检测到“需要给出明确政策”关键词时,把温度降到 0.2 并注入知识库引用,减少翻车。下面代码把“重试、流式、超时、截断”全包圆,复制即可跑。
import openai, tiktoken, time, os from tenacity import retry, stop_after_attempt, wait_exponential openai.api_key = os.getenv("OPENAI_API_KEY") encoding = tiktoken.encoding_for_model("gpt-3.5-turbo") MAX_TOKENS = 3500 TARGET_REPLY = 500 TEMPERATURE = 0.4 def num_tokens(messages): return sum(len(encoding.encode(m["content"])) for m in messages) def slide_history(messages, max_tokens): """丢太远的轮次,保留 system 和最近几条""" sys_msg = [m for m in messages if m["role"] == "system"] others = [m for m in messages if m["role"] != "system"] while num_tokens(sys_msg + others) > max_tokens: others.pop(0) return sys_msg + others @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) def chat_completion(messages, stream=True, timeout=25): messages = slide_history(messages, MAX_TOKENS - TARGET_REPLY) return openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=messages, temperature=TEMPERATURE, max_tokens=TARGET_REPLY, stream=stream, request_timeout=timeout ) def reply(user_input, context): context.append({"role": "user", "content": user_input}) try: resp = chat_completion(context, stream=True) reply_text = "" for chunk in resp: delta = chunk.choices[0].delta if delta.get("content"): reply_text += delta.content print(delta.content, end="", flush=True) # 前端可换 WebSocket 推送 context.append({"role": "assistant", "content": reply_text}) return reply_text except openai.error.RateLimitError: return "系统繁忙,请稍后再试" except Exception as e: return f"服务异常:{e}" if __name__ == "__main__": ctx = [{"role": "system", "content": "你是 ACME 商城客服,说话简洁友好。"}] while True: user = input("\nUser: ") if user == "q": break reply(user, ctx)要点说明:
tiktoken精确计数,避免“目测” token 导致超限。tenacity做指数退避,防止突发 429/502 把前端带崩。文章看到这里,你已经知道:
下一步,轮到你把对话能力嫁接到自己的场景:电商客服、售后工单、内部知识库、游戏 NPC、甚至英语口语陪练。先跑通 MVP(最小可用原型),再按真实流量一点点把“温度”降下来,把“速度”提上去,把“成本”压下来。
如果你更想“先跑起来再优化”,可以试下这条捷径:从0打造个人豆包实时通话AI。实验把 ASR→LLM→TTS 整条链路包成可执行项目,本地装好依赖就能对着麦克风说话,亲测十分钟出效果。等跑通后,再把今天文章里的缓存、截断、状态机方案移植过去,就能快速得到一个低延迟、可扩展、可定制的语音对话系统。祝你玩得开心,记得把踩到的新坑也写成笔记分享出来。