2026 年上半年智谱营收达到 9.54 亿元同比增长 399.7%亏损进一步收窄到 20.71 亿元这个数据在 AI 圈子里引发了不少讨论。很多人第一反应是“大模型公司终于开始赚钱了”但作为技术开发者我更关心的是当一家大模型厂商进入商业化快车道后我们这些做应用的人应该怎样把大模型能力真正落地到业务里本文不打算聊财报解读而是以智谱开放平台为切入点拆解大模型 API 接入、对话应用开发、流式响应、成本控制和企业级落地的完整流程。无论你是刚接触大模型开发的初学者还是已经在做 AI 应用集成的中级开发者都可以从里面找到可以直接复用的代码和排错思路。1. 为什么大模型厂商增长快应用开发更要稳1.1 智谱增长背后的技术信号营收同比增长接近 4 倍亏损收窄意味着大模型服务已经从“技术展示”走向“真实生产”。在过去一年里很多企业不再观望而是把大模型 API 接入客服、知识库、审批辅助、代码生成等具体流程中。智谱的 GLM 系列模型在中文语义理解、长文本处理、复杂指令跟随方面表现稳定所以成了很多国内开发者的首选。从技术角度看这种增长其实带来了一个新的挑战当 API 调用量变大、业务场景变复杂我们不能只停留在“调一下接口返回结果”的玩具阶段而是要考虑并发、超时、限流、成本、内容安全、数据隐私这些问题。换句话说大模型厂商卖的是“模型能力”而我们要构建的是“稳定的应用系统”。1.2 本文适合的读者如果你符合下面任意一条这篇文章就是为你准备的想快速把大模型 API 接入自己的 Python 后端服务已经在用智谱或其他国内大模型平台但想优化流式输出和错误处理需要在公司内部落地一个 AI 客服、文档助理或知识库问答系统对 token 计费、并发控制、数据安全等工程问题没有完整思路。文章会围绕一个核心项目展开用 Python 构建一个可扩展的大模型 API 接入层包含同步对话、流式输出、历史消息管理、成本预估和异常重试。整个项目不需要复杂框架但结构上可以直接迁移到 FastAPI、Django 等生产环境。2. 环境准备与账号申请2.1 运行环境本文示例代码的本地环境如下版本不需要完全一致重点是思路操作系统Windows 11 / macOS 14 / Ubuntu 22.04 均可语言版本Python 3.10依赖库requests、openai以 OpenAI 兼容协议为例可选工具FastAPI、Redis用于生产限流与缓存建议新建一个虚拟环境避免污染系统 Pythonpython -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install requests openai2.2 获取 API Key大模型平台目前普遍采用 API Key 鉴权。你需要登录智谱开放平台在控制台完成实名认证后创建 API Key。Key 是敏感信息不要把 Key 写进代码或提交到 Git 仓库。推荐使用环境变量export ZHIPU_API_KEY你的_API_Key2.3 两种调用方式说明智谱开放平台提供了官方 SDK同时也支持 OpenAI 兼容的接口规范。这意味着你之前用过 OpenAI SDK 的项目只需要修改base_url和api_key就能切换到智谱的模型服务。这是当前国内大模型平台的主流做法好处是迁移成本低、社区生态丰富。本文示例以 OpenAI 兼容接口为例核心代码不绑定特定厂商。如果你更习惯官方 SDK原理也是相通的。3. 大模型 API 接入的核心概念3.1 Token 与上下文窗口大模型处理文本时并不是按字数计费而是按 token 计费。Token 可以理解为模型的最小语义单元中文通常一个汉字对应 1 到 2 个 token英文一个单词往往对应 1 到 2 个 token。上下文窗口决定了模型一次能接收的最大 token 数量包括用户输入和模型输出。开发时要注意如果输入太长会超出窗口限制服务端会返回错误。所以对大文本需要做截断或摘要处理。3.2 消息结构一次对话请求通常由多条消息组成每条消息包含role和content。常见的 role 有三种role含义system系统设定用来约束模型行为和回答风格user用户输入assistant模型的历史回复构建对话时要按时间先后排列消息列表。不能把 system 消息放在中间也不要让 user 连续出现多条而缺少 assistant 响应否则部分模型会表现异常。3.3 同步调用与流式调用同步调用发起请求后等待完整回复适合内部测试、离线任务。流式调用服务端逐段返回 token前端可以实时显示打字机效果大幅降低首字延迟感。流式调用在生产环境更常用因为它能让用户更早看到输出体验更好。但流式调用的开发复杂度更高必须处理好数据切片和中断恢复。4. 完整实战搭建大模型 API 接入层下面我们逐步实现一个可运行的 Python 模块。这个模块可以独立测试也可以作为 FastAPI 接口的服务层。4.1 项目结构llm-demo/ ├── config.py # 配置项 ├── llm_client.py # 大模型调用封装 ├── chat_service.py # 业务服务层 ├── test_openai.py # 同步调用测试 └── test_stream.py # 流式调用测试4.2 配置文件# config.py import os API_KEY os.getenv(ZHIPU_API_KEY, ) BASE_URL os.getenv(ZHIPU_BASE_URL, https://open.bigmodel.cn/api/paas/v4) MODEL_NAME os.getenv(ZHIPU_MODEL, glm-4-plus) TIMEOUT 30注意MODEL_NAME建议根据智谱官方文档选择一般模型 ID 会随版本更新。代码里把模型名做成环境变量方便切换。4.3 封装 OpenAI 兼容客户端这里我们直接用openai库来调用# llm_client.py from openai import OpenAI import config client OpenAI( api_keyconfig.API_KEY, base_urlconfig.BASE_URL, timeoutconfig.TIMEOUT, ) def chat(messages, temperature0.7, max_tokens1024): 同步调用对话模型 :param messages: 消息列表格式为 [{role: system, content: ...}] :param temperature: 采样温度越高越随机 :param max_tokens: 最大生成 token 数 :return: 模型回复文本 if not config.API_KEY: raise ValueError(缺少 ZHIPU_API_KEY 环境变量) response client.chat.completions.create( modelconfig.MODEL_NAME, messagesmessages, temperaturetemperature, max_tokensmax_tokens, ) return response.choices[0].message.content这里有一个容易被忽略的坑max_tokens指的是生成回复的最大 token 数不包含输入 token 数。设置太小会导致回答被截断设置太大会增加单次请求的成本。4.4 添加流式输出# llm_client.py 中新增函数 def chat_stream(messages, temperature0.7, max_tokens1024): 流式调用对话模型返回一个迭代器 每次迭代返回一个字符串片段 if not config.API_KEY: raise ValueError(缺少 ZHIPU_API_KEY 环境变量) response client.chat.completions.create( modelconfig.MODEL_NAME, messagesmessages, temperaturetemperature, max_tokensmax_tokens, streamTrue, ) for chunk in response: # 不同 SDK 版本的 chunk 结构略有差异需要以实际打印为准 delta chunk.choices[0].delta if delta and delta.content: yield delta.content流式响应的每一块是流式传输的一部分不能直接把整段 buffer 起来再一次性返回否则就失去了流式的意义。正确做法是通过生成器逐段返回给上层调用方例如 FastAPI 的StreamingResponse。4.5 构建业务服务层实际项目中我们不会直接在生产代码里到处调用chat()而是会做一个服务层统一处理消息历史、多轮对话、异常捕获。# chat_service.py from typing import List, Dict from llm_client import chat, chat_stream SYSTEM_PROMPT 你是一个专业的 AI 助手请用简洁准确的中文回答问题。 class ChatService: def __init__(self, system_prompt: str SYSTEM_PROMPT): self.system_prompt system_prompt def build_messages( self, user_input: str, history: List[Dict[str, str]] None ) - List[Dict[str, str]]: 构建符合模型要求的消息列表 history 是之前的多轮对话格式 [{role: user, content: ...}, {role: assistant, content: ...}] messages [] if self.system_prompt: messages.append({role: system, content: self.system_prompt}) if history: messages.extend(history) messages.append({role: user, content: user_input}) return messages def get_answer(self, user_input: str, history: List[Dict[str, str]] None) - str: try: messages self.build_messages(user_input, history) answer chat(messages) return answer except Exception as e: # 生产环境应记录日志并降级处理 return f抱歉服务暂时不可用{e} def get_answer_stream(self, user_input: str, history: List[Dict[str, str]] None): messages self.build_messages(user_input, history) return chat_stream(messages)为什么要单独抽出build_messages因为大多数场景下我们需要维护多轮语境而消息列表的拼装逻辑是复用的。后续如果要接入向量检索、知识库可以在build_messages中注入检索结果形成“检索增强生成”RAG的基础链路。4.6 编写测试脚本# test_openai.py from chat_service import ChatService if __name__ __main__: service ChatService() # 第一轮 answer1 service.get_answer(介绍一下大模型 token 的概念) print(Assistant 1:, answer1) # 第二轮携带历史消息 history [ {role: user, content: 介绍一下大模型 token 的概念}, {role: assistant, content: answer1}, ] answer2 service.get_answer(那我怎么计算 token 数量, history) print(Assistant 2:, answer2)运行方式python test_openai.py如果配置正确你会看到模型连续回答两轮并且第二轮能结合第一轮对话内容进行补充。如果第二轮回答完全没有上下文关联优先检查history里的内容是否完整是否把 assistant 回复漏掉了。# test_stream.py from chat_service import ChatService if __name__ __main__: service ChatService() parts [] for text in service.get_answer_stream(用一句话解释什么是并发): print(text, end, flushTrue) parts.append(text) print(\n完整结果:, .join(parts))运行后界面会像打字机一样逐字输出。这个效果在前端 Web 页面中尤其重要。5. 接入 FastAPI 提供 HTTP 服务单机测试没问题后下一步是封装成 HTTP 接口方便前端或其他后端服务调用。这里使用 FastAPI完整示例代码如下# main.py from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import List, Dict, Optional from chat_service import ChatService app FastAPI() service ChatService() class ChatRequest(BaseModel): user_input: str history: Optional[List[Dict[str, str]]] [] class ChatResponse(BaseModel): answer: str app.post(/chat, response_modelChatResponse) async def chat_with_llm(req: ChatRequest): answer service.get_answer(req.user_input, req.history) return ChatResponse(answeranswer) app.post(/chat/stream) async def chat_with_llm_stream(req: ChatRequest): def gen(): for text in service.get_answer_stream(req.user_input, req.history): yield fdata: {text}\n\n return StreamingResponse(gen(), media_typetext/event-stream)启动服务uvicorn main:app --host 0.0.0.0 --port 8000通过/chat/stream接口前端可以使用 EventSource 或 fetch 流式读取结果实现类似各种 AI 助手页面的实时输出。注意这个示例没有加鉴权和限流正式环境必须补上。6. 成本控制与性能优化6.1 Token 成本估算大模型 API 按 token 计费成本主要来自输入和输出两部分。一个常见误区是只关注模型输出 token忽略了系统提示词和对话历史里的 token 消耗。可以在请求前后分别读取用量字段def chat_with_usage(messages, **kwargs): response client.chat.completions.create( modelconfig.MODEL_NAME, messagesmessages, **kwargs ) usage response.usage print(f输入 tokens: {usage.prompt_tokens}) print(f输出 tokens: {usage.completion_tokens}) print(f总 tokens: {usage.total_tokens}) return response.choices[0].message.content, usage在业务中建议把每次调用的 token 用量写入日志表按天汇总。这样才能精确评估不同场景的单次成本并为后面做缓存策略提供数据支撑。6.2 控制上下文长度多轮对话会随轮次越来越长成本也会指数上升。常见优化策略保留最近 N 轮对话更早的历史存到数据库对超长文本先做摘要再把摘要放进上下文设置max_tokens上限避免无意义的长回复把系统提示词精简到最必要程度。6.3 缓存与重试对于相同或高度相似的请求可以引入缓存。例如把“问题”的 hash 作为 Redis key短时间命中后直接返回历史答案减少重复调用。import hashlib import redis r redis.Redis(hostlocalhost, port6379, db0) def get_answer_with_cache(user_input: str, ttl: int 3600): key fllm:cache:{hashlib.md5(user_input.encode()).hexdigest()} cached r.get(key) if cached: return cached.decode(utf-8) answer service.get_answer(user_input) r.setex(key, ttl, answer) return answer注意缓存只适合答案对时效性不敏感的场景。如果用户问“当前时间”或“最新股票价格”绝不能用缓存。网络波动和限流是生产环境的常态。建议对请求做指数退避重试import time from typing import Callable def retry_on_failure(func: Callable, retries: int 3, base_delay: float 1.0): for attempt in range(retries): try: return func() except Exception as e: if attempt retries - 1: raise e delay base_delay * (2 ** attempt) print(f请求失败{delay} 秒后重试{e}) time.sleep(delay)重试要设置最大次数不能无限重试否则容易造成请求积压。6.4 并发控制大模型 API 一般有并发限制超过限制会返回 429 或者连接超时。在后端服务中需要根据自己的账号等级配置并发信号量import asyncio import semaphore llm_semaphore asyncio.Semaphore(5) async def limited_chat(messages): async with llm_semaphore: loop asyncio.get_event_loop() return await loop.run_in_executor(None, chat, messages)如果使用同步openaiSDK建议部署多个 Worker并结合消息队列削峰。7. 安全与合规注意事项7.1 API Key 保护生产环境不要把 API Key 写到环境变量之外的地方尤其是不要通过后端接口直接返回给前端。正确做法是前端请求自己的后端由后端持有 Key 并调用大模型平台。7.2 输入输出过滤模型生成内容可能存在不确定风险必须在业务层增加人工审核或自动关键词过滤机制。对金融、医疗等敏感领域模型结果只能作为辅助不能直接作为最终决策依据。7.3 数据隐私调用外部大模型 API 时发送的数据会被传输到第三方服务。如果业务数据涉及用户隐私或商业机密需要先做脱敏处理或者使用私有化部署方案。同时在用户协议中应明确告知数据会被用于模型处理。8. 常见问题与排查思路问题现象常见原因解决思路AuthenticationErrorAPI Key 错误或过期检查环境变量控制台重新生成 KeyModelNotFoundError模型名称错误或当前账号无权限对照官方文档修改MODEL_NAME确认是否选择对应版本模型请求超时网络不通或响应时间过长增加timeout检查网络代理缩短输入文本返回内容被截断max_tokens设置过小调大max_tokens或把回复拆成多段多轮对话答非所问历史消息顺序错误或丢失 assistant 回复调试打印messages验证历史结构429 限流并发超过账号阈值降低并发数增加重试退避升级账号配额流式接口前端无法解析返回格式不是标准 SSE检查响应头text/event-stream确保每段格式为data: ...\n\n成本飙升忘记控制上下文长度或缓存失效通过 usage 日志定位高消耗场景设置 token 上限排查时建议先做一个最小化测试只传一条user消息不带历史记录使用同步调用确认基础链路通不通。如果最小化测试通过再逐步增加历史消息和流式逻辑。9. 工程落地的进一步建议如果你准备把大模型能力真正放到生产环境有几个点值得提前规划第一把大模型调用封装成独立微服务与其他业务系统解耦。这样模型升级、配置调整不会频繁触发整个系统的发布。第二建立完整的日志链路。记录每次请求的request_id、token 用量、耗时、模型名称、错误信息。没有日志线上问题排查会非常痛苦。第三设计“熔断降级”机制。当大模型接口连续失败时可以返回预设的兜底文案或切换到备用模型避免核心业务完全不可用。第四提前考虑多模型支持。不同场景可能适合不同模型例如简单分类任务用轻量模型复杂推理用旗舰模型。通过工厂模式动态切换模型比写死某一个模型更适合长期演进。我在实际项目中还发现很多人忽略了“评估环节”。模型换版本后表现可能提升也可能下降。建议搭建一个离线评测集包含几十条典型问题每次切换模型或修改提示词时批量跑一遍对比输出质量。这是避免线上事故最有效的手段之一。如果你刚开始接触这一块不要急着上复杂架构先把本文的同步调用跑通再实现流式然后是 FastAPI 封装和 Redis 缓存。每一步都验证通过后再考虑多机部署和模型评测体系。技术底座稳了业务增长才稳。对于智谱这类快速增长的平台尽早培养工程化习惯会让你的项目在后续迭代中少踩很多坑。