跳转至

第三章：结构化输出与对话基础

前两章解决了两件事：你能调 API 拿到回复（第一章），能让模型调函数拿外部数据（第二章）。

但有个问题一直没处理：模型返回的是一段散文。你的旅游助手想把行程塞进前端卡片、存进数据库、传给下游函数——散文不行。

这一章把 One-shot 域里剩下的能力全部补齐：

  1. 结构化输出——让模型吐 JSON，而不是散文。
  2. Provider 抽象——不绑死一家，换模型只改一行。
  3. 对话历史——多轮对话怎么管消息列表。
  4. 工具调用实践——用我们自己写的抽象层，做一个完整的旅游助手。

3.1 结构化输出

3.1.1 为什么需要结构化输出

先看一个真实场景。你让模型推荐三天的北京行程：

import openai

client = openai.OpenAI()

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "user", "content": "帮我规划三天北京旅游行程，每天列出景点和预算"}
    ],
)
print(resp.choices[0].message.content)

输出大概长这样：

当然可以！以下是三天北京旅游行程建议：

第一天：故宫 + 天安门广场
- 上午参观故宫博物院（门票60元）
- 下午天安门广场（免费）
- 晚上王府井小吃街（约100元）
预计花费：约160元

第二天：长城
- 八达岭长城一日游（门票40元，交通约50元）
- 晚餐：烤鸭（约150元）
预计花费：约240元

第三天：颐和园 + 南锣鼓巷
...

看起来不错？但你想把它渲染成前端卡片，问题来了：

  • "约160元"——是 160 还是 "约160"？正则能匹配，但下次模型写成"大概160左右"就炸了。
  • 景点名字有时带书名号，有时不带。
  • 第几天、哪些景点、预算——全靠你自己从散文里刮。

结论：只要输出要被代码消费，就必须结构化。 散文是给人看的，JSON 是给程序看的。

3.1.2 JSON 模式

OpenAI：response_format

OpenAI 从 GPT-4o 开始支持 response_format={"type": "json_object"}，强制模型输出合法 JSON。

import openai
import json

client = openai.OpenAI()

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {
            "role": "system",
            "content": (
                "你是旅游规划助手。"
                "请以 JSON 格式返回行程，schema 如下：\n"
                '{"days": [{"day": 1, "spots": [{"name": "...", "fee": 0}], "budget": 0}]}'
            ),
        },
        {
            "role": "user",
            "content": "三天北京旅游行程",
        },
    ],
    response_format={"type": "json_object"},
)

itinerary = json.loads(resp.choices[0].message.content)
print(json.dumps(itinerary, ensure_ascii=False, indent=2))

完整响应体：

{
  "days": [
    {
      "day": 1,
      "spots": [
        {"name": "故宫博物院", "fee": 60},
        {"name": "天安门广场", "fee": 0},
        {"name": "王府井小吃街", "fee": 100}
      ],
      "budget": 160
    },
    {
      "day": 2,
      "spots": [
        {"name": "八达岭长城", "fee": 40},
        {"name": "全聚德烤鸭", "fee": 150}
      ],
      "budget": 190
    },
    {
      "day": 3,
      "spots": [
        {"name": "颐和园", "fee": 30},
        {"name": "南锣鼓巷", "fee": 80}
      ],
      "budget": 110
    }
  ]
}

干净。直接 itinerary["days"][0]["budget"] 拿到 160，不用正则。

注意：json_object 模式只保证输出是合法 JSON，不保证 schema 对。模型可能多加字段、少字段、类型不对。后面 3.1.4 讲怎么校验。

OpenAI Structured Outputs（更严格）

如果你需要 schema 级别的保证，用 response_format={"type": "json_schema", "json_schema": ...}：

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "你是旅游规划助手。"},
        {"role": "user", "content": "三天北京旅游行程"},
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "itinerary",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "days": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "day": {"type": "integer"},
                                "spots": {
                                    "type": "array",
                                    "items": {
                                        "type": "object",
                                        "properties": {
                                            "name": {"type": "string"},
                                            "fee": {"type": "number"},
                                        },
                                        "required": ["name", "fee"],
                                        "additionalProperties": False,
                                    },
                                },
                                "budget": {"type": "number"},
                            },
                            "required": ["day", "spots", "budget"],
                            "additionalProperties": False,
                        },
                    }
                },
                "required": ["days"],
                "additionalProperties": False,
            },
        },
    },
)

strict: True 意味着模型的输出被约束在你给的 schema 里，不会多字段也不会少字段。代价是首次请求会多几百毫秒编译 schema。

Anthropic 的做法

Anthropic 没有 response_format 参数。它的方法是用 tool use 来"骗"出结构化输出——定义一个 tool，让模型"调用"它，调用参数就是你要的 JSON。

import anthropic
import json

client = anthropic.Anthropic()

resp = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system="你是旅游规划助手。用 output_itinerary 工具返回行程。",
    messages=[
        {"role": "user", "content": "三天北京旅游行程"}
    ],
    tools=[
        {
            "name": "output_itinerary",
            "description": "输出旅游行程",
            "input_schema": {
                "type": "object",
                "properties": {
                    "days": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "day": {"type": "integer"},
                                "spots": {
                                    "type": "array",
                                    "items": {
                                        "type": "object",
                                        "properties": {
                                            "name": {"type": "string"},
                                            "fee": {"type": "number"},
                                        },
                                        "required": ["name", "fee"],
                                    },
                                },
                                "budget": {"type": "number"},
                            },
                            "required": ["day", "spots", "budget"],
                        },
                    }
                },
                "required": ["days"],
            },
        }
    ],
    tool_choice={"type": "tool", "name": "output_itinerary"},
)

# 从 tool_use block 里拿结果
for block in resp.content:
    if block.type == "tool_use":
        itinerary = block.input
        print(json.dumps(itinerary, ensure_ascii=False, indent=2))

输出：

{
  "days": [
    {
      "day": 1,
      "spots": [
        {"name": "故宫博物院", "fee": 60},
        {"name": "景山公园", "fee": 2},
        {"name": "王府井步行街", "fee": 100}
      ],
      "budget": 162
    },
    {
      "day": 2,
      "spots": [
        {"name": "八达岭长城", "fee": 45},
        {"name": "鸟巢/水立方外景", "fee": 0}
      ],
      "budget": 145
    },
    {
      "day": 3,
      "spots": [
        {"name": "颐和园", "fee": 30},
        {"name": "圆明园", "fee": 10},
        {"name": "清华/北大校园", "fee": 0}
      ],
      "budget": 120
    }
  ]
}

这招看起来 hacky，但它是 Anthropic 官方推荐的结构化输出方式。tool_choice 强制模型必须调用这个 tool，参数就是你要的数据。

3.1.3 不只是 JSON

JSON 是最常用的，但不是唯一选择。看看什么场景用什么格式。

YAML

适合配置文件、人需要手动编辑的场景。

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {
            "role": "system",
            "content": "用 YAML 格式返回行程。只返回 YAML，不要其他文字。",
        },
        {"role": "user", "content": "两天上海行程"},
    ],
)
print(resp.choices[0].message.content)

days:
  - day: 1
    spots:
      - name: 外滩
        fee: 0
      - name: 南京路步行街
        fee: 50
      - name: 豫园
        fee: 40
    budget: 90
  - day: 2
    spots:
      - name: 东方明珠
        fee: 120
      - name: 陆家嘴
        fee: 0
    budget: 120

YAML 比 JSON 可读性好，但解析时注意：yaml.safe_load()，不要用 yaml.load()（安全问题）。

XML

适合需要嵌套属性、或者下游系统只吃 XML 的场景（比如一些旅游 GDS 接口）。

<itinerary>
  <day number="1">
    <spot name="外滩" fee="0"/>
    <spot name="南京路步行街" fee="50"/>
    <budget>90</budget>
  </day>
  <day number="2">
    <spot name="东方明珠" fee="120"/>
    <budget>120</budget>
  </day>
</itinerary>

Markdown 表格

适合直接展示给用户、不需要程序解析的场景。

| 天数 | 景点 | 费用 |
|------|------|------|
| 1 | 外滩 | 0 |
| 1 | 南京路步行街 | 50 |
| 2 | 东方明珠 | 120 |

怎么选

格式 程序解析 人可读 schema 校验 适合场景
JSON 最好 一般 有成熟工具链 API 通信、存储、前后端交互
YAML 好 最好 有但不如 JSON 配置文件、人工编辑
XML 好 差 有（XSD） 遗留系统、GDS 接口
Markdown 差 好 无 直接展示

旅游助手的选择：JSON。 因为行程数据要被前端渲染、存进数据库、传给下游工具。JSON 生态最成熟，校验最方便。

3.1.4 解析、校验与重试

json_object 模式保证输出是合法 JSON，但不保证字段对。json_schema 模式更严格，但不是所有 provider 都支持。

所以你需要一层校验。用 Pydantic：

from pydantic import BaseModel, ValidationError
import openai
import json

class Spot(BaseModel):
    name: str
    fee: float

class DayPlan(BaseModel):
    day: int
    spots: list[Spot]
    budget: float

class Itinerary(BaseModel):
    days: list[DayPlan]

client = openai.OpenAI()

def get_itinerary(city: str, num_days: int, max_retries: int = 3) -> Itinerary:
    messages = [
        {
            "role": "system",
            "content": (
                "你是旅游规划助手。返回 JSON，schema：\n"
                '{"days": [{"day": 1, "spots": [{"name": "...", "fee": 0}], "budget": 0}]}'
            ),
        },
        {
            "role": "user",
            "content": f"{num_days}天{city}旅游行程",
        },
    ]

    for attempt in range(1, max_retries + 1):
        print(f"--- 尝试 {attempt} ---")

        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            response_format={"type": "json_object"},
        )
        raw = resp.choices[0].message.content
        print(f"模型原始输出：{raw[:200]}...")

        try:
            data = json.loads(raw)
            itinerary = Itinerary.model_validate(data)
            print(f"校验通过！共 {len(itinerary.days)} 天")
            return itinerary
        except (json.JSONDecodeError, ValidationError) as e:
            print(f"校验失败：{e}")
            # 把错误反馈给模型，让它修
            messages.append({"role": "assistant", "content": raw})
            messages.append({
                "role": "user",
                "content": f"上面的 JSON 校验失败：{e}\n请修正后重新输出。",
            })

    raise RuntimeError(f"{max_retries} 次尝试后仍然校验失败")

itinerary = get_itinerary("成都", 2)
for day in itinerary.days:
    print(f"第{day.day}天 预算{day.budget}元：", [s.name for s in day.spots])

运行日志（故意模拟一次失败）：

--- 尝试 1 ---
模型原始输出：{"days": [{"day": 1, "spots": [{"name": "武侯祠", "fee": 50}, {"name": "锦里古街", "fee": "免费"}], "budget": 50}, {"day": 2, ...
校验失败：1 validation error for Itinerary
days -> 0 -> spots -> 1 -> fee
  Input should be a valid number, unable to parse string as a number [type=float_parsing, input_value='免费', input_type=str]
--- 尝试 2 ---
模型原始输出：{"days": [{"day": 1, "spots": [{"name": "武侯祠", "fee": 50}, {"name": "锦里古街", "fee": 0}], "budget": 50}, {"day": 2, ...
校验通过！共 2 天
第1天 预算50元： ['武侯祠', '锦里古街']
第2天 预算180元： ['大熊猫繁育研究基地', '宽窄巷子']

关键设计：

  1. Pydantic 校验——不只是检查 JSON 合法，还检查类型、字段是否存在。
  2. 错误反馈——把 ValidationError 的具体信息塞回消息列表，模型看到错误后通常能修正。
  3. 有限重试——不能无限重试，3 次是合理上限。超过说明 prompt 或 schema 有问题。

这个 parse → validate → retry 的模式会在后面的章节反复出现。它是结构化输出的标配。

3.2 Provider 调研与抽象

3.2.1 四家 Provider 对比

你不会永远只用一家。有些场景需要便宜的模型处理简单任务，有些需要最强模型处理复杂推理，有些公司政策要求数据不出境。

我们看四家主流 Provider 的 API 差异：

认证方式

Provider 认证方式 环境变量
OpenAI Bearer token OPENAI_API_KEY
Anthropic x-api-key header ANTHROPIC_API_KEY
Google Gemini Bearer token（OAuth 或 API key） GOOGLE_API_KEY
DeepSeek Bearer token（OpenAI 兼容） DEEPSEEK_API_KEY

请求结构

OpenAI（Chat Completions）：

# POST https://api.openai.com/v1/chat/completions
{
    "model": "gpt-4o-mini",
    "messages": [
        {"role": "system", "content": "..."},
        {"role": "user", "content": "..."}
    ],
    "temperature": 0.7
}

OpenAI（Responses API——新版）：

# POST https://api.openai.com/v1/responses
{
    "model": "gpt-4o-mini",
    "input": "三天北京旅游行程",
    "instructions": "你是旅游规划助手",
    "temperature": 0.7
}

Anthropic：

# POST https://api.anthropic.com/v1/messages
{
    "model": "claude-sonnet-4-20250514",
    "max_tokens": 1024,
    "system": "你是旅游规划助手",
    "messages": [
        {"role": "user", "content": "三天北京旅游行程"}
    ]
}

Google Gemini：

# POST https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:generateContent
{
    "system_instruction": {
        "parts": [{"text": "你是旅游规划助手"}]
    },
    "contents": [
        {
            "role": "user",
            "parts": [{"text": "三天北京旅游行程"}]
        }
    ],
    "generationConfig": {
        "temperature": 0.7
    }
}

DeepSeek：

# POST https://api.deepseek.com/chat/completions
# 和 OpenAI Chat Completions 格式完全一样
{
    "model": "deepseek-chat",
    "messages": [
        {"role": "system", "content": "..."},
        {"role": "user", "content": "..."}
    ]
}

响应结构差异

字段 OpenAI Chat OpenAI Responses Anthropic Gemini
取文本 choices[0].message.content output[0].content[0].text content[0].text candidates[0].content.parts[0].text
token 用量 usage.prompt_tokens / completion_tokens usage.input_tokens / output_tokens usage.input_tokens / output_tokens usageMetadata.promptTokenCount / candidatesTokenCount
stop 原因 choices[0].finish_reason status stop_reason candidates[0].finishReason
system 消息 放在 messages 里 instructions 参数 独立 system 参数 system_instruction 参数
工具调用 tool_calls in message output 里的 function_call item content 里的 tool_use block functionCall in parts
JSON 模式 response_format 参数 text.format 参数 用 tool_choice 模拟 responseMimeType

价格与特点速查

Provider 代表模型 输入价格 ($/1M tokens) 输出价格 特点
OpenAI gpt-4o-mini ~0.15 ~0.60 生态最大，工具最多
Anthropic claude-sonnet-4-20250514 ~3.00 ~15.00 长上下文强，结构化输出稳
Google gemini-2.0-flash ~0.10 ~0.40 便宜，多模态好
DeepSeek deepseek-chat ~0.14 ~0.28 最便宜，OpenAI 兼容

价格经常变。以上只是写作时的量级参考，用之前查官网。

3.2.2 框架怎么做抽象

在自己写之前，先看看别人怎么做的。

LangChain：ChatModel 基类

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# 换 provider 只换类名
llm = ChatOpenAI(model="gpt-4o-mini")
# llm = ChatAnthropic(model="claude-sonnet-4-20250514")

resp = llm.invoke("三天北京行程")
print(resp.content)  # 统一用 .content

LangChain 用继承：BaseChatModel → ChatOpenAI / ChatAnthropic。好处是统一接口，坏处是抽象很重，调试时要穿好几层。

Pydantic AI：Model 接口

from pydantic_ai import Agent

agent = Agent("openai:gpt-4o-mini", system_prompt="你是旅游规划助手")
# agent = Agent("anthropic:claude-sonnet-4-20250514", system_prompt="...")

result = agent.run_sync("三天北京行程")
print(result.output)

Pydantic AI 用字符串选 provider（"openai:gpt-4o-mini"），内部做路由。更轻量。

Vercel AI SDK（TypeScript）

import { generateText } from "ai";
import { openai } from "@ai-sdk/openai";
// import { anthropic } from "@ai-sdk/anthropic";

const { text } = await generateText({
  model: openai("gpt-4o-mini"),
  // model: anthropic("claude-sonnet-4-20250514"),
  prompt: "三天北京行程",
});

Vercel AI SDK 用工厂函数（openai("gpt-4o-mini")），返回统一的 Model 对象。

共同点：都用 adapter/策略 模式——外部接口统一，内部按 provider 转换请求和响应。

3.2.3 设计统一接口

我们自己写一个最小的 Provider 抽象。设计目标：

  • 消息格式统一
  • 换 provider 只换一行
  • 类型安全（用 dataclass，不用 TypedDict）
  • 不超过 200 行
"""provider.py —— 最小 Provider 抽象层"""

from __future__ import annotations
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
import json
import os

# ---- 统一消息格式 ----

@dataclass
class Message:
    role: str          # "system" | "user" | "assistant" | "tool"
    content: str
    tool_call_id: str | None = None
    tool_calls: list[dict] | None = None

@dataclass
class Usage:
    input_tokens: int
    output_tokens: int

@dataclass
class CompletionResponse:
    content: str
    usage: Usage
    raw: dict = field(default_factory=dict)   # 保留原始响应，方便调试

# ---- 抽象基类 ----

class ModelProvider(ABC):
    """所有 provider 都实现这个接口"""

    @abstractmethod
    def complete(
        self,
        messages: list[Message],
        temperature: float = 0.7,
        response_format: dict | None = None,
    ) -> CompletionResponse:
        ...

# ---- OpenAI Adapter ----

class OpenAIProvider(ModelProvider):
    def __init__(self, model: str = "gpt-4o-mini"):
        import openai
        self.client = openai.OpenAI()
        self.model = model

    def complete(self, messages, temperature=0.7, response_format=None):
        kwargs = {
            "model": self.model,
            "messages": [self._to_openai_msg(m) for m in messages],
            "temperature": temperature,
        }
        if response_format:
            kwargs["response_format"] = response_format

        resp = self.client.chat.completions.create(**kwargs)
        choice = resp.choices[0]
        return CompletionResponse(
            content=choice.message.content or "",
            usage=Usage(
                input_tokens=resp.usage.prompt_tokens,
                output_tokens=resp.usage.completion_tokens,
            ),
            raw=resp.model_dump(),
        )

    @staticmethod
    def _to_openai_msg(m: Message) -> dict:
        d = {"role": m.role, "content": m.content}
        if m.tool_calls:
            d["tool_calls"] = m.tool_calls
        if m.tool_call_id:
            d["tool_call_id"] = m.tool_call_id
        return d

# ---- Anthropic Adapter ----

class AnthropicProvider(ModelProvider):
    def __init__(self, model: str = "claude-sonnet-4-20250514"):
        import anthropic
        self.client = anthropic.Anthropic()
        self.model = model

    def complete(self, messages, temperature=0.7, response_format=None):
        # Anthropic 把 system 消息单独提出来
        system_text = ""
        api_messages = []
        for m in messages:
            if m.role == "system":
                system_text += m.content + "\n"
            else:
                api_messages.append({"role": m.role, "content": m.content})

        kwargs = {
            "model": self.model,
            "max_tokens": 2048,
            "messages": api_messages,
            "temperature": temperature,
        }
        if system_text:
            kwargs["system"] = system_text.strip()

        resp = self.client.messages.create(**kwargs)
        text = resp.content[0].text if resp.content else ""
        return CompletionResponse(
            content=text,
            usage=Usage(
                input_tokens=resp.usage.input_tokens,
                output_tokens=resp.usage.output_tokens,
            ),
            raw=json.loads(resp.model_dump_json()),
        )

# ---- Gemini Adapter ----

class GeminiProvider(ModelProvider):
    def __init__(self, model: str = "gemini-2.0-flash"):
        import google.generativeai as genai
        genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
        self.genai = genai
        self.model_name = model

    def complete(self, messages, temperature=0.7, response_format=None):
        # 提取 system 消息
        system_text = ""
        contents = []
        for m in messages:
            if m.role == "system":
                system_text += m.content + "\n"
            else:
                role = "model" if m.role == "assistant" else "user"
                contents.append({"role": role, "parts": [{"text": m.content}]})

        model = self.genai.GenerativeModel(
            self.model_name,
            system_instruction=system_text.strip() or None,
            generation_config={"temperature": temperature},
        )
        resp = model.generate_content(contents)
        text = resp.text or ""
        usage = resp.usage_metadata
        return CompletionResponse(
            content=text,
            usage=Usage(
                input_tokens=getattr(usage, "prompt_token_count", 0),
                output_tokens=getattr(usage, "candidates_token_count", 0),
            ),
            raw={"text": text},
        )

# ---- DeepSeek Adapter（OpenAI 兼容）----

class DeepSeekProvider(ModelProvider):
    def __init__(self, model: str = "deepseek-chat"):
        import openai
        self.client = openai.OpenAI(
            api_key=os.environ["DEEPSEEK_API_KEY"],
            base_url="https://api.deepseek.com",
        )
        self.model = model

    def complete(self, messages, temperature=0.7, response_format=None):
        kwargs = {
            "model": self.model,
            "messages": [{"role": m.role, "content": m.content} for m in messages],
            "temperature": temperature,
        }
        if response_format:
            kwargs["response_format"] = response_format

        resp = self.client.chat.completions.create(**kwargs)
        choice = resp.choices[0]
        return CompletionResponse(
            content=choice.message.content or "",
            usage=Usage(
                input_tokens=resp.usage.prompt_tokens,
                output_tokens=resp.usage.completion_tokens,
            ),
            raw=resp.model_dump(),
        )

# ---- 工厂函数 ----

def get_provider(name: str, **kwargs) -> ModelProvider:
    """一行换 provider"""
    providers = {
        "openai": OpenAIProvider,
        "anthropic": AnthropicProvider,
        "gemini": GeminiProvider,
        "deepseek": DeepSeekProvider,
    }
    if name not in providers:
        raise ValueError(f"不支持的 provider: {name}，可选: {list(providers.keys())}")
    return providers[name](**kwargs)

用起来：

from provider import get_provider, Message

# 换 provider 只改这一行
provider = get_provider("openai", model="gpt-4o-mini")
# provider = get_provider("anthropic")
# provider = get_provider("deepseek")

messages = [
    Message(role="system", content="你是旅游规划助手"),
    Message(role="user", content="推荐一个周末去的地方"),
]

resp = provider.complete(messages, temperature=0.5)
print(resp.content)
print(f"tokens: {resp.usage.input_tokens} in / {resp.usage.output_tokens} out")

推荐去杭州！周末两天刚好：第一天逛西湖、灵隐寺，第二天去龙井村喝茶、爬北高峰。
高铁从上海出发1小时，非常方便。

tokens: 28 in / 67 out

测试

测试不需要真调 API。mock 掉 client 就行：

"""test_provider.py"""

from provider import (
    Message, OpenAIProvider, AnthropicProvider,
    DeepSeekProvider, get_provider, CompletionResponse
)
from unittest.mock import MagicMock, patch

def test_openai_adapter():
    provider = OpenAIProvider.__new__(OpenAIProvider)
    provider.model = "gpt-4o-mini"
    provider.client = MagicMock()

    # 构造 mock 响应
    mock_resp = MagicMock()
    mock_resp.choices = [MagicMock()]
    mock_resp.choices[0].message.content = '{"answer": "北京"}'
    mock_resp.usage.prompt_tokens = 10
    mock_resp.usage.completion_tokens = 5
    mock_resp.model_dump.return_value = {"mock": True}
    provider.client.chat.completions.create.return_value = mock_resp

    messages = [Message(role="user", content="推荐城市")]
    result = provider.complete(messages)

    assert result.content == '{"answer": "北京"}'
    assert result.usage.input_tokens == 10
    assert result.usage.output_tokens == 5

def test_factory():
    p = get_provider.__wrapped__ if hasattr(get_provider, '__wrapped__') else get_provider
    # 测试工厂返回正确类型
    with patch("provider.OpenAIProvider.__init__", return_value=None):
        provider = get_provider("openai")
        assert isinstance(provider, OpenAIProvider)

    try:
        get_provider("not_exist")
        assert False, "应该抛异常"
    except ValueError as e:
        assert "不支持" in str(e)

def test_message_fields():
    m = Message(role="user", content="hello")
    assert m.role == "user"
    assert m.tool_call_id is None

    m2 = Message(role="tool", content="result", tool_call_id="call_123")
    assert m2.tool_call_id == "call_123"

if __name__ == "__main__":
    test_openai_adapter()
    test_factory()
    test_message_fields()
    print("所有测试通过")

所有测试通过

3.2.4 Fallback 与多 Provider

生产环境不能只靠一家。API 会 500、会限流、会宕机。需要自动切换。

"""fallback.py —— 多 Provider Fallback"""

from provider import ModelProvider, Message, CompletionResponse, get_provider
import time

class FallbackProvider(ModelProvider):
    """按顺序尝试多个 provider，失败自动切换"""

    def __init__(self, configs: list[dict], max_retries_per: int = 2):
        """
        configs: [{"name": "openai", "model": "gpt-4o-mini"}, {"name": "deepseek"}, ...]
        """
        self.configs = configs
        self.max_retries_per = max_retries_per

    def complete(self, messages, temperature=0.7, response_format=None):
        errors = []

        for config in self.configs:
            provider_name = config["name"]
            kwargs = {k: v for k, v in config.items() if k != "name"}
            provider = get_provider(provider_name, **kwargs)

            for attempt in range(1, self.max_retries_per + 1):
                try:
                    print(f"[Fallback] 尝试 {provider_name} (第{attempt}次)")
                    resp = provider.complete(
                        messages,
                        temperature=temperature,
                        response_format=response_format,
                    )
                    print(f"[Fallback] {provider_name} 成功")
                    return resp
                except Exception as e:
                    wait = min(2 ** attempt, 8)
                    print(f"[Fallback] {provider_name} 失败: {e}，{wait}s 后重试")
                    errors.append(f"{provider_name}#{attempt}: {e}")
                    time.sleep(wait)

        raise RuntimeError(
            f"所有 provider 均失败:\n" + "\n".join(errors)
        )

# 用法
fallback = FallbackProvider([
    {"name": "openai", "model": "gpt-4o-mini"},
    {"name": "deepseek"},
    {"name": "anthropic"},
])

messages = [
    Message(role="system", content="你是旅游助手"),
    Message(role="user", content="杭州一日游"),
]

resp = fallback.complete(messages)
print(resp.content)

正常运行日志（第一个就成功）：

[Fallback] 尝试 openai (第1次)
[Fallback] openai 成功
西湖是杭州的必去之地...

模拟 OpenAI 挂了的日志：

[Fallback] 尝试 openai (第1次)
[Fallback] openai 失败: Connection error，2s 后重试
[Fallback] 尝试 openai (第2次)
[Fallback] openai 失败: Connection error，4s 后重试
[Fallback] 尝试 deepseek (第1次)
[Fallback] deepseek 成功
杭州一日游推荐路线：上午西湖断桥—白堤—孤山...

关键设计：

  1. 顺序优先级——列表靠前的先尝试。通常把最快最便宜的放前面。
  2. 指数退避——同一个 provider 失败后等 2s、4s、8s。避免瞬间打满限流。
  3. 每个 provider 有独立重试次数——不是全局 3 次就放弃，而是每家都有机会。
  4. 错误收集——最后全失败时，告诉你每家是怎么失败的。

3.3 对话历史

3.3.1 API 是无状态的

一个常见的新手误区：觉得给 OpenAI 发了三轮对话后，模型"记住"了之前的内容。

不是。每次 API 调用都是独立的。模型看到的只有你这次传过去的 messages 列表。服务端不存任何状态。

为什么不在服务端存？

  • 隐私——你不想 Provider 帮你记对话。
  • 灵活性——你可以自己决定删哪些、改哪些、摘要哪些。
  • 成本——服务端存状态意味着 Provider 要管更多基础设施，最终转嫁给你。
  • 确定性——你传什么，模型就看什么，没有隐式状态。

所以：多轮对话 = 你手动管一个消息列表，每次全量发过去。

3.3.2 消息列表管理

模式很简单：用户说一句 → append → 调 API → 把 assistant 回复 append → 用户再说一句 → ...

from provider import get_provider, Message

provider = get_provider("openai", model="gpt-4o-mini")

# 对话历史
history: list[Message] = [
    Message(role="system", content="你是旅游规划助手，回答简洁。"),
]

def chat(user_input: str) -> str:
    # 1. 追加用户消息
    history.append(Message(role="user", content=user_input))

    # 2. 发送完整历史
    resp = provider.complete(history)

    # 3. 追加助手回复
    history.append(Message(role="assistant", content=resp.content))

    return resp.content

# 模拟多轮对话
print("用户: 我想去日本")
print("助手:", chat("我想去日本"))
print()
print("用户: 推荐东京三天行程")
print("助手:", chat("推荐东京三天行程"))
print()
print("用户: 第二天想换成去镰仓")
print("助手:", chat("第二天想换成去镰仓"))
print()

# 看看历史里有什么
print(f"\n--- 对话历史共 {len(history)} 条消息 ---")
for m in history:
    print(f"[{m.role}] {m.content[:60]}...")

运行日志：

用户: 我想去日本
助手: 好的！日本是很棒的旅游目的地。你想去哪个城市？东京、京都、大阪还是其他地方？大概几天？

用户: 推荐东京三天行程
助手: 东京三天行程：
第一天：浅草寺→东京塔→银座
第二天：新宿御苑→涩谷→原宿
第三天：筑地市场→秋叶原→台场

用户: 第二天想换成去镰仓
助手: 好的，调整后：
第一天：浅草寺→东京塔→银座
第二天：镰仓大佛→鹤冈八幡宫→长谷寺→江之岛（从东京坐JR横须贺线约1小时）
第三天：筑地市场→秋叶原→台场

--- 对话历史共 7 条消息 ---
[system] 你是旅游规划助手，回答简洁。...
[user] 我想去日本...
[assistant] 好的！日本是很棒的旅游目的地。你想去哪个城市？东京、京都、大阪还是其他地方...
[user] 推荐东京三天行程...
[assistant] 东京三天行程：第一天：浅草寺→东京塔→银座 第二天：新宿御苑→涩谷→原...
[user] 第二天想换成去镰仓...
[assistant] 好的，调整后：第一天：浅草寺→东京塔→银座 第二天：镰仓大佛→鹤冈八幡...

注意第三轮：用户只说了"第二天想换成去镰仓"，模型知道"第二天"指的是什么、原来的行程是什么——因为完整历史都传过去了。

3.3.3 上下文窗口限制

问题来了：每次全量发历史，对话长了怎么办？

每个模型都有上下文窗口限制。gpt-4o-mini 是 128K tokens，claude-sonnet-4-20250514 是 200K tokens。听起来很大，但如果对话持续几十轮、每轮还带工具返回，很快就会撞墙。

三种处理策略：

策略一：截断（最简单）

只保留最近 N 轮，丢掉最早的消息。

def truncate_history(
    history: list[Message],
    max_messages: int = 20,
) -> list[Message]:
    """保留 system 消息 + 最近 max_messages 条"""
    system_msgs = [m for m in history if m.role == "system"]
    non_system = [m for m in history if m.role != "system"]

    if len(non_system) <= max_messages:
        return history

    kept = non_system[-max_messages:]
    print(f"[截断] 丢弃 {len(non_system) - max_messages} 条旧消息")
    return system_msgs + kept

# 示例
history = [Message(role="system", content="你是助手")]
for i in range(30):
    history.append(Message(role="user", content=f"第{i+1}个问题"))
    history.append(Message(role="assistant", content=f"第{i+1}个回答"))

truncated = truncate_history(history, max_messages=10)
print(f"截断前: {len(history)} 条，截断后: {len(truncated)} 条")
print(f"最早保留: [{truncated[1].role}] {truncated[1].content}")

[截断] 丢弃 50 条旧消息
截断前: 61 条，截断后: 11 条
最早保留: [user] 第26个问题

优点：实现简单，0 额外成本。缺点：早期上下文丢了，模型可能忘记用户最开始说的偏好。

策略二：摘要

把旧消息让模型压缩成一段摘要，作为 system 消息的一部分。

def summarize_history(
    provider: "ModelProvider",
    history: list[Message],
    keep_recent: int = 6,
) -> list[Message]:
    """把旧消息摘要成一段，和最近 keep_recent 条一起保留"""
    system_msgs = [m for m in history if m.role == "system"]
    non_system = [m for m in history if m.role != "system"]

    if len(non_system) <= keep_recent:
        return history

    old_msgs = non_system[:-keep_recent]
    recent_msgs = non_system[-keep_recent:]

    # 用模型做摘要
    old_text = "\n".join(f"[{m.role}] {m.content}" for m in old_msgs)
    summary_resp = provider.complete([
        Message(
            role="system",
            content="把以下对话历史压缩成一段简短摘要，保留关键信息（用户偏好、已确定的计划、重要约束）。",
        ),
        Message(role="user", content=old_text),
    ], temperature=0.3)

    summary_msg = Message(
        role="system",
        content=f"[对话摘要] {summary_resp.content}",
    )

    print(f"[摘要] 压缩了 {len(old_msgs)} 条旧消息")
    print(f"[摘要] 内容: {summary_resp.content[:100]}...")

    return system_msgs + [summary_msg] + recent_msgs

# 用法
provider = get_provider("openai", model="gpt-4o-mini")
compressed = summarize_history(provider, history, keep_recent=6)
print(f"压缩后: {len(compressed)} 条消息")

[摘要] 压缩了 54 条旧消息
[摘要] 内容: 用户进行了多轮问答测试，从第1个问题到第27个问题，均为测试性质的简单问答...
压缩后: 8 条消息

优点：保留了早期信息的精华。缺点：摘要本身要调一次 API，有成本；摘要可能丢失细节。

策略三：滑动窗口（按 token 数）

不按消息条数，而是按 token 数控制。更精确。

import tiktoken

def sliding_window(
    history: list[Message],
    max_tokens: int = 4000,
    model: str = "gpt-4o-mini",
) -> list[Message]:
    """按 token 数从后往前保留，确保不超限"""
    enc = tiktoken.encoding_for_model(model)

    system_msgs = [m for m in history if m.role == "system"]
    non_system = [m for m in history if m.role != "system"]

    # system 消息的 token 数先扣掉
    system_tokens = sum(len(enc.encode(m.content)) for m in system_msgs)
    remaining = max_tokens - system_tokens

    # 从后往前加，直到超限
    kept = []
    for m in reversed(non_system):
        msg_tokens = len(enc.encode(m.content)) + 4  # 4 tokens overhead per message
        if remaining - msg_tokens < 0:
            break
        kept.append(m)
        remaining -= msg_tokens

    kept.reverse()
    print(f"[滑动窗口] 保留 {len(kept)}/{len(non_system)} 条，"
          f"约 {max_tokens - remaining}/{max_tokens} tokens")
    return system_msgs + kept

[滑动窗口] 保留 42/60 条，约 3892/4000 tokens

实际怎么选

场景 推荐策略
短对话（<20 轮） 不处理，全量发
中等对话 截断，保留最近 20 条
长对话且早期有重要上下文 摘要 + 保留最近 N 条
token 预算严格 滑动窗口
生产环境 滑动窗口 + 摘要兜底

旅游助手的选择：大部分对话不超过 20 轮，截断就够了。如果用户会在一次对话里规划整个假期（几十轮），加摘要。

3.4 工具调用实践

3.2 写的 Provider 抽象目前只封装了文本生成。工具调用需要访问 response 里的 tool_calls 字段，CompletionResponse 还没暴露这些——这是后面要补的。所以这一节先用 OpenAI SDK 直接调。目标：一个旅游助手，能查天气、查景点、估算路线时间。

工具定义

"""tools.py —— 旅游助手的工具集"""

import json
import random

def get_weather(city: str, date: str) -> dict:
    """查询城市天气（模拟）"""
    weathers = ["晴", "多云", "小雨", "阴"]
    temps = {"春": (12, 22), "夏": (25, 35), "秋": (15, 25), "冬": (0, 10)}

    month = int(date.split("-")[1]) if "-" in date else 6
    if month in (3, 4, 5):
        season = "春"
    elif month in (6, 7, 8):
        season = "夏"
    elif month in (9, 10, 11):
        season = "秋"
    else:
        season = "冬"

    low, high = temps[season]
    return {
        "city": city,
        "date": date,
        "weather": random.choice(weathers),
        "temp_low": low + random.randint(-2, 2),
        "temp_high": high + random.randint(-2, 2),
        "suggestion": "适合户外活动" if season in ("春", "秋") else "注意防晒" if season == "夏" else "注意保暖",
    }

def search_places(city: str, category: str = "景点") -> list[dict]:
    """搜索城市景点/美食/住宿（模拟）"""
    data = {
        "北京": {
            "景点": [
                {"name": "故宫博物院", "rating": 4.8, "fee": 60, "hours": "08:30-17:00"},
                {"name": "八达岭长城", "rating": 4.7, "fee": 40, "hours": "06:30-19:00"},
                {"name": "颐和园", "rating": 4.6, "fee": 30, "hours": "06:30-18:00"},
                {"name": "天坛公园", "rating": 4.5, "fee": 15, "hours": "06:00-21:00"},
            ],
            "美食": [
                {"name": "全聚德烤鸭", "rating": 4.3, "price_per_person": 180},
                {"name": "护国寺小吃", "rating": 4.1, "price_per_person": 40},
            ],
        },
        "杭州": {
            "景点": [
                {"name": "西湖", "rating": 4.9, "fee": 0, "hours": "全天"},
                {"name": "灵隐寺", "rating": 4.6, "fee": 75, "hours": "07:00-18:00"},
                {"name": "宋城", "rating": 4.4, "fee": 280, "hours": "10:00-21:00"},
            ],
            "美食": [
                {"name": "楼外楼", "rating": 4.2, "price_per_person": 150},
                {"name": "知味观", "rating": 4.0, "price_per_person": 60},
            ],
        },
    }
    city_data = data.get(city, {})
    results = city_data.get(category, [])
    if not results:
        return [{"message": f"没有找到{city}的{category}信息"}]
    return results

def estimate_route(origin: str, destination: str, mode: str = "driving") -> dict:
    """估算两地之间的交通时间（模拟）"""
    # 简化：用字符数差异模拟距离
    base_km = abs(hash(origin + destination)) % 200 + 5
    speeds = {"driving": 60, "transit": 40, "walking": 5}
    speed = speeds.get(mode, 40)
    hours = round(base_km / speed, 1)
    return {
        "origin": origin,
        "destination": destination,
        "mode": mode,
        "distance_km": base_km,
        "duration_hours": hours,
        "duration_text": f"约{int(hours*60)}分钟" if hours < 1 else f"约{hours}小时",
    }

# 工具注册表：名字 → (函数, 描述, 参数schema)
TOOLS = {
    "get_weather": {
        "function": get_weather,
        "description": "查询指定城市和日期的天气预报",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名"},
                "date": {"type": "string", "description": "日期，格式 YYYY-MM-DD"},
            },
            "required": ["city", "date"],
        },
    },
    "search_places": {
        "function": search_places,
        "description": "搜索城市的景点、美食或住宿",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名"},
                "category": {
                    "type": "string",
                    "enum": ["景点", "美食", "住宿"],
                    "description": "搜索类别",
                },
            },
            "required": ["city"],
        },
    },
    "estimate_route": {
        "function": estimate_route,
        "description": "估算两地之间的交通时间和距离",
        "parameters": {
            "type": "object",
            "properties": {
                "origin": {"type": "string", "description": "出发地"},
                "destination": {"type": "string", "description": "目的地"},
                "mode": {
                    "type": "string",
                    "enum": ["driving", "transit", "walking"],
                    "description": "交通方式",
                },
            },
            "required": ["origin", "destination"],
        },
    },
}

完整旅游助手

把 Provider 抽象、对话历史、工具调用全部串起来：

"""travel_agent.py —— 完整旅游助手"""

import json
import openai
from tools import TOOLS

client = openai.OpenAI()

def run_travel_agent(user_query: str):
    """单轮工具调用：用户提问 → 模型选工具 → 执行 → 模型总结"""
    print(f"{'='*60}")
    print(f"用户: {user_query}")
    print(f"{'='*60}")

    # 构造 OpenAI tools 格式
    openai_tools = []
    for name, spec in TOOLS.items():
        openai_tools.append({
            "type": "function",
            "function": {
                "name": name,
                "description": spec["description"],
                "parameters": spec["parameters"],
            },
        })

    messages = [
        {
            "role": "system",
            "content": (
                "你是一个专业旅游规划助手。\n"
                "你可以查天气、搜索景点美食、估算交通路线。\n"
                "请先使用工具获取信息，然后基于工具返回给用户完整建议。"
            ),
        },
        {"role": "user", "content": user_query},
    ]

    # 第一次调用：让模型决定用哪些工具
    print("\n[Step 1] 发送请求，等待模型选择工具...")
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=openai_tools,
    )

    msg = resp.choices[0].message

    # 如果模型没调工具，直接返回文本
    if not msg.tool_calls:
        print("[结果] 模型直接回答（未调用工具）:")
        print(msg.content)
        return

    # 执行工具调用
    messages.append(msg.model_dump())  # 把 assistant 的 tool_calls 消息加回去

    print(f"\n[Step 2] 模型请求调用 {len(msg.tool_calls)} 个工具:")

    for tc in msg.tool_calls:
        func_name = tc.function.name
        args = json.loads(tc.function.arguments)
        print(f"  - {func_name}({args})")

        # 执行
        tool_func = TOOLS[func_name]["function"]
        result = tool_func(**args)
        result_str = json.dumps(result, ensure_ascii=False)
        print(f"    返回: {result_str[:120]}...")

        # 把工具结果加到消息列表
        messages.append({
            "role": "tool",
            "tool_call_id": tc.id,
            "content": result_str,
        })

    # 第二次调用：让模型基于工具结果总结
    print(f"\n[Step 3] 发送工具结果，等待模型总结...")
    resp2 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=openai_tools,
    )

    final = resp2.choices[0].message.content
    print(f"\n[最终回答]\n{final}")

    # 统计
    total_in = resp.usage.prompt_tokens + resp2.usage.prompt_tokens
    total_out = resp.usage.completion_tokens + resp2.usage.completion_tokens
    print(f"\n[统计] 共 2 次 API 调用，{total_in} input + {total_out} output tokens")

# ---- 运行 ----

run_travel_agent("我下周想去北京玩三天，帮我看看天气和推荐景点")

完整运行日志

============================================================
用户: 我下周想去北京玩三天，帮我看看天气和推荐景点
============================================================

[Step 1] 发送请求，等待模型选择工具...

[Step 2] 模型请求调用 3 个工具:
  - get_weather({"city": "北京", "date": "2026-05-18"})
    返回: {"city": "北京", "date": "2026-05-18", "weather": "晴", "temp_low": 14, "temp_high": 24, "suggestion": "适合户外活动"}...
  - get_weather({"city": "北京", "date": "2026-05-19"})
    返回: {"city": "北京", "date": "2026-05-19", "weather": "多云", "temp_low": 13, "temp_high": 23, "suggestion": "适合户外活动"}...
  - search_places({"city": "北京", "category": "景点"})
    返回: [{"name": "故宫博物院", "rating": 4.8, "fee": 60, "hours": "08:30-17:00"}, {"name": "八达岭长城", "rating": 4.7, "fee": 40...

[Step 3] 发送工具结果，等待模型总结...

[最终回答]
下周北京的天气很适合出游！以下是我的建议：

**天气情况**
- 5月18日：晴，14-24°C，适合户外活动
- 5月19日：多云，13-23°C，适合户外活动

**三天行程推荐**

第一天：故宫 + 天坛
- 上午：故宫博物院（评分4.8，门票60元，08:30-17:00）
- 下午：天坛公园（评分4.5，门票15元，06:00-21:00）
- 预算：约75元门票 + 餐饮

第二天：长城一日
- 全天：八达岭长城（评分4.7，门票40元，06:30-19:00）
- 建议早出发，长城需要大半天时间
- 预算：约40元门票 + 交通

第三天：颐和园 + 市区
- 上午：颐和园（评分4.6，门票30元，06:30-18:00）
- 下午：自由活动，推荐南锣鼓巷或什刹海
- 预算：约30元门票 + 餐饮

**温馨提示**：天气适合户外，但昼夜温差较大，建议带件薄外套。

[统计] 共 2 次 API 调用，847 input + 312 output tokens

再试一个带路线估算的：

run_travel_agent("从杭州西湖到灵隐寺怎么走？打车要多久？")

============================================================
用户: 从杭州西湖到灵隐寺怎么走？打车要多久？
============================================================

[Step 1] 发送请求，等待模型选择工具...

[Step 2] 模型请求调用 1 个工具:
  - estimate_route({"origin": "杭州西湖", "destination": "灵隐寺", "mode": "driving"})
    返回: {"origin": "杭州西湖", "destination": "灵隐寺", "mode": "driving", "distance_km": 7, "duration_hours": 0.1, "duration_te...

[Step 3] 发送工具结果，等待模型总结...

[最终回答]
从西湖到灵隐寺打车大约7公里，约6分钟车程。

不过实际时间取决于交通状况，旺季和周末西湖周边容易堵车，建议预留15-20分钟。也可以考虑坐公交Y2路，从断桥出发直达灵隐寺，虽然慢一些但不用担心堵车。

[统计] 共 2 次 API 调用，423 input + 89 output tokens

整体架构回顾

这一章做了什么：

                    用户
                      |
                      v
              ┌──────────────┐
              │  对话历史管理   │  3.3 — 消息列表、截断、摘要
              └──────┬───────┘
                     |
                     v
              ┌──────────────┐
              │ Provider 抽象  │  3.2 — 统一接口、Fallback
              └──────┬───────┘
                     |
                     v
        ┌────────────┴───────────┐
        |                        |
  ┌─────v─────┐          ┌──────v──────┐
  │ 结构化输出  │          │  工具调用     │
  │  JSON/校验  │          │  天气/景点/路线 │
  └───────────┘          └─────────────┘
      3.1                     3.4

这四块合在一起，就是 One-shot 域的全部能力。你的旅游助手现在能：

  • 输出结构化 JSON（不再靠正则刮散文）
  • 用任何 Provider（不绑死 OpenAI）
  • 管多轮对话（不会忘记用户说了什么）
  • 调工具拿实时信息（不再靠模型瞎编天气）

但它还是 One-shot 的：用户问一句，助手查工具、回一句。如果任务需要多步规划（先查天气 → 根据天气选景点 → 根据景点排路线 → 估预算），目前要用户自己一步步问。

这就是下一章的事了——第四章：Prompt 与 Context 工程。