GLM API Python：聊天、流式输出和工具示例

安装和首次请求

创建虚拟环境并安装 SDK。本文假设使用版本 1 或更高版本的 openai 包。

Copypython -m venv .venv
source .venv/bin/activate
pip install --upgrade openai

从 API 密钥页面 获取密钥；新用户可一键领取 $0.50 试用额度，有效期 7 天且无需信用卡。将密钥保存在环境变量中，然后进行首次调用：

Copyimport os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.glmapi.cc/v1",
    api_key=os.environ["GLM_API_KEY"],
)

resp = client.chat.completions.create(
    model="uncensored",
    messages=[
        {"role": "system", "content": "You are a concise writing assistant."},
        {"role": "user", "content": "Suggest five titles for a space-western novel."},
    ],
    temperature=0.8,
)

print(resp.choices[0].message.content)
print(resp.usage)

这就是整个集成过程。该服务是一个独立的、与 OpenAI 兼容的接口，仅针对一个开放权重模型，因此它不是 GLM 背后的厂商或其他任何人的官方 API。仅支持 POST /v1/chat/completions 和 GET /v1/models。快速入门 列出了支持的请求字段。

两个小习惯从一开始就能帮到你。首先，在冒烟测试中调用一次 client.models.list()：它会触发 GET /v1/models，从而在花费任何 token 进行补全之前，证明密钥、base URL 和网络路径均有效。其次，在开发期间打印 resp.usage。提示词和补全的 token 计数能让你尽早发现系统提示词是否超出预期，并且按每 1M 输入 token $0.15 和每 1M 输出 token $0.50 的价格，你可以心算出一次调用的成本。

多轮聊天和采样设置

该 API 是无状态的，因此你需要自行维护对话列表并在每次请求时发送。一个简单的类就足够了：

Copyclass Chat:
    def __init__(self, client, system, max_turns=12):
        self.client = client
        self.system = system
        self.max_turns = max_turns
        self.history = []

    def say(self, text):
        self.history.append({"role": "user", "content": text})
        window = self.history[-self.max_turns * 2:]
        resp = self.client.chat.completions.create(
            model="uncensored",
            messages=[{"role": "system", "content": self.system}] + window,
            temperature=0.9,
            top_p=0.95,
            frequency_penalty=0.3,
        )
        reply = resp.choices[0].message.content
        self.history.append({"role": "assistant", "content": reply})
        return reply

切片 self.history[-self.max_turns * 2:] 保留最后十二轮用户和助手的对话。实际应用中，还可以固定几条早期消息（如角色设定或关键事实），并对中间部分进行摘要。将对话截断为最后 12 轮，可以在对话增长时保持输入成本不变。上下文窗口为 100,000 个 token（包含提示词和补全），因此这更多是一个成本选择而非必要措施。参数 temperature、top_p、stop、seed 以及惩罚项会直接透传。在测试提示词时，设置 seed 以获得可重复的结果；除非要限制长度，否则不要设置 max_tokens。

流式响应

流式输出使用服务器发送事件（SSE），并在生成 token 时即时提供，这对于任何交互式工具都很重要。传入 stream=True 并遍历：

Copystream = client.chat.completions.create(
    model="uncensored",
    messages=[{"role": "user", "content": "Describe a harbour at dawn in 120 words."}],
    stream=True,
)

parts = []
for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta.content
    if delta:
        parts.append(delta)
        print(delta, end="", flush=True)

text = "".join(parts)

对 chunk.choices 的检查很重要，因为最后一个数据块包含使用信息，且可能包含空的 choices 列表。一个操作要点：流式请求会占用并发槽位，直到流结束。将循环包裹在 try 块中，并加上 finally 子句，如果代码中途抛出异常（例如客户端断开连接），则关闭流。务必结束或关闭流，否则放弃流的循环会在没有执行有用工作时达到并发限制。

JSON 模式

当输出用于其他代码时，要求返回 JSON 对象。在提示词中也提及 JSON，以便模型知道预期的格式：

Copyimport json

resp = client.chat.completions.create(
    model="uncensored",
    response_format={"type": "json_object"},
    messages=[
        {"role": "system", "content": "Reply with a JSON object only."},
        {"role": "user", "content": "Give a character with keys name, age, motto."},
    ],
)

data = json.loads(resp.choices[0].message.content)
print(data["name"])

在生产代码中，仍应将 json.loads 包裹在 try 块中。JSON 模式使输出有效的可能性很高，但被停止序列或你自己的限制截断的请求仍可能留下截断的对象。使用相同的输入重试一次是一个低成本修复方案，由于失败的请求不收费，重试的成本仅是成功的那次调用。

函数调用

函数调用遵循 OpenAI 的格式：你在 tools 中描述函数，模型返回 tool_calls，你执行这些函数并使用 role tool 将结果发回。完整的往返过程：

Copyimport json

tools = [{
    "type": "function",
    "function": {
        "name": "roll_dice",
        "description": "Roll a die with the given number of sides.",
        "parameters": {
            "type": "object",
            "properties": {"sides": {"type": "integer"}},
            "required": ["sides"],
        },
    },
}]

messages = [{"role": "user", "content": "Roll a twenty-sided die for me."}]
first = client.chat.completions.create(
    model="uncensored", messages=messages, tools=tools, tool_choice="auto"
)
msg = first.choices[0].message

if msg.tool_calls:
    messages.append(msg)
    for call in msg.tool_calls:
        args = json.loads(call.function.arguments)
        result = {"value": 17, "sides": args["sides"]}  # replace with real logic
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": json.dumps(result),
        })
    final = client.chat.completions.create(
        model="uncensored", messages=messages, tools=tools
    )
    print(final.choices[0].message.content)
else:
    print(msg.content)

循环比骰子的细节更重要。模型请求调用，你的代码执行它，然后将助手消息和工具结果都追加后再请求。如果跳过助手消息，结果中的 tool_call_id 将没有匹配项，这通常是导致第二次回复混乱的原因。当 tool_choice 设置为 auto 时，模型决定是否调用工具，因此始终需要处理它直接回答的分支，如上面的代码所示。流式响应也可以携带 tool_calls，以片段形式交付，你需要按索引拼接后再解析参数。

在执行操作前验证参数。模型可能会生成格式错误的调用，你的代码应将参数视为不可信输入，就像对待来自用户的任何输入一样。

限制内的异步批次

同时处理多个提示词时，使用 AsyncOpenAI 并设置与并发限制匹配的信号量。完整密钥允许 8 个并发请求，试用密钥允许 2 个：

Copyimport asyncio, os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.glmapi.cc/v1",
    api_key=os.environ["GLM_API_KEY"],
)
gate = asyncio.Semaphore(8)  # use 2 on a trial key

async def one(prompt):
    async with gate:
        r = await client.chat.completions.create(
            model="uncensored",
            messages=[{"role": "user", "content": prompt}],
        )
        return r.choices[0].message.content

async def main():
    prompts = [f"Write a one-line tagline for product {i}." for i in range(40)]
    results = await asyncio.gather(*(one(p) for p in prompts))
    print(len(results))

asyncio.run(main())

信号量处理并发，但不处理每分钟 300 次请求的上限。对于长批次，请添加少量延迟或令牌桶限制器。输入 token 每 100 万 $0.15，输出 token 每 100 万 $0.50，40 条标语请求，输入约 60 个 token，输出 25 个 token，总成本为 $0.00036 + $0.0005 = $0.00086，因此限制而非价格决定了批次设计。请查看定价页面进行充值。

速率限制、重试和标头

超过限制时，API 返回 HTTP 429。错误代码为 rate_limited（每分钟请求过多）和 concurrency（同时请求过多）。临时过载也返回 429，并带有 Retry-After: 5 标头。SDK 会自动重试某些失败，你可以在客户端设置 max_retries，但显式处理更清晰：

Copyimport time
from openai import RateLimitError

def ask(prompt, tries=4):
    for attempt in range(tries):
        try:
            raw = client.chat.completions.with_raw_response.create(
                model="uncensored",
                messages=[{"role": "user", "content": prompt}],
            )
            print("balance:", raw.headers.get("X-Balance-USD"))
            return raw.parse().choices[0].message.content
        except RateLimitError as e:
            wait = float(e.response.headers.get("Retry-After", 2 ** attempt))
            time.sleep(wait)
    raise RuntimeError("still rate limited")

此处 with_raw_response 可访问 X-Balance-USD、X-Request-Id 和速率限制标头。将请求 ID 与任何需要报告的错误一起记录。切勿将密钥放在你提交的笔记本或前端代码中。

不消耗额度进行测试及常见错误

Python 集成的大部分是基础工作，而基础工作不需要实时模型。将网络调用封装在一个函数中，你的测试就可以替换它。仅使用标准库：

Copyfrom unittest.mock import MagicMock

def reply_text(client, prompt):
    r = client.chat.completions.create(
        model="uncensored",
        messages=[{"role": "user", "content": prompt}],
    )
    return r.choices[0].message.content.strip()

def test_reply_text_strips_whitespace():
    fake = MagicMock()
    fake.chat.completions.create.return_value.choices = [
        MagicMock(message=MagicMock(content="  hello  "))
    ]
    assert reply_text(fake, "hi") == "hello"

仅在边缘运行少量真实调用：每次部署进行一次冒烟测试，并偶尔运行一次保存的提示词集。一次约 100 token 输入和 20 token 输出的冒烟测试成本为 $0.000015 + $0.00001 = $0.000025，这微不足道，且失败的请求完全不收费。$0.50 的试用额度远超测试套件的需求。

对于提示词工作，固定 seed 和 temperature，以便在同等条件下比较两个版本的系统提示词。如果在运行之间同时更改提示词和采样设置，结果将无法解读。将每次运行的输出与来自 X-Request-Id 标头的请求 ID 一起存储，以便稍后追踪令人惊讶的回答。

常见的 Python 错误

这些是 Python 用户提出支持问题的主要原因：

• 仅传递密钥作为 OPENAI_API_KEY。 如果你在代码中设置了 base_url 但忘记了密钥变量，SDK 会发送空或错误的密钥。请显式传递 api_key。
• 忘记 /v1。 base URL 以 /v1 结尾。SDK 会自动添加 /chat/completions。
• 使用来自其他服务的模型名称。 此处唯一有效的模型是 uncensored。
• 无限制的并发。 对数千个协程使用 asyncio.gather 而没有信号量，会产生大量的 concurrency 错误。
• 将流式回复视为单个字符串。 拼接数据片段，并跳过包含空 choices 列表的数据块。
• 忽略上下文限制。历史记录加上补全文本必须控制在 100,000 个 token 以内。在达到限制前修剪旧对话。

如需充值和当前价目表，请参阅 定价页面。该服务仅支持文本，因此无需连接图像、音频或嵌入调用，且提示词不用于训练。

更多指南

• GLM API 基础 URL：连接任何 OpenAI 客户端
• GLM API 定价：每 1M token 输入 $0.15，输出 $0.50
• Gemma API：常见误区与事实