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, []

6. 总结：写给正在调试的你

你不需要记住所有参数，也不必理解Transformer架构。你只需要清楚三件事：

• 接口地址永远是 http://127.0.0.1:8000/chat，不是Gradio端口，不是模型仓库地址；
• 请求必须用 json= 发送标准JSON，字段名一个字母都不能错，类型必须匹配；
• 所有异常都要捕获并给出明确提示，而不是让程序静默失败。

这篇指南里的每一段代码，都来自真实部署场景——不是实验室玩具，而是每天处理上千次请求的脚本。它不追求炫技，只确保你花10分钟复制粘贴后，就能把ChatGLM-6B真正用起来。

下一步，你可以：

• 把chatglm_api_call封装成公司内部SDK；
• 接入钉钉/飞书机器人，实现自动答疑；
• 和数据库联动，让AI根据业务表结构生成SQL；
• 甚至做成CLI工具：chatglm "帮我写个Python爬虫"。

技术的价值，从来不在模型多大，而在它能不能安静地、可靠地，帮你把一件事做完。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。