理解 LLM 与 API 请求

import json

def generate(messages, max_output_chars=80):
    """离线教学替身，不调用模型，也不计算真实 Token。"""
    if not isinstance(messages, list) or not messages:
        raise ValueError("messages 必须是非空列表")

    for message in messages:
        if not isinstance(message, dict):
            raise ValueError("每条消息必须是字典")
        if not isinstance(message.get("content"), str):
            raise ValueError("content 必须是字符串")

    question = messages[-1]["content"]
    answer = f"模拟收到：{question}"
    return {
        "text": answer[:max_output_chars],
        "finish_reason": "limit" if len(answer) > max_output_chars else "stop",
        "mode": "offline_fixture",
    }

request = [{"role": "user", "content": "用一句话介绍 Agent"}]
wire_text = json.dumps(request, ensure_ascii=False)
response = generate(json.loads(wire_text))
print(json.dumps(response, ensure_ascii=False, indent=2))

从环境变量读取凭据
按所选服务当前文档构造请求
设置连接与读取超时
发送请求
检查 HTTP 状态与服务错误
提取内容、结束原因和真实用量
转换为应用统一的返回结构

import json
import os
import socket
import urllib.error
import urllib.parse
import urllib.request

def chat(messages, tools=None):
    config = {name: os.environ.get(name, "").strip() for name in (
        "LLM_BASE_URL", "LLM_API_KEY", "LLM_MODEL"
    )}
    missing = [name for name, value in config.items() if not value]
    if missing:
        raise RuntimeError("缺少环境变量：" + ", ".join(missing))
    base = config["LLM_BASE_URL"].rstrip("/")
    if urllib.parse.urlparse(base).scheme != "https":
        raise RuntimeError("真实服务必须使用 HTTPS 地址")
    payload = {
        "model": config["LLM_MODEL"], "messages": messages,
        "stream": False, "max_tokens": 1200,
    }
    if tools is not None:
        payload["tools"] = tools
    request = urllib.request.Request(
        base + "/chat/completions",
        data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        headers={
            "Content-Type": "application/json",
            "Authorization": "Bearer " + config["LLM_API_KEY"],
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=60) as response:
            result = json.load(response)
    except urllib.error.HTTPError as error:
        hints = {401: "检查凭据", 402: "检查账户额度", 403: "检查权限", 429: "请求过快或额度受限"}
        raise RuntimeError(f"HTTP {error.code}：{hints.get(error.code, '请检查服务状态和请求配置')}") from None
    except (urllib.error.URLError, TimeoutError, socket.timeout):
        raise RuntimeError("网络失败或请求超时；本例不会自动重试") from None
    except (json.JSONDecodeError, UnicodeDecodeError):
        raise RuntimeError("服务未返回合法 JSON") from None
    if not isinstance(result, dict):
        raise RuntimeError("响应根节点必须是对象")
    choices = result.get("choices")
    if not isinstance(choices, list) or not choices or not isinstance(choices[0], dict):
        raise RuntimeError("响应缺少有效 choices")
    choice = choices[0]
    if not isinstance(choice.get("message"), dict):
        raise RuntimeError("响应缺少 assistant 消息")
    if choice.get("finish_reason") not in ("stop", "tool_calls"):
        raise RuntimeError("生成未正常完成：" + str(choice.get("finish_reason")))
    return result

if __name__ == "__main__":
    try:
        result = chat([{"role": "user", "content": "用一句中文说明什么是 AI Agent。"}])
        print(result["choices"][0]["message"].get("content"))
        print("服务返回的实际用量：", result.get("usage", "服务未提供"))
    except RuntimeError as error:
        raise SystemExit(str(error))