GLM-5.2发布后，大家都在测"能不能跑通"。但真正用到生产环境， two problems always show up: 流式输出不稳定、多轮对话上下文溢出。本文用可运行的代码，把这两个问题讲透。

---

一、为什么流式输出是必选项

非流式调用（stream=False）的问题是：等完整响应回来，用户已经走了。

调用方式	首字延迟	用户体验	适用场景
非流式	2~3秒（整段返回）	差，像"卡住了"	后台批处理
流式	0.3秒（逐字输出）	好，像对话	前端聊天界面

GLM-5.2的流式输出兼容OpenAI格式，但有几个官方文档没写的细节。

---

二、流式输出：最小可用代码

基础版本（能跑，但不完整）

from openai import OpenAI

client = OpenAI(
    api_key="your-glm-api-key",
    base_url="https://open.bigmodel.cn/api/compat/v1"
)

# stream=True，返回的是生成器
stream = client.chat.completions.create(
    model="glm-5.2",
    messages=[{"role": "user", "content": "用Python写一个快速排序"}],
    stream=True  # 关键参数
)

# 逐块读取
for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

这段代码的输出效果：内容会逐字打印到终端，像打字机一样。

但这段代码有三个问题，生产环境必须用升级版。

---

三、流式输出的三个实战问题

问题1：网络中断导致流断裂

GLM-5.2的流式连接在网络抖动时会静默失败——生成器停止迭代，但你不收到任何错误提示，代码"静默结束"，用户看到"回答中断"。

解决方案：用 try-except 包裹 + 超时控制

import sys
from openai import OpenAI, APIConnectionError, APITimeoutError

client = OpenAI(
    api_key="your-glm-api-key",
    base_url="https://open.bigmodel.cn/api/compat/v1",
    timeout=30.0  # 关键：设置超时，默认无超时
)

def stream_with_recovery(messages, max_retries=2):
    for attempt in range(max_retries):
        try:
            stream = client.chat.completions.create(
                model="glm-5.2",
                messages=messages,
                stream=True,
                timeout=30  # 单次请求超时30秒
            )
            collected_content = []
            for chunk in stream:
                content = chunk.choices[0].delta.content
                if content:
                    print(content, end="", flush=True)
                    collected_content.append(content)
            return "".join(collected_content)  # 返回完整内容，供调用方保存
        
        except (APIConnectionError, APITimeoutError) as e:
            print(f"\n[第{attempt+1}次重试] 连接中断: {e}")
            if attempt == max_retries - 1:
                return "[回答中断，请重试]"
            continue
        except Exception as e:
            print(f"\n[异常] {e}")
            return "[服务异常]"
    
    return "[重试次数耗尽]"

# 用法
result = stream_with_recovery([
    {"role": "user", "content": "详细介绍Python的GIL机制"}
])

关键改进：

• timeout=30.0：防止连接挂起
• 捕获 APIConnectionError：网络中断时优雅降级
• 返回完整字符串：方便你把结果存数据库

---

问题2：前端怎么接流式数据？

后端用流式，前端必须用 EventSource 或 fetch 的流式读取，否则白搞。

后端（Python/Flask）完整示例：

from flask import Flask, Response, stream_with_context
from openai import OpenAI

app = Flask(__name__)
client = OpenAI(
    api_key="your-glm-api-key",
    base_url="https://open.bigmodel.cn/api/compat/v1"
)

@app.route('/chat_stream')
def chat_stream():
    def generate():
        stream = client.chat.completions.create(
            model="glm-5.2",
            messages=[{"role": "user", "content": "解释量子计算"}],
            stream=True
        )
        for chunk in stream:
            content = chunk.choices[0].delta.content
            if content:
                # SSE格式：data: {content}\n\n
                yield f"data: {content}\n\n"
        yield "data: [DONE]\n\n"
    
    return Response(
        stream_with_context(generate()),
        mimetype='text/event-stream',
        headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}
    )

if __name__ == '__main__':
    app.run(port=5000)

前端（JavaScript）完整示例：

// 前端用 fetch 读取流式响应
async function callStream() {
    const response = await fetch('/chat_stream');
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    
    let buffer = "";
    while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        
        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split('\n');
        buffer = lines.pop();  // 最后一行可能不完整
        
        for (const line of lines) {
            if (line.startsWith('data: ')) {
                const content = line.slice(6);  // 去掉 "data: " 前缀
                if (content === '[DONE]') return;
                // 追加到页面
                document.getElementById('chat-output').innerHTML += content;
            }
        }
    }
}

第三个坑： GLM-5.2的流式chunk里，chunk.choices 有时是空数组（最后一个chunk），直接访问会IndexError。生产代码必须判断：

for chunk in stream:
    if not chunk.choices:  # 最后一个chunk，choices为空
        continue
    content = chunk.choices[0].delta.content
    if content:
        print(content, end="", flush=True)

---

四、上下文管理：多轮对话的核心问题

问题本质

GLM-5.2的上下文窗口是 128K tokens（约10万字中文）。但"有128K"不等于"你能随便用"——

两个问题：

1. 每次调用API，都要把完整对话历史传上去，128K是单次请求的上限
2. 超出128K后，模型会静默截断前面的内容，导致"失忆"

错误做法：无脑传全量历史

# ❌ 错误：对话轮数多了之后，这个列表会超出128K
messages = []
while True:
    user_input = input("你：")
    messages.append({"role": "user", "content": user_input})
    response = client.chat.completions.create(
        model="glm-5.2",
        messages=messages  # 每次都传全部历史
    )
    assistant_reply = response.choices[0].message.content
    messages.append({"role": "assistant", "content": assistant_reply})
    print(f"AI：{assistant_reply}")

跑20轮之后，这个 messages 列表可能已经有5万字，远远超出128K。

---

正确做法：滑动窗口 + 自动截断

策略： 始终保留最近 N 轮对话，超出部分做摘要压缩。

import tiktoken

# 用tiktoken估算token数（GLM-5.2用cl100k_base编码，与GPT-4相同）
encoding = tiktoken.get_encoding("cl100k_base")

def count_tokens(messages):
    """计算messages列表的总token数"""
    total = 0
    for msg in messages:
        # 每条消息：角色(1) + 内容 + 2个分隔符
        total += len(encoding.encode(msg["content"])) + 4
    return total

def trim_messages(messages, max_tokens=100000):
    """截断策略：保留system prompt + 最近N轮"""
    if count_tokens(messages) <= max_tokens:
        return messages
    
    # 分离system prompt（必须保留）
    system_msgs = [m for m in messages if m["role"] == "system"]
    conversation = [m for m in messages if m["role"] != "system"]
    
    # 从尾部向前保留，直到接近上限
    trimmed = []
    total = 0
    for msg in reversed(conversation):
        msg_tokens = len(encoding.encode(msg["content"])) + 4
        if total + msg_tokens > max_tokens - 4000:  # 留4000给新回复
            break
        trimmed.insert(0, msg)
        total += msg_tokens
    
    # 如果截断太狠（少于3轮），用摘要代替
    if len(trimmed) < 3 and len(conversation) > 3:
        summary = summarize_history(conversation[:-len(trimmed)])
        trimmed.insert(0, {"role": "system", "content": f"历史对话摘要：{summary}"})
    
    return system_msgs + trimmed

def summarize_history(old_messages):
    """用GLM-5.2给自己做摘要（节省token的黄金技巧）"""
    history_text = "\n".join([f"{m['role']}: {m['content']}" for m in old_messages])
    response = client.chat.completions.create(
        model="glm-5.2",
        messages=[{"role": "user", "content": f"请用100字概括以下对话的核心信息：\n{history_text}"}]
    )
    return response.choices[0].message.content

# 完整用法
messages = [
    {"role": "system", "content": "你是一个严谨的Python编程助手"}
]

while True:
    user_input = input("你：")
    messages.append({"role": "user", "content": user_input})
    
    # 截断后再发送
    messages = trim_messages(messages, max_tokens=100000)
    
    response = client.chat.completions.create(
        model="glm-5.2",
        messages=messages
    )
    assistant_reply = response.choices[0].message.content
    messages.append({"role": "assistant", "content": assistant_reply})
    print(f"AI：{assistant_reply}")

---

五、上下文管理的三个实战技巧

技巧1：用"隐藏标记"检测是否失忆

# 在第一条消息里埋一个"校验码"
messages = [
    {"role": "user", "content": "【校验码：WX82F】请用Python写一个排序算法"}
]

# 10轮对话后，问模型
messages.append({"role": "user", "content": "你还记得我们第一次对话的校验码吗？"})

# 如果模型回答不出，说明上下文已被截断，需要重新摘要

技巧2：重要信息用system prompt重复强调

# 用户说了关键信息后，动态更新system prompt
if "我叫" in user_input:
    user_name = user_input.split("我叫")[1].split()[0]
    # 把名字写入system prompt
    messages[0]["content"] += f"\n用户的名字是{user_name}，每次回复都要称呼他。"

# system prompt不计入"对话轮数"，不会被截断

技巧3：用token计数决定何时主动摘要

# 每10轮检查一次token数，超过8万时主动触发摘要
if len(messages) > 10 and count_tokens(messages) > 80000:
    summary = summarize_history(messages[1:-4])  # 保留最近3轮
    messages = [messages[0]] + [{"role": "system", "content": f"历史摘要：{summary}"}] + messages[-4:]
    print("[系统] 对话历史已压缩，继续聊天...")

---

六、流式 + 上下文管理的完整整合

把上面两个功能整合，是一个生产可用的聊天程序：

import tiktoken
from openai import OpenAI, APIConnectionError

client = OpenAI(
    api_key="your-glm-api-key",
    base_url="https://open.bigmodel.cn/api/compat/v1",
    timeout=30.0
)
encoding = tiktoken.get_encoding("cl100k_base")

def count_tokens(messages):
    return sum(len(encoding.encode(m["content"])) + 4 for m in messages)

def chat(user_input, messages, stream=True):
    messages.append({"role": "user", "content": user_input})
    
    # 上下文保护：超过10万token就截断
    if count_tokens(messages) > 100000:
        # 简单策略：保留system + 最近8条
        system = [m for m in messages if m["role"] == "system"]
        recent = messages[-8:]
        messages = system + recent
    
    if not stream:
        response = client.chat.completions.create(
            model="glm-5.2",
            messages=messages,
            stream=False
        )
        reply = response.choices[0].message.content
        messages.append({"role": "assistant", "content": reply})
        return reply
    
    # 流式输出（带断线恢复）
    collected = []
    try:
        stream = client.chat.completions.create(
            model="glm-5.2",
            messages=messages,
            stream=True
        )
        for chunk in stream:
            if not chunk.choices:
                continue
            content = chunk.choices[0].delta.content
            if content:
                print(content, end="", flush=True)
                collected.append(content)
    except APIConnectionError:
        print("\n[网络中断，已停止]")
    
    full_reply = "".join(collected)
    messages.append({"role": "assistant", "content": full_reply})
    return full_reply

# 使用示例
messages = [{"role": "system", "content": "你是Python编程助手"}]

print("开始对话（输入exit退出）")
while True:
    user_input = input("\n你：")
    if user_input == "exit":
        break
    print("AI：", end="")
    chat(user_input, messages)
    print()  # 换行

---

七、总结：这两个功能的生产 checklist

流式输出 checklist

• stream=True + 超时设置
• 捕获 APIConnectionError 和 APITimeoutError
• 判断 chunk.choices 是否为空（最后一个chunk）
• 前端用 fetch + ReadableStream 读取
• 设置 X-Accel-Buffering: no 防止nginx缓冲

上下文管理 checklist

• 用 tiktoken 估算token数（不要盲目相信"128K可用"）
• 保留system prompt，截断user/assistant历史
• 超过80K token时主动摘要
• 用"隐藏校验码"检测失忆
• system prompt里放重要用户信息（不会被截断）

---

GLM-5.2的编程能力确实强，但把模型跑起来和跑进生产环境，中间差的就是这些细节。

如果有其他GLM-5.2的实战问题，欢迎评论区交流。下一篇准备写Function Calling实战和Token计费对比，关注不迷路。

---

测试环境：Python 3.11，openai SDK 1.52.3，GLM-5.2 API（智谱官方+聚合渠道混合测试）