OpenAI API 实战指南

成长的小鸟

654人浏览 · 2026-05-11 18:17:40

成长的小鸟 · 2026-05-11 18:17:40 发布

OpenAI API 实战指南

本文是我学习 OpenAI API 过程中的总结与梳理。在写完《Prompt 工程指南》后，我发现一个很实际的问题：Prompt 设计得再好，不知道怎么通过代码发给 LLM，不知道怎么处理返回结果，那终究是纸上谈兵。于是我开始啃 OpenAI 的官方文档，边学边写代码，把 Chat Completions、流式输出（SSE）、Function Calling、Token 计算、错误处理、多轮对话管理这些核心能力一个个摸了一遍。这篇文章就是我的学习笔记，每种技术都附上了我实际跑通的代码，还顺手搭了一个功能完整的 AI 对话应用来验证所学。

声明：本文为作者在学习过程中的总结与梳理，仅供学习参考。由于作者水平有限，文中可能存在表述不准确或遗漏之处，欢迎读者提出指正与交流。配套示例项目 github: ai-chat-app/ gitee: ai-chat-app 中找到并直接运行。

引言：从 Prompt 到 API —— 理论到实践的桥梁
环境准备与认证配置
Chat Completions API 深度解析
- 3.1 消息角色体系
- 3.2 核心参数详解
- 3.3 响应结构解析
流式输出：原理与实现
- 4.1 SSE 协议原理
- 4.2 Python 流式客户端实现
- 4.3 前端流式渲染
Function Calling：让 LLM 使用工具
- 5.1 Function Calling 原理
- 5.2 工具定义与注册
- 5.3 完整调用流程
- 5.4 与 Prompt 工程的结合
Token 计算与成本控制
- 6.1 Token 计算原理
- 6.2 成本估算与优化
错误处理与重试策略
- 7.1 错误类型分类
- 7.2 指数退避重试
- 7.3 限流处理
多轮对话管理
- 8.1 对话上下文窗口管理
- 8.2 对话摘要与压缩
结构化输出
生产环境最佳实践
配套项目：AI 对话应用
总结与建议

1. 引言：从 Prompt 到 API —— 理论到实践的桥梁

1.1 为什么要写这篇文章

在写完《Prompt 工程指南》之后，我掌握了五要素结构、五大推理技术、分层 Prompt 架构这些设计层面的知识。但很快我就发现一个问题：我知道怎么写一个好的 Prompt，但我不知道怎么通过代码把它发给 LLM，不知道怎么处理返回结果，遇到报错也不知道该怎么办。

于是我开始动手写代码，从最简单的 Chat Completions 调用开始，一步步摸索流式输出、Function Calling、Token 计算、错误处理……这篇文章记录的就是我这一路踩坑和收获的过程。

下面这张图是我理解的两层关系——Prompt 工程是"设计层"，OpenAI API 是"实现层"：

Prompt 工程（设计层）          OpenAI API（实现层）
      │                              │
      ├─ 五要素结构          →       ├─ messages 参数构建
      ├─ Few-shot / CoT     →       ├─ 多轮对话中的 Prompt 注入
      ├─ 结构化输出          →       ├─ response_format / Function Calling
      ├─ 分层 Prompt 架构    →       ├─ System / User / Assistant 消息分层
      ├─ A/B 测试           →       ├─ 多模型并行调用对比
      └─ 安全防护            →       └─ 输入过滤 + 输出审核

1.2 我学到了什么

经过这段时间的学习和实践，我掌握了以下能力：

独立搭建一个支持流式输出的 AI 对话应用
实现 Function Calling，让 LLM 调用外部 API 和工具
精确计算 Token 消耗，控制 API 成本
处理各种 API 错误（限流、超时、内容过滤等）
管理多轮对话的上下文窗口
将 Prompt 工程的最佳实践落地到生产代码中

1.3 我的学习体会

回头看这段学习经历，有几个感受想分享：

流式输出是体验的分水岭，加了之后整个应用的感觉完全不一样了，从"等结果"变成了"看结果"
Function Calling 是最让我兴奋的部分，它让 LLM 从"聊天机器人"变成了真正能"干活"的助手
错误处理一开始觉得繁琐，但实际跑起来才发现，网络波动、限流、超时这些情况太常见了，不处理根本没法用
如果你是刚开始学，建议按顺序来，每节的代码我都验证过可以直接跑；如果已经有基础，可以直接跳到流式输出和 Function Calling 这两节，这是我觉得最有意思的部分

2. 环境准备与认证配置

2.1 安装依赖

pip install openai tiktoken streamlit python-dotenv

2.2 API Key 配置

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

class APIConfig:
    """OpenAI API 配置"""

    # 从环境变量读取，避免硬编码
    API_KEY = os.getenv("OPENAI_API_KEY")
    BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")

    # 模型配置
    DEFAULT_MODEL = "gpt-4o-mini"       # 性价比最高的默认模型
    REASONING_MODEL = "gpt-4o"          # 复杂推理任务
    FAST_MODEL = "gpt-4o-mini"          # 简单快速任务

    # 请求配置
    DEFAULT_TEMPERATURE = 0.7
    DEFAULT_MAX_TOKENS = 2048
    REQUEST_TIMEOUT = 60                 # 请求超时（秒）

    # 重试配置
    MAX_RETRIES = 3
    RETRY_BACKOFF_FACTOR = 2            # 指数退避因子
    RETRY_INITIAL_DELAY = 1             # 初始延迟（秒）

    # 限流配置
    MAX_REQUESTS_PER_MINUTE = 500       # RPM 限制（取决于你的 Tier）
    MAX_TOKENS_PER_MINUTE = 200000      # TPM 限制

    @classmethod
    def validate(cls):
        """验证配置是否完整"""
        if not cls.API_KEY:
            raise ValueError(
                "未找到 OPENAI_API_KEY，请在 .env 文件中设置或设置环境变量"
            )
        if cls.API_KEY == "sk-your-api-key-here":
            raise ValueError("请将 OPENAI_API_KEY 替换为你的真实 API Key")

2.3 .env 文件

# .env
OPENAI_API_KEY=sk-your-api-key-here
OPENAI_BASE_URL=https://api.openai.com/v1

2.4 客户端初始化

# llm_client.py - 基础客户端
from openai import OpenAI
from config import APIConfig


def create_client() -> OpenAI:
    """创建 OpenAI 客户端实例"""
    APIConfig.validate()
    return OpenAI(
        api_key=APIConfig.API_KEY,
        base_url=APIConfig.BASE_URL,
        timeout=APIConfig.REQUEST_TIMEOUT,
        max_retries=0,  # 我们自己管理重试逻辑
    )

3. Chat Completions API 深度解析

3.1 消息角色体系

OpenAI Chat API 用结构化的消息列表来组织对话，这是我刚开始接触时觉得最巧妙的设计。三种消息角色各司其职：

┌─────────────────────────────────────────────────────────┐
│                    消息角色体系                           │
│                                                         │
│  system    : 系统级指令，定义 AI 的行为边界和角色          │
│              → 对应 Prompt 工程中的"角色设定"+"约束条件"    │
│              → 权重最高，AI 会优先遵循                     │
│                                                         │
│  user      : 用户输入，代表人类的问题或指令                 │
│              → 对应 Prompt 工程中的"任务指令"+"上下文"      │
│                                                         │
│  assistant : AI 的历史回复，用于维持对话连贯性              │
│              → 在多轮对话中，需要把历史 assistant 消息传回   │
│                                                         │
│  tool      : 工具调用的结果，Function Calling 专用          │
│              → AI 调用工具后，结果以 tool 角色传回          │
└─────────────────────────────────────────────────────────┘

from openai import OpenAI

client = create_client()

# 基础调用示例
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {
            "role": "system",
            "content": "你是一位Python技术专家。回答要包含可运行的代码示例。"
        },
        {
            "role": "user",
            "content": "如何用Python读取CSV文件并计算某一列的平均值？"
        }
    ],
    temperature=0.7,
    max_tokens=1000,
)

# 提取回复内容
answer = response.choices[0].message.content
print(answer)

# 查看 Token 使用情况
print(f"输入 Token: {response.usage.prompt_tokens}")
print(f"输出 Token: {response.usage.completion_tokens}")
print(f"总计 Token: {response.usage.total_tokens}")

3.2 核心参数详解

参数	类型	默认值	说明
`model`	str	必填	模型名称，如 `gpt-4o`、`gpt-4o-mini`
`messages`	list	必填	消息列表，按对话顺序排列
`temperature`	float	1.0	控制随机性，0=确定，2=最大随机
`max_tokens`	int	无限制	输出 Token 上限（不含输入）
`top_p`	float	1.0	核采样，与 temperature 二选一即可
`n`	int	1	生成多少个候选回复
`stream`	bool	False	是否启用流式输出
`stop`	str/list	None	停止生成的标记
`presence_penalty`	float	0	-2.0~2.0，正值鼓励谈论新话题
`frequency_penalty`	float	0	-2.0~2.0，正值减少重复
`response_format`	dict	None	`{"type": "json_object"}` 强制 JSON
`tools`	list	None	Function Calling 工具定义
`tool_choice`	str/dict	“auto”	工具选择策略

Temperature 选择指南

temperature 值的选择取决于任务类型：

0.0 - 0.2  → 事实性问答、代码生成、数学计算、JSON 输出
             特点：输出稳定、可复现，但缺乏创造性

0.3 - 0.5  → 技术写作、翻译、摘要、数据分析
             特点：在准确性和灵活性之间取得平衡

0.6 - 0.8  → 创意写作、头脑风暴、对话、角色扮演
             特点：输出多样、有创意，但可能不够精确

0.9 - 1.5  → 诗歌创作、故事生成、探索性任务
             特点：高度随机，输出不可预测

3.3 响应结构解析

# 完整的响应对象结构
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "你好"}],
)

# response 对象的关键属性
print(f"响应ID: {response.id}")
print(f"模型: {response.model}")
print(f"创建时间: {response.created}")

# choices 列表（n=1 时通常只有一个）
choice = response.choices[0]
print(f"结束原因: {choice.finish_reason}")
# finish_reason 可能的值：
# - "stop": 正常结束
# - "length": 达到 max_tokens 限制
# - "content_filter": 内容被过滤
# - "tool_calls": 模型请求调用工具
# - "function_call": 模型请求调用函数（旧版）

print(f"消息角色: {choice.message.role}")
print(f"回复内容: {choice.message.content}")
print(f"工具调用: {choice.message.tool_calls}")

# usage 信息
print(f"输入Token: {response.usage.prompt_tokens}")
print(f"输出Token: {response.usage.completion_tokens}")
print(f"总计Token: {response.usage.total_tokens}")

4. 流式输出：原理与实现

4.1 SSE 协议原理

流式输出是我觉得最能提升用户体验的一个特性。它基于 SSE（Server-Sent Events） 协议，原理其实不复杂：

传统请求-响应模式：
客户端 ──请求──→ 服务器 ──等待LLM生成完毕──→ 客户端
         (等待 3-10 秒，用户看到空白)

流式输出模式：
客户端 ──请求──→ 服务器 ──token1──→ 客户端（显示"你好"）
                         ──token2──→ 客户端（显示"你好，我"）
                         ──token3──→ 客户端（显示"你好，我是"）
                         ──token4──→ 客户端（显示"你好，我是AI"）
                         ...
                         ──[DONE]──→ 客户端（完成）
         (用户看到文字逐字出现，体验流畅)

SSE 数据格式：

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"你好"}}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"，我"}}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"是"}}]}

data: [DONE]

4.2 Python 流式客户端实现

# llm_client.py - 流式客户端
from openai import OpenAI
from openai.types.chat import ChatCompletionChunk
from typing import Generator


class StreamingClient:
    """支持流式输出的 OpenAI 客户端"""

    def __init__(self, client: OpenAI, model: str = "gpt-4o-mini"):
        self.client = client
        self.model = model

    def chat_stream(
        self,
        messages: list[dict],
        temperature: float = 0.7,
        max_tokens: int = 2048,
    ) -> Generator[str, None, None]:
        """
        流式对话，逐 token 返回

        Yields:
            str: 每次返回一个增量文本片段
        """
        stream = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=True,
            stream_options={"include_usage": True},  # 获取最终 usage 信息
        )

        for chunk in stream:
            # 检查是否有内容
            if chunk.choices and len(chunk.choices) > 0:
                delta = chunk.choices[0].delta
                if delta.content:
                    yield delta.content

    def chat_stream_with_usage(
        self,
        messages: list[dict],
        temperature: float = 0.7,
        max_tokens: int = 2048,
    ) -> Generator[dict, None, None]:
        """
        流式对话，返回内容和 usage 信息

        Yields:
            dict: {"type": "content", "data": "文本"} 或
                  {"type": "usage", "data": {...}}
        """
        stream = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=True,
            stream_options={"include_usage": True},
        )

        for chunk in stream:
            if chunk.choices and len(chunk.choices) > 0:
                delta = chunk.choices[0].delta
                if delta.content:
                    yield {"type": "content", "data": delta.content}

            # 最后一个 chunk 包含 usage 信息
            if hasattr(chunk, "usage") and chunk.usage:
                yield {
                    "type": "usage",
                    "data": {
                        "prompt_tokens": chunk.usage.prompt_tokens,
                        "completion_tokens": chunk.usage.completion_tokens,
                        "total_tokens": chunk.usage.total_tokens,
                    }
                }


# 使用示例
if __name__ == "__main__":
    client = create_client()
    stream_client = StreamingClient(client)

    messages = [
        {"role": "system", "content": "你是一位Python技术专家。"},
        {"role": "user", "content": "用一句话解释什么是装饰器。"},
    ]

    print("AI 回复: ", end="", flush=True)
    for chunk in stream_client.chat_stream(messages):
        print(chunk, end="", flush=True)
    print()

4.3 前端流式渲染

前端接收流式输出，我了解到通常用 EventSource 或 Fetch API + ReadableStream：

// 前端流式接收示例（JavaScript）
async function chatWithStream(userMessage) {
    const response = await fetch('/api/chat/stream', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ message: userMessage }),
    });

    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = '';

    while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });

        // 解析 SSE 数据
        const lines = buffer.split('\n');
        buffer = lines.pop() || '';  // 保留不完整的行

        for (const line of lines) {
            if (line.startsWith('data: ')) {
                const data = line.slice(6);
                if (data === '[DONE]') return;

                try {
                    const parsed = JSON.parse(data);
                    const content = parsed.choices?.[0]?.delta?.content;
                    if (content) {
                        // 逐字追加到页面
                        appendToChat(content);
                    }
                } catch (e) {
                    // 忽略解析错误
                }
            }
        }
    }
}

5. Function Calling：让 LLM 使用工具

5.1 Function Calling 原理

Function Calling 是我学习过程中最兴奋的发现。它让 LLM 能够"意识到"自己需要外部工具，并生成结构化的工具调用请求。刚开始我以为 LLM 真的会去"执行"函数，后来才搞明白：

核心流程：

用户: "今天北京天气怎么样？"
        │
        ▼
┌──────────────────┐
│   LLM 分析意图    │  → "我需要查询天气，但我没有实时数据"
│   决定调用工具     │  → "应该调用 get_weather 函数"
└──────┬───────────┘
       │ 返回 tool_calls（不是文本回复）
       ▼
┌──────────────────┐
│   你的代码执行     │  → 调用天气 API，获取真实数据
│   获取工具结果     │  → {"temperature": 22, "condition": "晴"}
└──────┬───────────┘
       │ 将结果以 tool 角色传回
       ▼
┌──────────────────┐
│   LLM 整合信息    │  → "北京今天晴天，气温22°C，适合户外活动"
│   生成最终回复     │
└──────────────────┘

我的理解：Function Calling 不是 LLM 真的"执行"了函数，而是 LLM 告诉你"我想调用哪个函数、传什么参数"，由你的代码来实际执行。这个设计很巧妙——LLM 负责"决策"，代码负责"执行"，各司其职。

5.2 工具定义与注册

# function_tools.py
import json
import math
from typing import Any
from datetime import datetime


# ============================================================
# 工具定义（JSON Schema 格式）
# ============================================================

TOOL_DEFINITIONS = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "获取指定城市的当前天气信息。当用户询问天气相关问题时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如'北京'、'上海'、'Tokyo'",
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位，celsius（摄氏度）或 fahrenheit（华氏度）",
                    },
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学计算。支持基本运算、三角函数、对数等。当用户需要进行数学计算时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如'2+3*4'、'sqrt(16)'、'sin(pi/2)'",
                    },
                },
                "required": ["expression"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": "在内部知识库中搜索信息。当用户询问需要专业知识或内部文档的问题时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索查询词",
                    },
                    "top_k": {
                        "type": "integer",
                        "description": "返回结果数量，默认3",
                        "default": 3,
                    },
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "获取当前日期和时间。当用户询问当前时间、日期时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "timezone": {
                        "type": "string",
                        "description": "时区，如'Asia/Shanghai'、'America/New_York'，默认为'Asia/Shanghai'",
                    },
                },
            },
        },
    },
]


# ============================================================
# 工具实现
# ============================================================

class ToolExecutor:
    """工具执行器"""

    def __init__(self):
        # 注册工具名称到实现函数的映射
        self._tools = {
            "get_current_weather": self._get_weather,
            "calculate": self._calculate,
            "search_knowledge_base": self._search_kb,
            "get_current_time": self._get_time,
        }

    def execute(self, tool_name: str, arguments: dict) -> str:
        """执行指定的工具"""
        if tool_name not in self._tools:
            return json.dumps({"error": f"未知工具: {tool_name}"})

        try:
            result = self._tools[tool_name](**arguments)
            return json.dumps(result, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)}, ensure_ascii=False)

    def _get_weather(self, city: str, unit: str = "celsius") -> dict:
        """获取天气（模拟数据，实际应调用天气API）"""
        # 实际项目中替换为真实天气 API 调用
        # 例如：requests.get(f"https://api.weather.com/v1/current?city={city}")
        weather_data = {
            "北京": {"temperature": 22, "condition": "晴", "humidity": 45},
            "上海": {"temperature": 26, "condition": "多云", "humidity": 65},
            "广州": {"temperature": 30, "condition": "阵雨", "humidity": 80},
            "深圳": {"temperature": 29, "condition": "阴", "humidity": 75},
            "Tokyo": {"temperature": 20, "condition": "晴", "humidity": 50},
        }

        default = {"temperature": 18, "condition": "未知", "humidity": 60}
        weather = weather_data.get(city, default)

        if unit == "fahrenheit":
            weather["temperature"] = round(weather["temperature"] * 9 / 5 + 32, 1)

        return {
            "city": city,
            "unit": unit,
            **weather,
            "timestamp": datetime.now().isoformat(),
        }

    def _calculate(self, expression: str) -> dict:
        """安全地执行数学计算"""
        # 白名单：只允许安全的数学函数
        allowed_names = {
            "abs": abs, "round": round, "min": min, "max": max,
            "sum": sum, "pow": pow,
            "sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
            "tan": math.tan, "log": math.log, "log10": math.log10,
            "pi": math.pi, "e": math.e,
        }

        try:
            # 编译表达式（安全沙箱）
            code = compile(expression, "<calc>", "eval")

            # 检查是否使用了未允许的名称
            for name in code.co_names:
                if name not in allowed_names:
                    raise ValueError(f"不允许的函数或变量: {name}")

            result = eval(code, {"__builtins__": {}}, allowed_names)
            return {"expression": expression, "result": result}
        except Exception as e:
            return {"expression": expression, "error": str(e)}

    def _search_kb(self, query: str, top_k: int = 3) -> dict:
        """搜索知识库（模拟数据）"""
        # 实际项目中替换为向量数据库搜索
        kb = {
            "python": "Python 是一种解释型、面向对象的高级编程语言，由 Guido van Rossum 于 1991 年发布。",
            "openai": "OpenAI 是一家人工智能研究公司，开发了 GPT 系列大语言模型。其 API 提供了文本生成、图像生成、语音识别等服务。",
            "rag": "RAG（检索增强生成）是一种结合信息检索和文本生成的 AI 技术架构，通过先检索相关文档再生成回答来提高准确性。",
            "prompt": "Prompt 工程是设计和优化输入给大语言模型的提示文本的技术，目的是获得更准确、更有用的输出。",
        }

        results = []
        for key, value in kb.items():
            if query.lower() in key.lower() or query.lower() in value.lower():
                results.append({"source": key, "content": value})

        return {
            "query": query,
            "results": results[:top_k],
            "total_found": len(results),
        }

    def _get_time(self, timezone: str = "Asia/Shanghai") -> dict:
        """获取当前时间"""
        now = datetime.now()
        return {
            "datetime": now.isoformat(),
            "date": now.strftime("%Y-%m-%d"),
            "time": now.strftime("%H:%M:%S"),
            "weekday": ["周一", "周二", "周三", "周四", "周五", "周六", "周日"][now.weekday()],
            "timezone": timezone,
        }

5.3 完整调用流程

# llm_client.py - Function Calling 客户端
import json
from openai import OpenAI
from function_tools import TOOL_DEFINITIONS, ToolExecutor


class FunctionCallingClient:
    """支持 Function Calling 的客户端"""

    def __init__(self, client: OpenAI, model: str = "gpt-4o-mini"):
        self.client = client
        self.model = model
        self.tool_executor = ToolExecutor()

    def chat_with_tools(
        self,
        messages: list[dict],
        tools: list[dict] = None,
        temperature: float = 0.7,
        max_iterations: int = 5,
    ) -> dict:
        """
        支持工具调用的对话

        会自动处理多轮工具调用，直到 LLM 不再请求工具或达到最大迭代次数。

        Returns:
            dict: {
                "answer": "最终回复文本",
                "tool_calls_made": [...],  # 所有工具调用记录
                "total_tokens": 1234,
            }
        """
        if tools is None:
            tools = TOOL_DEFINITIONS

        tool_calls_made = []
        total_tokens = 0

        for iteration in range(max_iterations):
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=tools,
                tool_choice="auto",  # 让模型自己决定是否调用工具
                temperature=temperature,
            )

            choice = response.choices[0]
            total_tokens += response.usage.total_tokens

            # 情况1：模型想调用工具
            if choice.finish_reason == "tool_calls" and choice.message.tool_calls:
                # 将模型的工具调用请求加入消息历史
                messages.append({
                    "role": "assistant",
                    "content": choice.message.content,
                    "tool_calls": [
                        {
                            "id": tc.id,
                            "type": "function",
                            "function": {
                                "name": tc.function.name,
                                "arguments": tc.function.arguments,
                            },
                        }
                        for tc in choice.message.tool_calls
                    ],
                })

                # 执行每个工具调用
                for tc in choice.message.tool_calls:
                    tool_name = tc.function.name
                    arguments = json.loads(tc.function.arguments)

                    # 执行工具
                    result = self.tool_executor.execute(tool_name, arguments)

                    tool_calls_made.append({
                        "tool": tool_name,
                        "arguments": arguments,
                        "result": result,
                    })

                    # 将工具结果加入消息历史
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": result,
                    })

                # 继续循环，让模型处理工具结果
                continue

            # 情况2：模型直接回复文本
            if choice.message.content:
                return {
                    "answer": choice.message.content,
                    "tool_calls_made": tool_calls_made,
                    "total_tokens": total_tokens,
                }

            # 情况3：异常情况
            return {
                "answer": "抱歉，我无法处理这个请求。",
                "tool_calls_made": tool_calls_made,
                "total_tokens": total_tokens,
            }

        # 达到最大迭代次数
        return {
            "answer": "处理超时，请简化您的问题后重试。",
            "tool_calls_made": tool_calls_made,
            "total_tokens": total_tokens,
        }


# 使用示例
if __name__ == "__main__":
    client = create_client()
    fc_client = FunctionCallingClient(client)

    messages = [
        {"role": "system", "content": "你是一个有用的助手，可以使用工具来回答问题。"},
        {"role": "user", "content": "北京今天天气怎么样？顺便帮我算一下 123 * 456 等于多少。"},
    ]

    result = fc_client.chat_with_tools(messages)
    print(f"回复: {result['answer']}")
    print(f"工具调用: {json.dumps(result['tool_calls_made'], ensure_ascii=False, indent=2)}")
    print(f"Token 消耗: {result['total_tokens']}")

5.4 与 Prompt 工程的结合

Function Calling 不是 Prompt 工程的替代，而是增强。回顾《Prompt 工程指南》中学到的内容，我尝试把两者结合起来：

# prompt_templates.py - 结合 Prompt 工程的 Function Calling
from function_tools import TOOL_DEFINITIONS


class PromptEnhancedToolCalling:
    """结合 Prompt 工程最佳实践的 Function Calling"""

    @staticmethod
    def build_system_prompt_with_tools(
        role: str,
        task: str,
        rules: list[str] = None,
        tool_usage_guide: str = None,
    ) -> str:
        """
        构建包含工具使用指导的 System Prompt

        对应 Prompt 工程中的"分层 Prompt 架构"：
        Layer 1（系统层）：角色 + 规则 + 工具使用指导
        """
        prompt = f"你是一位{role}。\n\n"

        prompt += f"【核心任务】\n{task}\n\n"

        if rules:
            prompt += "【行为规则】\n"
            for i, rule in enumerate(rules, 1):
                prompt += f"{i}. {rule}\n"
            prompt += "\n"

        if tool_usage_guide:
            prompt += f"【工具使用指导】\n{tool_usage_guide}\n\n"
        else:
            prompt += """【工具使用指导】
- 当需要实时数据（天气、时间）或精确计算时，使用工具获取
- 一次可以调用多个工具，工具调用是并行的
- 获取工具结果后，用自然语言整合信息回复用户
- 如果工具返回错误，告知用户并尝试其他方式
"""

        return prompt

    @staticmethod
    def build_few_shot_tool_examples() -> list[dict]:
        """
        构建 Few-shot 工具调用示例

        对应 Prompt 工程中的"少样本提示（Few-shot Prompting）"
        通过示例教会模型何时使用工具、如何使用
        """
        return [
            {"role": "user", "content": "上海今天热吗？"},
            {
                "role": "assistant",
                "content": None,
                "tool_calls": [{
                    "id": "call_example_1",
                    "type": "function",
                    "function": {
                        "name": "get_current_weather",
                        "arguments": '{"city": "上海", "unit": "celsius"}',
                    },
                }],
            },
            {
                "role": "tool",
                "tool_call_id": "call_example_1",
                "content": '{"city": "上海", "temperature": 26, "condition": "多云", "humidity": 65}',
            },
            {
                "role": "assistant",
                "content": "上海今天多云，气温26°C，湿度65%。体感偏热，建议穿短袖，出门注意防晒。",
            },
        ]

    @staticmethod
    def build_cot_tool_prompt(question: str) -> str:
        """
        构建 CoT + 工具调用的 Prompt

        对应 Prompt 工程中的"思维链提示（Chain-of-Thought）"
        引导模型先思考需要什么信息，再决定调用什么工具
        """
        return f"""请按以下步骤思考并回答：

步骤1：分析问题，确定需要哪些信息
步骤2：判断哪些信息可以通过工具获取
步骤3：调用相应的工具
步骤4：整合工具返回的信息，给出最终答案

问题：{question}

请开始分析："""

6. Token 计算与成本控制

6.1 Token 计算原理

Token 是 LLM 处理文本的基本单位。刚开始我以为是按字符或单词算的，后来才发现完全不是那么回事。搞清楚 Token 怎么算，是控制成本的第一步：

Token 不等于字符，也不等于单词：

英文：1 token ≈ 0.75 个单词 ≈ 4 个字符
      "Hello, world!" → 4 tokens

中文：1 token ≈ 0.5-1 个汉字
      "你好世界" → 4-6 tokens（取决于模型的分词器）

代码：1 token ≈ 3-4 个字符
      "def foo():" → 4 tokens

# token_tracker.py
import tiktoken
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class TokenUsage:
    """Token 使用记录"""
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0

    def __add__(self, other: "TokenUsage") -> "TokenUsage":
        return TokenUsage(
            prompt_tokens=self.prompt_tokens + other.prompt_tokens,
            completion_tokens=self.completion_tokens + other.completion_tokens,
            total_tokens=self.total_tokens + other.total_tokens,
        )


class TokenCalculator:
    """Token 计算器"""

    # 各模型的价格（每百万 Token，美元）
    # 价格可能变动，请以 OpenAI 官网为准
    PRICING = {
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "gpt-4-turbo": {"input": 10.00, "output": 30.00},
        "gpt-4": {"input": 30.00, "output": 60.00},
        "gpt-3.5-turbo": {"input": 0.50, "output": 1.50},
    }

    def __init__(self, model: str = "gpt-4o-mini"):
        self.model = model
        try:
            self.encoding = tiktoken.encoding_for_model(model)
        except KeyError:
            self.encoding = tiktoken.get_encoding("cl100k_base")

    def count_tokens(self, text: str) -> int:
        """计算文本的 Token 数量"""
        return len(self.encoding.encode(text))

    def count_messages_tokens(self, messages: list[dict]) -> int:
        """
        计算消息列表的 Token 数量

        参考 OpenAI 的计数规则：
        - 每条消息有固定的开销（约 3-4 tokens）
        - 角色名称计入 token
        - 工具定义也计入 token
        """
        tokens_per_message = 3  # 每条消息的固定开销
        num_tokens = 0

        for message in messages:
            num_tokens += tokens_per_message
            for key, value in message.items():
                if key == "content" and value:
                    num_tokens += self.count_tokens(str(value))
                elif key == "tool_calls":
                    # 工具调用也计入 token
                    num_tokens += self.count_tokens(str(value))
                elif key == "name":
                    num_tokens += self.count_tokens(str(value))

        num_tokens += 3  # 每次回复的固定开销
        return num_tokens

    def estimate_cost(
        self,
        prompt_tokens: int,
        completion_tokens: int,
        model: str = None,
    ) -> dict:
        """估算 API 调用成本"""
        model = model or self.model
        pricing = self.PRICING.get(model, {"input": 0, "output": 0})

        input_cost = (prompt_tokens / 1_000_000) * pricing["input"]
        output_cost = (completion_tokens / 1_000_000) * pricing["output"]
        total_cost = input_cost + output_cost

        return {
            "model": model,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "input_cost_usd": round(input_cost, 6),
            "output_cost_usd": round(output_cost, 6),
            "total_cost_usd": round(total_cost, 6),
            "total_cost_cny": round(total_cost * 7.2, 6),  # 粗略汇率
        }


class TokenTracker:
    """Token 使用追踪器（用于统计整个会话的成本）"""

    def __init__(self, model: str = "gpt-4o-mini"):
        self.model = model
        self.calculator = TokenCalculator(model)
        self.usage = TokenUsage()
        self.call_history: list[dict] = []

    def record(self, prompt_tokens: int, completion_tokens: int):
        """记录一次 API 调用的 Token 使用"""
        self.usage.prompt_tokens += prompt_tokens
        self.usage.completion_tokens += completion_tokens
        self.usage.total_tokens += prompt_tokens + completion_tokens

        cost = self.calculator.estimate_cost(prompt_tokens, completion_tokens)
        self.call_history.append(cost)

    def get_summary(self) -> dict:
        """获取使用摘要"""
        total_cost = self.calculator.estimate_cost(
            self.usage.prompt_tokens,
            self.usage.completion_tokens,
        )
        return {
            "model": self.model,
            "total_calls": len(self.call_history),
            "total_prompt_tokens": self.usage.prompt_tokens,
            "total_completion_tokens": self.usage.completion_tokens,
            "total_tokens": self.usage.total_tokens,
            "total_cost_usd": total_cost["total_cost_usd"],
            "total_cost_cny": total_cost["total_cost_cny"],
            "call_history": self.call_history,
        }


# 使用示例
if __name__ == "__main__":
    calc = TokenCalculator("gpt-4o-mini")

    text = "Prompt 工程是设计和优化输入给大语言模型的提示文本的技术。"
    tokens = calc.count_tokens(text)
    print(f"文本: {text}")
    print(f"Token 数: {tokens}")

    messages = [
        {"role": "system", "content": "你是一个有用的助手。"},
        {"role": "user", "content": "解释什么是 RAG。"},
    ]
    msg_tokens = calc.count_messages_tokens(messages)
    print(f"消息 Token 数: {msg_tokens}")

    cost = calc.estimate_cost(500, 200)
    print(f"预估成本: ${cost['total_cost_usd']} (约 {cost['total_cost_cny']} 元)")

6.2 成本估算与优化

# 成本优化策略汇总

class CostOptimizer:
    """API 成本优化器"""

    @staticmethod
    def select_model(task_complexity: str) -> str:
        """
        根据任务复杂度选择模型

        简单任务用便宜模型，复杂任务用强模型
        """
        model_map = {
            "simple": "gpt-4o-mini",      # 简单问答、分类 → $0.15/M input
            "moderate": "gpt-4o-mini",    # 摘要、翻译 → $0.15/M input
            "complex": "gpt-4o",          # 复杂推理、代码生成 → $2.50/M input
            "critical": "gpt-4o",         # 关键业务 → $2.50/M input
        }
        return model_map.get(task_complexity, "gpt-4o-mini")

    @staticmethod
    def optimize_messages(
        messages: list[dict],
        max_context_tokens: int = 4000,
    ) -> list[dict]:
        """
        优化消息列表，控制上下文长度

        策略：
        1. System Prompt 始终保留
        2. 保留最近的 N 轮对话
        3. 超出部分用摘要替代
        """
        if len(messages) <= 3:
            return messages

        calc = TokenCalculator()
        system_msgs = [m for m in messages if m["role"] == "system"]
        other_msgs = [m for m in messages if m["role"] != "system"]

        # 从后往前保留，直到接近 token 限制
        kept = []
        current_tokens = sum(calc.count_tokens(m.get("content", "")) for m in system_msgs)

        for msg in reversed(other_msgs):
            msg_tokens = calc.count_tokens(msg.get("content", ""))
            if current_tokens + msg_tokens > max_context_tokens:
                break
            kept.insert(0, msg)
            current_tokens += msg_tokens

        return system_msgs + kept

    @staticmethod
    def estimate_conversation_cost(
        avg_input_tokens: int,
        avg_output_tokens: int,
        conversations_per_day: int,
        model: str = "gpt-4o-mini",
    ) -> dict:
        """估算每日/每月成本"""
        calc = TokenCalculator(model)
        daily_input = avg_input_tokens * conversations_per_day
        daily_output = avg_output_tokens * conversations_per_day

        daily_cost = calc.estimate_cost(daily_input, daily_output)
        monthly_cost = calc.estimate_cost(daily_input * 30, daily_output * 30)

        return {
            "model": model,
            "conversations_per_day": conversations_per_day,
            "daily_cost_usd": daily_cost["total_cost_usd"],
            "daily_cost_cny": daily_cost["total_cost_cny"],
            "monthly_cost_usd": monthly_cost["total_cost_usd"],
            "monthly_cost_cny": monthly_cost["total_cost_cny"],
        }


# 使用示例
optimizer = CostOptimizer()
cost_estimate = optimizer.estimate_conversation_cost(
    avg_input_tokens=500,
    avg_output_tokens=300,
    conversations_per_day=100,
    model="gpt-4o-mini",
)
print(f"使用 gpt-4o-mini，每天 100 次对话：")
print(f"  每日成本: ${cost_estimate['daily_cost_usd']:.4f} (约 {cost_estimate['daily_cost_cny']:.2f} 元)")
print(f"  每月成本: ${cost_estimate['monthly_cost_usd']:.4f} (约 {cost_estimate['monthly_cost_cny']:.2f} 元)")

7. 错误处理与重试策略

7.1 错误类型分类

实际跑起来之后，我发现 API 调用出错的概率比想象中高得多。网络波动、限流、超时、内容过滤……各种情况都遇到过。我把常见的错误整理成了下面这张表：

错误类型	HTTP 状态码	典型场景	是否可重试
`RateLimitError`	429	超过 RPM/TPM 限制	是（等待后重试）
`APIConnectionError`	-	网络问题、DNS 解析失败	是
`APITimeoutError`	-	请求超时	是
`InternalServerError`	500	OpenAI 服务器内部错误	是
`ServiceUnavailableError`	503	服务暂时不可用	是
`AuthenticationError`	401	API Key 无效或过期	否
`PermissionDeniedError`	403	无权限访问该资源	否
`InvalidRequestError`	400	请求参数错误	否（修正参数后重试）
`ContentFilterError`	400	内容被安全过滤	否

7.2 指数退避重试

# llm_client.py - 带重试的客户端
import time
import random
from functools import wraps
from openai import (
    OpenAI,
    RateLimitError,
    APIConnectionError,
    APITimeoutError,
    InternalServerError,
    APIStatusError,
)


class RetryConfig:
    """重试配置"""

    def __init__(
        self,
        max_retries: int = 3,
        initial_delay: float = 1.0,
        backoff_factor: float = 2.0,
        max_delay: float = 60.0,
        jitter: bool = True,
        retryable_errors: tuple = (
            RateLimitError,
            APIConnectionError,
            APITimeoutError,
            InternalServerError,
        ),
    ):
        self.max_retries = max_retries
        self.initial_delay = initial_delay
        self.backoff_factor = backoff_factor
        self.max_delay = max_delay
        self.jitter = jitter
        self.retryable_errors = retryable_errors

    def get_delay(self, attempt: int) -> float:
        """计算第 N 次重试的延迟时间（指数退避）"""
        delay = self.initial_delay * (self.backoff_factor ** attempt)
        delay = min(delay, self.max_delay)

        if self.jitter:
            # 添加随机抖动，避免"惊群效应"
            delay = delay * (0.5 + random.random())

        return delay


def with_retry(retry_config: RetryConfig = None):
    """重试装饰器"""
    if retry_config is None:
        retry_config = RetryConfig()

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None

            for attempt in range(retry_config.max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except retry_config.retryable_errors as e:
                    last_error = e

                    if attempt == retry_config.max_retries:
                        # 达到最大重试次数，抛出最终错误
                        raise RuntimeError(
                            f"API 调用失败，已重试 {retry_config.max_retries} 次。"
                            f"最后错误: {type(e).__name__}: {str(e)[:200]}"
                        ) from e

                    delay = retry_config.get_delay(attempt)
                    print(
                        f"[重试 {attempt + 1}/{retry_config.max_retries}] "
                        f"{type(e).__name__}，等待 {delay:.1f} 秒..."
                    )
                    time.sleep(delay)

                except APIStatusError as e:
                    # 非可重试错误，直接抛出
                    raise RuntimeError(
                        f"API 返回错误 (HTTP {e.status_code}): {str(e)[:500]}"
                    ) from e

            # 理论上不会到这里
            raise last_error

        return wrapper
    return decorator


class RobustClient:
    """带重试和错误处理的 OpenAI 客户端"""

    def __init__(
        self,
        client: OpenAI,
        model: str = "gpt-4o-mini",
        retry_config: RetryConfig = None,
    ):
        self.client = client
        self.model = model
        self.retry_config = retry_config or RetryConfig()

    def chat(
        self,
        messages: list[dict],
        temperature: float = 0.7,
        max_tokens: int = 2048,
        **kwargs,
    ) -> dict:
        """
        带重试的对话接口

        Returns:
            dict: {
                "success": True/False,
                "content": "回复内容" or None,
                "error": "错误信息" or None,
                "usage": {...} or None,
                "retries": 0,
            }
        """
        last_error = None

        for attempt in range(self.retry_config.max_retries + 1):
            try:
                response = self.client.chat.completions.create(
                    model=self.model,
                    messages=messages,
                    temperature=temperature,
                    max_tokens=max_tokens,
                    **kwargs,
                )

                return {
                    "success": True,
                    "content": response.choices[0].message.content,
                    "error": None,
                    "usage": {
                        "prompt_tokens": response.usage.prompt_tokens,
                        "completion_tokens": response.usage.completion_tokens,
                        "total_tokens": response.usage.total_tokens,
                    },
                    "retries": attempt,
                    "finish_reason": response.choices[0].finish_reason,
                }

            except RateLimitError as e:
                last_error = e
                if attempt < self.retry_config.max_retries:
                    delay = self.retry_config.get_delay(attempt)
                    print(f"[限流] 等待 {delay:.1f} 秒后重试...")
                    time.sleep(delay)

            except (APIConnectionError, APITimeoutError, InternalServerError) as e:
                last_error = e
                if attempt < self.retry_config.max_retries:
                    delay = self.retry_config.get_delay(attempt)
                    print(f"[{type(e).__name__}] 等待 {delay:.1f} 秒后重试...")
                    time.sleep(delay)

            except Exception as e:
                # 不可重试的错误
                return {
                    "success": False,
                    "content": None,
                    "error": f"{type(e).__name__}: {str(e)[:300]}",
                    "usage": None,
                    "retries": attempt,
                    "finish_reason": None,
                }

        # 所有重试都失败了
        return {
            "success": False,
            "content": None,
            "error": f"重试 {self.retry_config.max_retries} 次后仍然失败: {type(last_error).__name__}",
            "usage": None,
            "retries": self.retry_config.max_retries,
            "finish_reason": None,
        }

7.3 限流处理

# rate_limiter.py
import time
import threading
from collections import deque
from dataclasses import dataclass


@dataclass
class RateLimitConfig:
    """限流配置"""
    max_requests_per_minute: int = 500
    max_tokens_per_minute: int = 200000
    # 安全余量：实际限制设为官方限制的 80%
    safety_margin: float = 0.8


class TokenBucketRateLimiter:
    """令牌桶限流器"""

    def __init__(self, config: RateLimitConfig = None):
        self.config = config or RateLimitConfig()
        self.lock = threading.Lock()

        # 计算带安全余量的实际限制
        self.max_rpm = int(self.config.max_requests_per_minute * self.config.safety_margin)
        self.max_tpm = int(self.config.max_tokens_per_minute * self.config.safety_margin)

        # 请求记录（滑动窗口）
        self.request_times: deque = deque()
        self.token_usage: deque = deque()  # (timestamp, tokens)

    def _clean_old_records(self):
        """清理超过 1 分钟的旧记录"""
        now = time.time()
        one_minute_ago = now - 60

        while self.request_times and self.request_times[0] < one_minute_ago:
            self.request_times.popleft()

        while self.token_usage and self.token_usage[0][0] < one_minute_ago:
            self.token_usage.popleft()

    def acquire(self, estimated_tokens: int = 0) -> bool:
        """
        尝试获取请求许可

        Args:
            estimated_tokens: 预估的 Token 消耗

        Returns:
            bool: 是否允许发送请求
        """
        with self.lock:
            self._clean_old_records()

            # 检查请求频率
            if len(self.request_times) >= self.max_rpm:
                return False

            # 检查 Token 消耗
            if estimated_tokens > 0:
                current_tokens = sum(t[1] for t in self.token_usage)
                if current_tokens + estimated_tokens > self.max_tpm:
                    return False

            # 记录本次请求
            now = time.time()
            self.request_times.append(now)
            if estimated_tokens > 0:
                self.token_usage.append((now, estimated_tokens))

            return True

    def wait_and_acquire(self, estimated_tokens: int = 0, timeout: float = 30.0) -> bool:
        """等待直到获取许可或超时"""
        start = time.time()

        while time.time() - start < timeout:
            if self.acquire(estimated_tokens):
                return True
            time.sleep(0.1)

        return False

    def get_stats(self) -> dict:
        """获取当前限流统计"""
        with self.lock:
            self._clean_old_records()
            return {
                "requests_this_minute": len(self.request_times),
                "requests_limit": self.max_rpm,
                "requests_usage_pct": round(len(self.request_times) / self.max_rpm * 100, 1),
                "tokens_this_minute": sum(t[1] for t in self.token_usage),
                "tokens_limit": self.max_tpm,
                "tokens_usage_pct": round(
                    sum(t[1] for t in self.token_usage) / self.max_tpm * 100, 1
                ),
            }

8. 多轮对话管理

多轮对话看起来简单，就是把历史消息传回去嘛。但实际做的时候发现有两个棘手的问题：一是上下文越来越长，Token 消耗越来越大；二是超过模型上下文窗口后，前面的对话就丢了。下面是我摸索出来的处理方案。

8.1 对话上下文窗口管理

# conversation_manager.py
from token_tracker import TokenCalculator


class ConversationManager:
    """多轮对话管理器"""

    def __init__(
        self,
        system_prompt: str,
        model: str = "gpt-4o-mini",
        max_context_tokens: int = 8000,
        max_turns: int = 20,
    ):
        self.system_prompt = system_prompt
        self.model = model
        self.max_context_tokens = max_context_tokens
        self.max_turns = max_turns
        self.token_calc = TokenCalculator(model)

        # 对话历史
        self.messages: list[dict] = [
            {"role": "system", "content": system_prompt}
        ]

    def add_user_message(self, content: str):
        """添加用户消息"""
        self.messages.append({"role": "user", "content": content})

    def add_assistant_message(self, content: str):
        """添加助手回复"""
        self.messages.append({"role": "assistant", "content": content})

    def get_messages(self) -> list[dict]:
        """获取当前消息列表（自动裁剪）"""
        self._trim_if_needed()
        return self.messages

    def _trim_if_needed(self):
        """如果上下文过长，自动裁剪"""
        total_tokens = self.token_calc.count_messages_tokens(self.messages)

        if total_tokens <= self.max_context_tokens:
            return

        # 策略：保留 system prompt + 最近的对话
        # 被裁剪的旧对话用摘要替代
        system_msg = self.messages[0]
        conversation = self.messages[1:]

        # 从后往前保留
        kept = []
        current_tokens = self.token_calc.count_tokens(system_msg["content"])

        for msg in reversed(conversation):
            msg_tokens = self.token_calc.count_tokens(msg.get("content", ""))
            if current_tokens + msg_tokens > self.max_context_tokens:
                break
            kept.insert(0, msg)
            current_tokens += msg_tokens

        # 如果有被裁剪的消息，在开头插入摘要提示
        if len(kept) < len(conversation):
            trimmed_count = len(conversation) - len(kept)
            summary_msg = {
                "role": "system",
                "content": f"[注意：由于上下文长度限制，已省略较早的 {trimmed_count} 轮对话。请基于当前可见的对话继续。]",
            }
            self.messages = [system_msg, summary_msg] + kept
        else:
            self.messages = [system_msg] + kept

    def get_conversation_summary(self) -> dict:
        """获取对话摘要"""
        user_messages = [m for m in self.messages if m["role"] == "user"]
        assistant_messages = [m for m in self.messages if m["role"] == "assistant"]

        total_tokens = self.token_calc.count_messages_tokens(self.messages)

        return {
            "total_messages": len(self.messages),
            "user_messages": len(user_messages),
            "assistant_messages": len(assistant_messages),
            "total_turns": len(user_messages),
            "total_tokens": total_tokens,
            "context_usage_pct": round(
                total_tokens / self.max_context_tokens * 100, 1
            ),
        }

    def clear(self):
        """清空对话历史（保留 system prompt）"""
        self.messages = [
            {"role": "system", "content": self.system_prompt}
        ]

8.2 对话摘要与压缩

class ConversationCompressor:
    """对话压缩器 —— 将长对话压缩为摘要"""

    def __init__(self, client: OpenAI, model: str = "gpt-4o-mini"):
        self.client = client
        self.model = model

    def summarize_history(
        self,
        messages: list[dict],
        keep_last_n: int = 3,
    ) -> list[dict]:
        """
        压缩对话历史

        策略：
        1. 保留最近 N 轮对话不变
        2. 将更早的对话压缩为一段摘要
        3. 摘要作为 system 消息的一部分注入
        """
        if len(messages) <= keep_last_n * 2 + 2:
            return messages  # 不需要压缩

        system_msg = messages[0]
        recent = messages[-(keep_last_n * 2):]  # 保留最近 N 轮（每轮 user+assistant）
        old = messages[1:-(keep_last_n * 2)]    # 需要压缩的旧对话

        # 构建摘要请求
        old_conversation = "\n".join([
            f"{'用户' if m['role'] == 'user' else '助手'}: {m.get('content', '')[:200]}"
            for m in old
            if m["role"] in ("user", "assistant")
        ])

        summary_prompt = f"""请将以下对话历史压缩为一段简洁的摘要（不超过200字）。
只保留关键信息：用户的主要问题、助手给出的重要结论、任何需要记住的上下文。

对话历史：
{old_conversation}

摘要："""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": summary_prompt}],
            temperature=0.3,
            max_tokens=300,
        )

        summary = response.choices[0].message.content

        # 构建新的消息列表
        compressed = [
            system_msg,
            {
                "role": "system",
                "content": f"[历史对话摘要] {summary}",
            },
        ] + recent

        return compressed

9. 结构化输出

让 LLM 输出 JSON 格式在实际开发中太常用了——对接前端、存入数据库、调用下游服务，都需要结构化的数据。我摸索了两种方式：

# structured_output.py
import json
from openai import OpenAI


class StructuredOutputClient:
    """结构化输出客户端"""

    def __init__(self, client: OpenAI, model: str = "gpt-4o-mini"):
        self.client = client
        self.model = model

    def json_output(
        self,
        prompt: str,
        system_prompt: str = None,
        temperature: float = 0.1,
    ) -> dict:
        """
        获取 JSON 格式的输出

        使用 response_format 参数强制 JSON 输出
        """
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})

        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            response_format={"type": "json_object"},
        )

        content = response.choices[0].message.content
        return json.loads(content)

    def json_with_schema(
        self,
        prompt: str,
        json_schema: dict,
        system_prompt: str = None,
    ) -> dict:
        """
        获取符合 JSON Schema 的输出

        通过在 Prompt 中嵌入 Schema 来约束输出结构
        """
        schema_text = json.dumps(json_schema, ensure_ascii=False, indent=2)

        full_prompt = f"""{prompt}

【输出格式要求】
请严格按照以下 JSON Schema 输出，确保所有必填字段都存在：

{schema_text}

只输出 JSON，不要包含其他文字。"""

        return self.json_output(full_prompt, system_prompt)


# 使用示例
if __name__ == "__main__":
    client = create_client()
    so_client = StructuredOutputClient(client)

    # 定义输出 Schema
    analysis_schema = {
        "type": "object",
        "properties": {
            "sentiment": {
                "type": "string",
                "enum": ["positive", "negative", "neutral"],
                "description": "情感倾向",
            },
            "confidence": {
                "type": "number",
                "minimum": 0,
                "maximum": 100,
                "description": "置信度",
            },
            "keywords": {
                "type": "array",
                "items": {"type": "string"},
                "description": "关键词列表",
            },
            "summary": {
                "type": "string",
                "maxLength": 100,
                "description": "一句话摘要",
            },
        },
        "required": ["sentiment", "confidence", "keywords", "summary"],
    }

    result = so_client.json_with_schema(
        prompt="分析以下评论：'产品质量很好，但物流太慢了，等了一周才到。'",
        json_schema=analysis_schema,
        system_prompt="你是一个专业的文本分析助手。",
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))

10. 生产环境最佳实践

把代码从"能跑"变成"能上线"，中间还有不少坑要填。下面是我整理的一些实践经验。

10.1 架构设计

┌─────────────────────────────────────────────────────────┐
│                   生产环境 API 调用架构                     │
│                                                         │
│  应用层                                                  │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐              │
│  │ 对话管理  │  │ Prompt   │  │ 结果缓存  │              │
│  │          │  │ 模板引擎  │  │          │              │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘              │
│       │             │             │                     │
│       └─────────────┼─────────────┘                     │
│                     │                                   │
│               ┌─────┴─────┐                             │
│               │  API 网关  │  ← 统一入口                  │
│               └─────┬─────┘                             │
│                     │                                   │
│  中间层           ┌──┴──┐                                │
│               │ 限流器 │  ← Token Bucket                │
│               └──┬──┘                                   │
│               ┌──┴──┐                                   │
│               │ 重试  │  ← 指数退避                      │
│               └──┬──┘                                   │
│               ┌──┴──┐                                   │
│               │ 降级  │  ← 主模型不可用时切换备用模型       │
│               └──┬──┘                                   │
│                     │                                   │
│  API 层         ┌───┴───┐                                │
│               │ OpenAI │                                │
│               │  API   │                                │
│               └───────┘                                 │
└─────────────────────────────────────────────────────────┘

10.2 我的实践清单

实践	说明	优先级
API Key 环境变量化	永远不要硬编码 API Key	必须
请求日志记录	记录每次调用的模型、Token、延迟、成本	必须
指数退避重试	处理网络波动和限流	必须
限流保护	避免超出 API 配额	必须
超时设置	避免请求无限等待	必须
内容安全过滤	输入和输出都需要安全检查	必须
模型降级	主模型不可用时切换备用模型	推荐
响应缓存	相同问题缓存结果，减少 API 调用	推荐
异步调用	使用异步客户端提升吞吐量	推荐
健康检查	定期检查 API 可用性	推荐

11. 配套项目：AI 对话应用

学完这些之后，我动手搭了一个基于 Streamlit 的 AI 对话应用，把上面讲的所有技术都集成进去了。跑通这个项目，基本就掌握了 OpenAI API 的核心用法。

项目结构

ai-chat-app/
├── app.py                 # Streamlit 主应用
├── config.py              # 配置管理
├── llm_client.py          # OpenAI API 客户端封装
├── function_tools.py      # Function Calling 工具定义与实现
├── prompt_templates.py    # Prompt 模板（结合 Prompt 工程）
├── token_tracker.py       # Token 计算与成本追踪
├── conversation_manager.py # 多轮对话管理
├── requirements.txt       # 项目依赖
└── .env.example           # 环境变量模板

应用功能

流式对话：AI 回复逐字显示，体验流畅
Function Calling：支持天气查询、数学计算、知识库搜索、时间查询
Token 追踪：实时显示 Token 消耗和成本估算
Prompt 模板切换：内置标准、Few-shot、CoT、ReAct、代码助手、翻译助手六种 Prompt 模式
错误处理：自动重试、限流保护、优雅降级
对话导出：支持导出对话记录

项目文件详见配套代码目录 `d:\aiagent\学习资料\ai-chat-app\`

12. 总结与回顾

我的学习历程

回头看这段学习经历，大概经历了四个阶段：

第一阶段：基础调用
  → 跑通第一个 Chat Completions API 调用
  → 搞懂消息角色体系（system/user/assistant）
  → 摸清 temperature、max_tokens 这些参数的作用

第二阶段：流式与工具
  → 实现流式输出，体验瞬间提升
  → 啃下 Function Calling，让 LLM 能"干活"了
  → 尝试把 Prompt 工程的最佳实践融入 API 调用

第三阶段：工程化
  → 学会计算 Token，开始关注成本
  → 加上错误处理和重试，应用不再动不动就崩
  → 处理多轮对话的上下文管理

第四阶段：打磨
  → 加上限流保护
  → 实现对话压缩
  → 搭了一个完整的 Streamlit 应用来验证

几点心得

API Key 是敏感信息：永远不要提交到版本控制，用 .env 管理
流式输出是体验的分水岭：加了之后用户感知的等待时间从 5 秒降到几乎为 0
Function Calling 是能力放大器：让 LLM 从"聊天机器人"变成"智能助手"
Token 即成本：每次 API 调用都在花钱，养成计算 Token 的习惯
错误处理不是可选项：生产环境中网络波动、限流、服务故障是常态，不处理根本没法用
Prompt 工程 + API 工程 = 完整的 AI 应用能力：两者缺一不可

与 Prompt 工程文档的关系

本文是《Prompt 工程指南》的实践篇。我个人是按以下顺序学习的，供参考：

先读《Prompt 工程指南》，理解 Prompt 设计的理论基础
再读本文，学习如何通过 API 将 Prompt 设计落地
跑通配套项目，在实践中巩固所学

我参考的资料

OpenAI API 官方文档：https://platform.openai.com/docs/api-reference
OpenAI Cookbook：https://cookbook.openai.com/
tiktoken 仓库：https://github.com/openai/tiktoken
Streamlit 文档：https://docs.streamlit.io/

文档版本：v1.0 | 更新时间：2026-05-10 | 适用读者：AI 应用开发者 / LLM 工程师 / 对 OpenAI API 感兴趣的学习者

本文为原创技术文档，配套项目代码位于 https://github.com/yxy008/ai-prompt.git。所有代码示例均可直接运行。如需转载，请注明出处。

2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

项目拆解（二）：模型输出是‘野‘的——三层兜底+重分析，把AI变成可靠零件

前两层都过了，数据入库了，最后一步是上屏。但这里还有坑——模型在"无风险"时会返回什么？真机实测：它返回了[""]，甚至[" "]（带空格的空串）。前端直接渲染，用户看到一个红色的空警告框，吓一跳。?${esc">无，这段代码挺干净</span>'">无，这段代码挺干净</span>'先t.trim()再。只挡不住[" "]（带空格的字符串是真值），必须先清空格再过滤。这是真机踩出来的：第一次只写