企业官网建设流程全解析

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（用于生产限流与缓存）

建议新建一个虚拟环境，避免污染系统 Python：

python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install requests openai

2.2 获取 API Key

大模型平台目前普遍采用 API Key 鉴权。你需要登录智谱开放平台，在控制台完成实名认证后，创建 API Key。Key 是敏感信息，不要把 Key 写进代码或提交到 Git 仓库。推荐使用环境变量：

export ZHIPU_API_KEY="你的_API_Key"

2.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 有三种：

构建对话时，要按时间先后排列消息列表。不能把 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_key=config.API_KEY, base_url=config.BASE_URL, timeout=config.TIMEOUT, ) def chat(messages, temperature=0.7, max_tokens=1024): """ 同步调用对话模型 :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( model=config.MODEL_NAME, messages=messages, temperature=temperature, max_tokens=max_tokens, ) return response.choices[0].message.content

这里有一个容易被忽略的坑：max_tokens指的是生成回复的最大 token 数，不包含输入 token 数。设置太小会导致回答被截断，设置太大会增加单次请求的成本。

4.4 添加流式输出

# llm_client.py 中新增函数 def chat_stream(messages, temperature=0.7, max_tokens=1024): """ 流式调用对话模型，返回一个迭代器 每次迭代返回一个字符串片段 """ if not config.API_KEY: raise ValueError("缺少 ZHIPU_API_KEY 环境变量") response = client.chat.completions.create( model=config.MODEL_NAME, messages=messages, temperature=temperature, max_tokens=max_tokens, stream=True, ) 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="", flush=True) 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_model=ChatResponse) async def chat_with_llm(req: ChatRequest): answer = service.get_answer(req.user_input, req.history) return ChatResponse(answer=answer) @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 f"data: {text}\n\n" return StreamingResponse(gen(), media_type="text/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( model=config.MODEL_NAME, messages=messages, **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(host="localhost", port=6379, db=0) def get_answer_with_cache(user_input: str, ttl: int = 3600): key = f"llm: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. 常见问题与排查思路

排查时建议先做一个最小化测试：只传一条user消息，不带历史记录，使用同步调用，确认基础链路通不通。如果最小化测试通过，再逐步增加历史消息和流式逻辑。

9. 工程落地的进一步建议

如果你准备把大模型能力真正放到生产环境，有几个点值得提前规划：

第一，把大模型调用封装成独立微服务，与其他业务系统解耦。这样模型升级、配置调整不会频繁触发整个系统的发布。

第二，建立完整的日志链路。记录每次请求的request_id、token 用量、耗时、模型名称、错误信息。没有日志，线上问题排查会非常痛苦。

第三，设计“熔断降级”机制。当大模型接口连续失败时，可以返回预设的兜底文案，或切换到备用模型，避免核心业务完全不可用。

第四，提前考虑多模型支持。不同场景可能适合不同模型，例如简单分类任务用轻量模型，复杂推理用旗舰模型。通过工厂模式动态切换模型，比写死某一个模型更适合长期演进。

我在实际项目中还发现，很多人忽略了“评估环节”。模型换版本后，表现可能提升也可能下降。建议搭建一个离线评测集，包含几十条典型问题，每次切换模型或修改提示词时，批量跑一遍，对比输出质量。这是避免线上事故最有效的手段之一。

如果你刚开始接触这一块，不要急着上复杂架构，先把本文的同步调用跑通，再实现流式，然后是 FastAPI 封装和 Redis 缓存。每一步都验证通过后，再考虑多机部署和模型评测体系。技术底座稳了，业务增长才稳。对于智谱这类快速增长的平台，尽早培养工程化习惯，会让你的项目在后续迭代中少踩很多坑。