ChatGLM-6B代码实例：Python调用API接口避坑指南-开发者社区

ChatGLM-6B代码实例：Python调用API接口避坑指南

1. 为什么需要自己写代码调用，而不是只用Web界面？

你可能已经试过在浏览器里打开http://127.0.0.1:7860，和ChatGLM-6B聊得挺开心——中英文切换自然、回答有逻辑、还能记住上一句。但很快就会遇到几个现实问题：

想把模型能力嵌入到自己的脚本里，比如自动写日报、批量处理客户咨询、生成测试用例；
Web界面不支持程序化调用，没法传参、没法控制超时、没法集成进CI/CD流程；
需要并发请求多个对话，或者做A/B测试不同参数效果；
想把回答结果直接存进数据库、发到企业微信、转成语音播报……

这时候，你就得绕过Gradio界面，直连后端服务。但别急着复制网上那些“几行代码搞定”的教程——它们大多默认你本地跑模型、用的是Hugging Face原生加载方式，而CSDN镜像里的ChatGLM-6B是封装好的生产级服务，走的是HTTP API，不是Python对象调用。

这篇文章不讲理论，不堆参数，只说你在真实环境里调用时一定会踩的坑、会卡住的点、会报错的场景，以及怎么用最简练的Python代码稳稳绕过去。

2. 先搞清这个镜像到底提供什么接口

很多新手一上来就翻Transformers文档，试图用pipeline()或AutoModelForSeq2SeqLM去加载远程模型——这完全走错了路。CSDN这个镜像没有暴露模型对象，只暴露了标准HTTP接口，就像调用一个轻量级聊天机器人API。

它底层用的是FastAPI（虽然没明说），启动后监听在http://127.0.0.1:8000（注意：不是Gradio的7860端口！）。Gradio只是个前端壳，真正干活的是背后这个API服务。

你可以用curl快速验证：

curl -X POST "http://127.0.0.1:8000/chat" \ -H "Content-Type: application/json" \ -d '{ "query": "你好，今天天气怎么样？", "history": [], "temperature": 0.7, "max_length": 2048 }'

如果返回类似这样的JSON，说明服务已就绪：

{ "response": "我无法获取实时天气信息，但你可以告诉我你所在的城市，我可以帮你查天气预报方法。", "history": [["你好，今天天气怎么样？", "我无法获取实时天气信息，但你可以告诉我你所在的城市，我可以帮你查天气预报方法。"]] }

注意三个关键事实：

接口地址是http://127.0.0.1:8000/chat，不是/api/chat也不是/v1/chat；
必须用POST方法，GET直接返回405错误；
请求体必须是合法JSON，不能是URL编码字符串，否则返回400。

3. Python调用实战：从能跑通到能落地

下面这段代码，是你能在CSDN镜像里真正稳定运行的最小可行示例。它不是教科书式优雅，而是为生产环境打磨过的——带重试、防超时、结构清晰、错误可读。

3.1 基础调用：一行不改就能跑

import requests import time def chatglm_api_call( query: str, history: list = None, temperature: float = 0.7, max_length: int = 2048, timeout: int = 60 ): """ 调用CSDN镜像中ChatGLM-6B的HTTP API 注意：此函数假设服务已在本地8000端口运行（通过SSH隧道映射） """ if history is None: history = [] url = "http://127.0.0.1:8000/chat" payload = { "query": query, "history": history, "temperature": temperature, "max_length": max_length } try: response = requests.post( url, json=payload, # 关键：必须用 json= 而不是 data= timeout=timeout ) response.raise_for_status() # 抛出4xx/5xx异常 result = response.json() return result["response"], result["history"] except requests.exceptions.Timeout: raise TimeoutError("请求超时，请检查服务是否正常运行或增大timeout参数") except requests.exceptions.ConnectionError: raise ConnectionError("无法连接到ChatGLM服务，请确认supervisor已启动且端口映射正确") except requests.exceptions.HTTPError as e: raise RuntimeError(f"HTTP错误：{e}，响应内容：{response.text}") except KeyError as e: raise ValueError(f"API响应格式异常，缺少字段：{e}，原始响应：{response.text}") except Exception as e: raise RuntimeError(f"未知错误：{e}") # 使用示例：单轮对话 answer, _ = chatglm_api_call("Python里怎么把列表去重？") print("回答：", answer) # 输出：回答： 可以用 set() 转换再转回 list，但会丢失顺序；若需保持顺序，推荐用 dict.fromkeys() 或循环判断。 # 使用示例：多轮对话（带上下文） history = [] q1 = "什么是Transformer模型？" a1, history = chatglm_api_call(q1, history) print("Q1:", q1) print("A1:", a1) q2 = "它和RNN有什么区别？" a2, history = chatglm_api_call(q2, history) # history自动传递 print("Q2:", q2) print("A2:", a2)

3.2 进阶技巧：让调用更健壮、更实用

自动重试机制（应对偶发性超时）

from functools import wraps import random def retry_on_failure(max_retries=3, backoff_factor=1.0): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): last_exception = None for attempt in range(max_retries): try: return func(*args, **kwargs) except (TimeoutError, ConnectionError) as e: last_exception = e if attempt < max_retries - 1: sleep_time = backoff_factor * (2 ** attempt) + random.uniform(0, 1) time.sleep(sleep_time) raise last_exception return wrapper return decorator @retry_on_failure(max_retries=2, backoff_factor=0.5) def robust_chatglm_call(*args, **kwargs): return chatglm_api_call(*args, **kwargs)

批量处理：一次发多个问题（节省网络开销）

def batch_chatglm_calls(queries: list, **kwargs) -> list: """ 批量调用，每个问题独立history，避免上下文污染 返回：[{"query": "...", "answer": "...", "history": [...]}, ...] """ results = [] for q in queries: try: answer, history = chatglm_api_call(q, history=[], **kwargs) results.append({ "query": q, "answer": answer, "history": history }) except Exception as e: results.append({ "query": q, "error": str(e), "answer": None, "history": [] }) return results # 使用 questions = [ "Python里如何读取CSV文件？", "怎么用pandas筛选数据？", "matplotlib画折线图的基本步骤是什么？" ] batch_results = batch_chatglm_calls(questions, temperature=0.5) for r in batch_results: print(f"Q: {r['query']}") print(f"A: {r.get('answer', ' 调用失败：' + r['error'])}\n")

4. 你一定会遇到的5个典型坑及解决方案

这些不是“可能遇到”，而是几乎每个第一次调用的人都会卡住10分钟以上的问题。我们按出现频率排序：

4.1 坑一：ConnectionRefusedError —— 服务根本没起来

现象：requests.exceptions.ConnectionError: HTTPConnectionPool(host='127.0.0.1', port=8000): Max retries exceeded...

原因：

supervisorctl start chatglm-service没执行，或执行失败；
服务启动日志里有CUDA内存不足报错（常见于小显存GPU）；
端口被其他进程占用（比如另一个FastAPI服务占了8000）。

排查命令：

# 看服务状态 supervisorctl status chatglm-service # 查看最近100行日志（重点找ERROR或CUDA out of memory） tail -100 /var/log/chatglm-service.log # 检查8000端口是否被监听 netstat -tuln | grep :8000 # 或 lsof -i :8000

解决：

如果状态是FATAL，重启服务：supervisorctl restart chatglm-service；
如果日志显示OOM，降低max_length或关闭--quantize（如果启用了量化）；
如果端口冲突，修改/etc/supervisor/conf.d/chatglm.conf中的port=8000为其他值（如8001），再supervisorctl reread && supervisorctl update。

4.2 坑二：400 Bad Request —— JSON格式不对

现象：requests.exceptions.HTTPError: 400 Client Error

原因：

用data=json.dumps(...)而不是json=...，导致Content-Type变成application/x-www-form-urlencoded；
history字段传了None或字符串，不是空列表[]；
temperature传了字符串"0.7"而不是浮点数0.7。

验证方法：

# 错误写法（触发400） requests.post(url, data=json.dumps(payload), headers={"Content-Type": "application/json"}) # 正确写法（requests自动设header） requests.post(url, json=payload) # 推荐

4.3 坑三：422 Unprocessable Entity —— 字段名拼错或缺失

现象：返回{"detail":[{"loc":["body","xxx"],"msg":"field required","type":"value_error.missing"}]}

原因：API要求必填字段是query、history、temperature、max_length，少一个就报错。注意不是input、prompt、text。

安全写法：

payload = { "query": query, "history": history or [], # 确保是list "temperature": float(temperature), # 强制转float "max_length": int(max_length) # 强制转int }

4.4 坑四：响应慢得像卡死 —— 实际是流式响应没关

现象：requests.post(...)卡住30秒以上才返回，但Web界面响应很快。

原因：CSDN镜像默认启用流式响应（streaming），但requests默认等待完整响应。而模型生成长文本时，流式会分块发送，requests却在等最后一块。

解决：显式关闭流式（API支持）：

# 在payload里加一个开关（实测有效） payload["stream"] = False # 加这一行

4.5 坑五：中文乱码或emoji显示异常

现象：返回的response字段里中文变成\u4f60\u597d，或表情符号显示为方块。

原因：requests默认用ISO-8859-1解码，而API返回UTF-8。

解决：强制指定编码：

response = requests.post(url, json=payload) response.encoding = "utf-8" # 加这一行 result = response.json()

5. 生产环境建议：不只是能跑，还要跑得稳

5.1 日志记录：别让问题消失在黑盒里

import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler("/var/log/chatglm_client.log"), logging.StreamHandler() ] ) def logged_chatglm_call(*args, **kwargs): logging.info(f"发起ChatGLM请求：query='{args[0][:50]}...' temp={kwargs.get('temperature', 0.7)}") try: result = chatglm_api_call(*args, **kwargs) logging.info("ChatGLM请求成功") return result except Exception as e: logging.error(f"ChatGLM请求失败：{e}") raise

5.2 资源监控：防止雪崩

ChatGLM-6B单次推理约占用3~4GB显存。如果你并发调用太多，服务会OOM崩溃。

简单限流方案（无需额外库）：

import threading import time class RateLimiter: def __init__(self, max_calls=2, period=1.0): self.max_calls = max_calls self.period = period self.calls = [] self.lock = threading.Lock() def acquire(self): with self.lock: now = time.time() # 清理过期记录 self.calls = [t for t in self.calls if now - t < self.period] if len(self.calls) >= self.max_calls: sleep_time = self.period - (now - self.calls[0]) if sleep_time > 0: time.sleep(sleep_time) # 重新清理 now = time.time() self.calls = [t for t in self.calls if now - t < self.period] self.calls.append(now) limiter = RateLimiter(max_calls=3, period=2.0) # 最多3次/2秒 def safe_chatglm_call(*args, **kwargs): limiter.acquire() return chatglm_api_call(*args, **kwargs)

5.3 错误降级：当ChatGLM挂了，你的程序不能跟着挂

def fallback_chatglm_call(query: str, fallback_response: str = "当前AI服务繁忙，请稍后再试。"): try: return chatglm_api_call(query) except Exception as e: logging.warning(f"ChatGLM调用失败，启用降级：{e}") return fallback_response, []