在第1讲中，我们成功让 Agent 调用了 get_current_time() 工具。但那个 Agent 还非常简陋：它只能执行预设好的动作，且没有真正的“大脑”——它只是按我们给的规则机械地运行。

真正的 Agent，核心在于大模型（LLM）的推理能力。它需要根据用户的输入，动态决定做什么、怎么做。所以，这一讲，我们要解决两个核心问题：

1. 如何稳定、高效地调用大模型 API？

2. 如何让模型输出像打字机一样“流式”显示，提升用户体验？

---

一、为什么选择 OpenAI 兼容协议？

目前市面上主流的大模型（DeepSeek、通义千问、GLM、Kimi、甚至本地部署的 Ollama）都提供了 OpenAI 兼容的 API 接口。这意味着：

• 一套代码，可以切换不同模型（只需改 BASE_URL 和 model 参数）

• 社区生态丰富（LangChain、LlamaIndex、各种 SDK 都支持）

• 文档齐全，上手快

我们以 DeepSeek​ 为例（你也可以换成通义、GLM 等），它的接口地址是：

https://api.deepseek.com/v1/chat/completions

请求格式完全遵循 OpenAI 标准：

{
  "model": "deepseek-chat",
  "messages": [
    {"role": "system", "content": "你是一个助手"},
    {"role": "user", "content": "你好"}
  ],
  "temperature": 0.7,
  "stream": false
}

---

二、基础 API 调用代码（同步版）

我们先写一个最基础的调用函数：

import requests
import json

API_KEY = "你的API_KEY"
BASE_URL = "https://api.deepseek.com/v1/chat/completions"

def call_llm_sync(messages, model="deepseek-chat", temperature=0.7):
    """同步调用大模型 API"""
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": model,
        "messages": messages,
        "temperature": temperature,
        "stream": False  # 非流式
    }

    try:
        response = requests.post(BASE_URL, headers=headers, json=payload, timeout=30)
        response.raise_for_status()
        data = response.json()
        return data["choices"][0]["message"]["content"]
    except Exception as e:
        print(f"[API 调用失败] {e}")
        return None

使用示例：

messages = [
    {"role": "system", "content": "你是一个专业的 Python 开发者，请用中文回答。"},
    {"role": "user", "content": "请用 Python 写一个快速排序算法。"}
]

result = call_llm_sync(messages)
print(result)

✅ 输出会是完整的代码块，一次性打印出来。

---

三、进阶：流式输出（Streaming）——像聊天一样“打字”

在实际应用中（比如网页聊天、命令行助手），用户希望看到模型“边想边说”，而不是等全部生成完才显示。这就需要 流式输出。

什么是流式输出？

大模型在生成回复时，不是一次性返回全部内容，而是逐段返回（token by token）。我们通过 stream=True 参数开启这个功能，然后逐行读取响应。

代码实现：

def call_llm_stream(messages, model="deepseek-chat", temperature=0.7):
    """流式调用大模型 API"""
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": model,
        "messages": messages,
        "temperature": temperature,
        "stream": True  # 开启流式
    }

    try:
        response = requests.post(BASE_URL, headers=headers, json=payload, stream=True, timeout=60)
        response.raise_for_status()

        for line in response.iter_lines():
            if line:
                decoded_line = line.decode('utf-8').strip()
                if decoded_line.startswith("data: "):
                    json_str = decoded_line[len("data: "):]
                    if json_str == "[DONE]":
                        break
                    try:
                        chunk = json.loads(json_str)
                        delta = chunk["choices"][0]["delta"]
                        if "content" in delta:
                            yield delta["content"]
                    except Exception as e:
                        print(f"[解析流失败] {e}")
    except Exception as e:
        print(f"[API 流式调用失败] {e}")

使用示例（命令行逐字打印）：

messages = [
    {"role": "system", "content": "你是一个幽默的AI助手，请用中文回复。"},
    {"role": "user", "content": "讲个笑话吧"}
]

print("AI: ", end="", flush=True)
for chunk in call_llm_stream(messages):
    print(chunk, end="", flush=True)
print()  # 换行

✅ 你会看到类似这样的效果：

AI: 有一天，小明问老师：“老师，为什么鱼不会说话？” 老师答：“因为鱼在水里，说话会被淹死。”

---

四、关键参数详解

参数	作用	建议值
model	模型名称	deepseek-chat, qwen-turbo, glm-4 等
temperature	控制随机性（创意度）	0.0（严谨）→ 0.7（平衡）→ 1.0（发散）
stream	是否流式输出	True（交互体验好） / False（简单场景）
timeout	请求超时时间	30~60 秒（避免卡死）
max_tokens	最大生成长度	根据需求设置（如 1024、2048）

---

五、成本意识 & Token 统计

大模型是按 Token 数量收费的（输入 + 输出）。虽然 DeepSeek 目前免费，但生产环境一定要监控：

def count_tokens(text):
    """粗略估算 token 数（英文按空格分词，中文按字符）"""
    # 简单估算：中文字符数 ≈ token 数，英文单词数 ≈ token 数
    return len(text)  # 简化版，实际可用 tiktoken 库

# 示例
user_input = "请用 Python 写一个快速排序算法。"
print(f"输入 token 数: {count_tokens(user_input)}")

更精确的做法是使用官方 tokenizer（如 tiktoken 对应 OpenAI，sentencepiece 对应 Qwen）。

---

六、常见错误 & 排坑指南

1. 流式输出卡住 / 没反应​

   → 检查 stream=True 是否设置，网络是否稳定，模型是否支持流式（部分模型不支持）。

2. 返回 None 或空内容​

   → 检查 API Key 是否有效，模型名称是否正确，请求体是否完整。

3. 中文乱码​

   → 确保 response.iter_lines() 后 .decode('utf-8')，不要用默认编码。

4. 超时异常​

   → 增加 timeout=60，或在代码中加入重试机制（见下文）。

---

七、增强健壮性：超时重试 + 错误兜底

import time

def call_llm_with_retry(messages, max_retries=3, delay=2):
    """带重试机制的调用"""
    for i in range(max_retries):
        try:
            result = call_llm_sync(messages)
            if result:
                return result
        except Exception as e:
            print(f"[重试 {i+1}/{max_retries}] 失败: {e}")
            time.sleep(delay)
    return "抱歉，服务暂时不可用，请稍后再试。"

---

八、课后作业

1. 修改第1讲的 Agent 代码，把 call_llm 替换成本讲的 call_llm_stream，让 Agent 的输出变成“打字机”效果。

2. 尝试切换模型：把 model="deepseek-chat" 改成 "qwen-turbo"（需换 API URL），看看输出风格有何不同。

3. 实现一个简单的“Token 计数器”：在每次调用前，打印输入 + 输出的总 token 数（可用 len(text) 粗略估算）。

---

九、总结

这一讲我们掌握了：

• 如何使用 OpenAI 兼容协议调用大模型 API

• 如何实现流式输出，提升交互体验

• 关键参数（temperature, stream, timeout）的作用

• 成本意识与 Token 统计的重要性

• 健壮性设计：超时重试、错误兜底

有了这些基础，你的 Agent 才真正“活”起来——它能思考、能输出、能应对网络波动。

下一讲，我们将进入 Agent 的核心能力：工具调用（Function Calling）——让大模型不仅能“说”，还能“做”。

---

🧰 开发之余，处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求，我常用一个纯前端本地工具箱：zz365.top（子页 PDF 大师：PDF 大师 - zz365工具箱）。所有计算在浏览器完成，文件不上传服务器，关页即清。免费、无登录、无广告，适合开发者当常驻标签页。