一、LLM API 的本质：不是函数调用，是对话续写

Java 开发者第一次看 LLM API 文档，最困惑的是：为什么输入不是 String prompt，而是一个 messages 数组？为什么调同一个接口，有时候 3 秒返回、有时候 30 秒？为什么流式输出和普通调用是两套完全不同的体验？

理解这些问题的关键，是认识到 LLM API 和传统 REST API 的根本差异。不是"接口风格不同"，而是底层处理机制完全不同。REST API 是"接收输入 -> 执行逻辑 -> 返回结果"，LLM API 是"接收对话历史 -> 预测下一个 token -> 逐字返回"。这个差异决定了从参数设计到错误处理的每一个工程决策。

1. messages 数组：会话状态的序列化

# 传统思维：一个请求包含一个输入
response = llm.chat("帮我写一个订单确认邮件")

# LLM 实际模型：一个请求包含完整对话历史
response = llm.chat(messages=[
    {"role": "system", "content": "你是客服助手，只处理订单相关问题"},
    {"role": "user", "content": "帮我写一个订单确认邮件"},
    {"role": "assistant", "content": "好的，请问订单号是多少？"},
    {"role": "user", "content": "订单号 #20250701-001"}
])
# 输出: 尊敬的客户，您的订单 #20250701-001 已确认...

设计原理： LLM 的底层能力是"续写对话"。messages 数组完整记录了对话上下文–system 定义角色和规则，user 是用户输入，assistant 是模型之前的回复。这种设计让模型能在多轮对话中保持上下文一致性，而不是每次调用都从零开始。

Java 对照： 这和 REST API 的无状态设计不同。REST API 每次请求自包含所有信息，而 LLM API 的 messages 数组本质上是"把会话状态塞进请求体"。你可以理解为–每次调用都把整个 HttpSession 序列化后发过去。这意味着每次请求的 payload 会随着对话轮数增长，直接影响延迟和成本。

2. 流式输出的工程价值

# 非流式：等 5 秒，一次性拿到全部结果
response = client.chat.completions.create(
    model="your-model",
    messages=[{"role": "user", "content": "写一首关于编程的诗"}]
)
print(response.choices[0].message.content)
# 输出: 代码如诗，逻辑为韵...

# 流式：0.5 秒出第一个字，之后逐字流出
stream = client.chat.completions.create(
    model="your-model",
    messages=[{"role": "user", "content": "写一首关于编程的诗"}],
    stream=True
)
for chunk in stream:
    print(chunk.choices[0].delta.content or "", end="")
# 输出: 代(0.3s)码(0.4s)如(0.5s)诗(0.6s)...

性能对比：

模式	首字延迟	完整耗时	用户体验	适用场景
非流式	3-8s	3-8s	等待焦虑	后台批处理、数据抽取
流式	0.3-1s	3-8s	即时反馈	面向用户的聊天界面

工程分析： 流式输出不是可选项，而是生产环境的必修课。用户对首字延迟极其敏感–等 3 秒才开始输出，体验远差于 0.5 秒出第一个字、之后逐字流出。Spring AI 中对应的是 stream() 方法返回 Flux<String>，和 WebFlux 的响应式编程模型完美契合。但要注意流式模式下错误处理更复杂：如果第 5 个 chunk 时网络断了，已经返回的内容无法收回，需要设计幂等补偿机制。

二、5 分钟调通第一个 LLM API

1. 环境准备与最小可用代码

安装 SDK 只需要一行：

pip install openai
# 输出: Successfully installed openai-1.x.x

发送第一个请求：

from openai import OpenAI

client = OpenAI(api_key="your-api-key")
resp = client.chat.completions.create(
    model="your-model",
    messages=[{"role": "user", "content": "用一句话解释什么是RAG"}],
    temperature=0
)
print(resp.choices[0].message.content)
# 输出: RAG是一种结合检索和生成的技术，先从知识库找到相关文档，再让大模型基于这些文档生成回答。

就这些。 没有 Spring Boot 的启动类、没有 @Configuration、没有 Bean 注入。Python 的 AI 开发门槛就是这么低。但"能调通"和"能用在生产环境"之间差了十万八千里，后面的章节会逐步展开。

2. 请求结构详解

很多教程直接贴 JSON 原文，其实理解结构比看原始 JSON 更重要：

层级	字段	类型	说明	必填
顶层	model	String	模型名称	是
顶层	messages	Array	对话数组	是
顶层	temperature	Float	随机性	否
顶层	max_tokens	Int	最大输出	否
顶层	stream	Bool	流式返回	否
顶层	top_p	Float	核采样	否
messages[]	role	String	system/user/assistant	是
messages[]	content	String	消息文本	是

2. 响应结构详解

层级	字段	说明	工程用途
顶层	id	请求 ID	日志追踪、工单排查
顶层	choices	回复列表	取回复内容
choices[]	message.content	回复文本	核心输出
choices[]	finish_reason	结束原因	判断是否截断
顶层	usage.prompt_tokens	输入 token 数	成本核算
顶层	usage.completion_tokens	输出 token 数	成本核算
顶层	usage.total_tokens	总 token 数	监控告警

关键字段说明： finish_reason 有三个值需要区分：

值	含义	工程处理
stop	正常结束	直接使用输出
length	被 max_tokens 截断	需要 continue 续写
content_filter	内容被安全过滤	记录日志并降级

Java 对照： 这和调用一个 RESTful API 没有本质区别。OpenAI SDK 底层就是 HTTP 客户端，帮你处理了序列化、重试、流式解析。Java 中用 OkHttp 或 WebClient 调用是一样的，只是代码多一些。

三、平台选型思路：云服务还是本地部署

1. 两种部署模式的本质差异

调通 API 之后，你会面临第一个工程决策：用云服务还是本地部署？这不是一个技术偏好问题，而是由业务约束驱动的工程决策。

维度	云服务 API	本地部署
初始成本	按量付费，近零门槛	GPU 硬件投入高
运维复杂度	平台托管，零运维	模型加载、显存管理、版本升级
延迟	网络往返 200-800ms	本地推理 50-200ms
并发能力	平台弹性扩容	受限于本地 GPU 数量
数据隐私	数据需发送到服务端	数据不出本地
模型选择	平台提供的模型列表	任意开源模型
可用性	依赖平台 SLA	依赖本地硬件稳定性
升级频率	平台自动更新	需手动下载新模型

选型原则： 生产环境根据数据隐私要求和延迟要求决定。对数据隐私要求高的场景（医疗、金融内部系统），本地部署是硬性约束；对响应速度要求高的场景，云服务的弹性扩容更可靠。

2. OpenAI 兼容协议：为什么切换成本这么低

一个好消息是，主流 LLM 平台和本地推理框架几乎都支持 OpenAI 兼容协议。这意味着你的业务代码只需要改一行配置就能切换平台：

# 云服务
client = OpenAI(api_key="your-key", base_url="https://api.example.com/v1")

# 本地部署（Ollama / vLLM 等推理框架）
client = OpenAI(api_key="not-needed", base_url="http://localhost:11434/v1")

# 后续调用代码完全一样，零改动切换
response = client.chat.completions.create(
    model="your-model",
    messages=[{"role": "user", "content": "你好"}]
)
print(response.choices[0].message.content)
# 输出: 你好！有什么可以帮助你的吗？

工程分析： 这就是 OpenAI 兼容协议的价值–生态锁定被打破了。你的业务代码不需要为每个平台写适配层。实际项目中建议把 base_url 和 model 放到配置文件里，通过环境变量切换，做到一套代码跑通开发、测试、生产三个环境。

3. 配置管理最佳实践

# config.py - 统一配置管理
import os

class LLMConfig:
    def __init__(self):
        self.base_url = os.getenv("LLM_BASE_URL", "http://localhost:11434/v1")
        self.api_key = os.getenv("LLM_API_KEY", "not-needed")
        self.model = os.getenv("LLM_MODEL", "your-model")
        self.temperature = float(os.getenv("LLM_TEMPERATURE", "0.7"))
        self.max_tokens = int(os.getenv("LLM_MAX_TOKENS", "1000"))
        self.timeout = int(os.getenv("LLM_TIMEOUT", "30"))

config = LLMConfig()
client = OpenAI(api_key=config.api_key, base_url=config.base_url)
print(f"连接到: {config.base_url}, 模型: {config.model}")
# 输出: 连接到: http://localhost:11434/v1, 模型: your-model

Java 对照： 这和 Java 项目中用 Spring Profile 管理不同环境的数据库连接是一个思路。application-dev.yml 指向本地模型，application-prod.yml 指向云服务 API，代码里注入的是同一个 ChatClient 接口，不关心底层用的是哪个平台。

四、Java 版核心代码

Java 开发者如果暂时不想碰 Python，用 Java 也能调通：

// Spring AI 方式（第07篇详讲）
// pom.xml 添加 spring-ai-openai-spring-boot-starter

// application.yml 配置
// spring.ai.openai.api-key=${LLM_API_KEY}
// spring.ai.openai.base-url=${LLM_BASE_URL}
// spring.ai.openai.chat.options.model=${LLM_MODEL}

ChatClient client = ChatClient.create(model);
String answer = client.prompt()
    .user("用一句话解释什么是RAG")
    .call().content();
System.out.println(answer);
// 输出: 和 Python 版结果一致

Java 对照： Python 版用了 7 行代码完成的事，Spring AI 压缩到了 3 行。但 Spring AI 的配置背后还有一套自动装配逻辑。Python 的显式调用在调试时更直观，Java 的声明式配置在生产环境更规范。两种方式各有优势，建议先 Python 理解原理，再 Java 落地工程。

五、多轮对话的工程实现

单轮调用只是起点。真实业务场景中，用户通常会连续提问，需要 AI 保持上下文。多轮对话是 LLM 应用最常见也最容易出错的场景。

1. 最简多轮对话

from openai import OpenAI

client = OpenAI(api_key="your-key")
messages = [
    {"role": "system", "content": "你是一个订单管理系统的客服助手"}
]

# 第一轮
messages.append({"role": "user", "content": "查询订单 #20250701-001 的状态"})
response = client.chat.completions.create(
    model="your-model", messages=messages, temperature=0
)
ai_reply = response.choices[0].message.content
messages.append({"role": "assistant", "content": ai_reply})
print(f"AI: {ai_reply}")
# 输出: AI: 订单 #20250701-001 当前状态为已发货，预计明天到达。

# 第二轮 - AI 能记住上一轮的对话
messages.append({"role": "user", "content": "那大概几点能到？"})
response = client.chat.completions.create(
    model="your-model", messages=messages, temperature=0
)
print(f"AI: {response.choices[0].message.content}")
# 输出: AI: 根据物流信息，预计明天下午 2-5 点之间送达。

2. 多轮对话的状态管理

问题	症状	原因	解决方案
上下文丢失	AI 忘记之前说的内容	messages 未带历史	每轮都带完整 messages
token 超限	报错 context_length_exceeded	消息累积过多	截断旧消息
成本递增	每轮都比上轮贵	输入 token 线性增长	控制保留轮数
角色混乱	system 提示被覆盖	system 位置不对	system 始终在第一行
响应变慢	第 10 轮明显比第 1 轮慢	输入变长导致推理变慢	限制上下文长度

工程分析： 每多一轮对话，messages 数组就多两条消息（user + assistant），输入 token 线性增长。到第 20 轮时，输入可能是第一轮的 10 倍。生产环境必须实现"滑动窗口"策略：保留 system 消息 + 最近 N 轮对话，丢弃更早的。

3. 滑动窗口实现

from collections import deque

class ChatSession:
    def __init__(self, system_prompt, max_history=10):
        self.system = {"role": "system", "content": system_prompt}
        self.history = deque(maxlen=max_history * 2)  # 每轮 2 条消息

    def add(self, role, content):
        self.history.append({"role": role, "content": content})

    def get_messages(self):
        return [self.system] + list(self.history)

# 使用
session = ChatSession("你是客服助手", max_history=5)
session.add("user", "订单 #001 到哪了？")
session.add("assistant", "已发货，明天到。")
session.add("user", "可以改地址吗？")
# 第 6 轮后，最早的消息自动被 deque 丢弃

Java 对照： 这和 Java 中用 LinkedHashMap 实现 LRU 缓存是同一个思路–固定容量，超出就淘汰最旧的。区别在于 LLM 场景下淘汰的是对话历史，不是缓存条目，需要考虑"被淘汰的消息是否包含关键信息"的问题。后续篇章会讲用 RAG 解决这个问题。

六、错误处理与重试策略

1. 常见报错速查

报错信息	HTTP 状态码	原因	解决方案
Unauthorized	401	API Key 错误或过期	检查环境变量是否正确加载
Rate Limit	429	请求频率超限	加重试退避（exponential backoff）
context_length_exceeded	400	输入 token 超限	截断历史消息或换更大窗口模型
Connection refused	-	网络不通/服务未启动	本地部署检查服务是否运行
model_not_found	404	模型名拼写错误	核对平台文档的模型 ID
Internal Server Error	500	平台内部错误	重试，持续报错切备用平台
Empty response	200	max_tokens 设太小	调大 max_tokens

2. 生产级重试封装

import time
from openai import OpenAI, RateLimitError, APIConnectionError

client = OpenAI(api_key="your-key")

def call_llm_with_retry(messages, max_retries=3):
    """带指数退避的重试封装"""
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="your-model",
                messages=messages,
                temperature=0.7,
                timeout=30
            )
            return response.choices[0].message.content
        except RateLimitError:
            wait = 2 ** attempt
            print(f"限流，{wait}s 后重试 ({attempt+1}/{max_retries})")
            time.sleep(wait)
        except APIConnectionError as e:
            print(f"连接失败: {e}")
            time.sleep(1)
    raise Exception(f"重试 {max_retries} 次后仍失败")

重试策略对比：

策略	实现	优点	缺点
固定间隔	sleep(2)	简单	可能加重限流
指数退避	2^attempt	自适应	首次等待可能过长
指数+抖动	2^attempt + random	分散重试	实现稍复杂
令牌桶	限流器	主动控制	需额外组件

Java 对照： 这些报错和 Java 中的 HttpClient 异常没有本质区别。401 相当于认证失败，429 相当于限流。Spring Retry 框架提供了 @Retryable 注解实现同样的指数退避逻辑，配置式而非编程式。

七、API 调用验证检查清单

基础验证

检查项	验证方式	通过标准
API Key 有效性	发送测试请求	返回 200 + 有效内容
网络连通性	curl 基础接口	响应时间 < 3s
模型可用性	指定 model 参数	返回非空 content
Token 消耗记录	检查 usage 字段	input/output token 合理
错误处理	模拟无效 Key	优雅降级而非崩溃
超时设置	设置 timeout 参数	30s 内超时返回
流式输出	stream=True 测试	逐 token 返回

生产环境额外检查

检查项	说明	验证方式
重试机制	429 错误时指数退避	模拟限流测试
日志脱敏	API Key 不出现在日志	grep 日志文件
成本监控	按 usage 统计每日消耗	设置预算告警
降级方案	主平台不可用时切备用	模拟平台故障
并发控制	控制请求并发数	压力测试
优雅关闭	等待进行中的请求完成	发送 SIGTERM
输出校验	检查返回内容格式	JSON schema 验证

八、踩坑总结

坑点	症状	原因	解决方案
API Key 写在代码里	代码提交后泄露	硬编码	用 .env + python-dotenv
不处理超时	请求挂起无限等待	默认无超时	设置 timeout=30
忽略 finish_reason	输出被截断但不知道	max_tokens 太小	检查是否为 length
messages 顺序错误	模型回答不知所云	system 位置不对	system 在前，user 在后
不记录 token 消耗	月底账单惊吓	缺少监控	每次调用记录 usage
流式不处理空 chunk	程序报错 NoneType	delta.content 为 None	用 or "" 兜底
多轮对话不做截断	突然报错超限	消息累积超 context	实现滑动窗口
system 消息被截断	AI 角色混乱	截断时误删 system	截断只针对 user/assistant

最高频坑点详解： 多轮对话不做截断是最常见的生产事故。假设每轮对话平均 300 token，第 15 轮时输入就达到 4500 token，第 50 轮就可能超过小窗口模型的限制。解决方案是在每次调用前检查 messages 总 token 数，超过阈值的 80% 就开始从最早的非 system 消息截断。注意：system 消息永远不能被截断，否则 AI 的行为会完全失控。

九、完整示例：封装一个生产可用的 LLM 客户端

把前面所有知识点整合起来，封装一个实际可用的 LLM 客户端类。这个类包含配置管理、多轮对话、滑动窗口、重试机制和成本统计：

import os
import time
import tiktoken
from openai import OpenAI, RateLimitError, APIConnectionError
from collections import deque

class ProductionLLMClient:
    """生产可用的 LLM 客户端封装"""

    def __init__(self, config=None):
        self.config = config or self._load_config()
        self.client = OpenAI(
            api_key=self.config["api_key"],
            base_url=self.config["base_url"]
        )
        self.encoding = tiktoken.get_encoding("cl100k_base")
        self.total_tokens = 0
        self.total_calls = 0

    def _load_config(self):
        return {
            "base_url": os.getenv("LLM_BASE_URL", "http://localhost:11434/v1"),
            "api_key": os.getenv("LLM_API_KEY", "not-needed"),
            "model": os.getenv("LLM_MODEL", "your-model"),
            "temperature": float(os.getenv("LLM_TEMP", "0.7")),
            "max_tokens": int(os.getenv("LLM_MAX_TOKENS", "1000")),
            "timeout": int(os.getenv("LLM_TIMEOUT", "30")),
        }

    def count_tokens(self, text):
        return len(self.encoding.encode(text))

    def chat(self, messages, temperature=None):
        """带重试的聊天调用"""
        temp = temperature if temperature is not None else self.config["temperature"]
        for attempt in range(3):
            try:
                resp = self.client.chat.completions.create(
                    model=self.config["model"],
                    messages=messages,
                    temperature=temp,
                    max_tokens=self.config["max_tokens"],
                    timeout=self.config["timeout"]
                )
                self.total_tokens += resp.usage.total_tokens
                self.total_calls += 1
                return resp.choices[0].message.content
            except RateLimitError:
                time.sleep(2 ** attempt)
            except APIConnectionError:
                time.sleep(1)
        raise Exception("LLM 调用失败，重试耗尽")

    def get_stats(self):
        return {"calls": self.total_calls, "tokens": self.total_tokens}

# 使用示例
client = ProductionLLMClient()
reply = client.chat([
    {"role": "system", "content": "你是 Java 技术专家"},
    {"role": "user", "content": "Spring Boot 启动流程是什么？"}
])
print(reply)
# 输出: Spring Boot 启动流程主要包括...

print(client.get_stats())
# 输出: {'calls': 1, 'tokens': 850}

封装要点总结：

封装层	职责	关键点
配置管理	环境变量加载	12-Factor 规范
重试机制	指数退避	最多 3 次
Token 统计	成本监控	记录 usage
超时控制	避免无限等待	默认 30s
编码器	Token 计算	cl100k_base

Java 对照： 这个封装对应 Java 中的一个 @Service 类，注入 ChatClient 和配置，对外暴露 chat() 方法。Spring AI 的 ChatClient 已经内置了重试和超时配置，但 Token 统计需要自己实现。Spring Boot Actuator 的自定义 metrics 可以用来暴露 Token 消耗到 Prometheus + Grafana 监控面板。

十、测试验证：确保 API 调用可靠性

LLM API 的测试和传统 API 测试不同：输出是不确定的（temperature > 0 时），你不能断言返回的具体内容。测试策略需要调整为验证"结构"和"行为"而非"内容"。

1. 测试维度划分

测试类型	验证目标	通过标准	工具
连通性测试	API 可达	返回 200	pytest
格式测试	响应结构正确	choices[0] 非空	pytest
参数测试	temperature 生效	t=0 两次输出一致	pytest
边界测试	超长输入处理	优雅报错而非崩溃	pytest
性能测试	响应时间	P95 < 5s	locust
成本测试	Token 消耗合理	单次 < 预算阈值	手动

2. 基础测试代码

import pytest
from your_module import ProductionLLMClient

@pytest.fixture
def client():
    return ProductionLLMClient()

def test_api_connectivity(client):
    """测试 API 连通性"""
    reply = client.chat([{"role": "user", "content": "hello"}])
    assert reply is not None
    assert len(reply) > 0

def test_temperature_zero_reproducibility(client):
    """temperature=0 时两次输出应一致"""
    msg = [{"role": "user", "content": "1+1="}]
    r1 = client.chat(msg, temperature=0)
    r2 = client.chat(msg, temperature=0)
    assert r1 == r2

def test_token_counting(client):
    """Token 统计功能正常"""
    tokens = client.count_tokens("你好世界")
    assert tokens > 0
    assert tokens < 10

测试输出示例：

$ pytest test_llm.py -v
test_api_connectivity PASSED
test_temperature_zero_reproducibility PASSED
test_token_counting PASSED
3 passed in 8.42s

Java 对照： 这和 Spring Boot Test 中用 @SpringBootTest 注入 Service 做集成测试是一个思路。区别在于 LLM 测试需要真实 API 调用（无法 Mock 模型行为），建议在 CI 中标记为 @Tag("integration") 单独运行，不阻塞主流程。

十一、从 API 调用到生产应用的路线图

阶段	目标	关键能力	对应篇章
调通 API	能发送请求拿到回复	SDK 使用、参数理解	本文（第03篇）
Prompt 工程	稳定控制输出格式	结构化提示词、变量插值	第05篇
函数调用	让 AI 调用外部工具	Function Calling	第06篇
RAG 检索	基于知识库回答	Embedding + 向量检索	第10-11篇
Agent 架构	多步骤自主执行	ReAct、Plan-Execute	第19篇
生产部署	稳定可靠上线	限流、降级、监控	第26篇

工程分析： 从调通 API 到生产部署，技术跨度远大于从 Hello World 到 CRUD。LLM 应用开发不是"调个接口"那么简单，而是涉及提示词工程、检索增强、工具调用、上下文管理、成本控制、安全防护等多个工程维度。每一步都有坑，每个坑都可能让线上系统出问题。后续篇章会逐个深入这些主题。

下一篇预告

下一篇我们进入 LLM 应用开发的核心能力–Prompt Engineering（提示词工程）。不是教你写"你是一个专业的 XX"这种模板，而是讲透提示词的结构化设计、变量插值、以及如何用 Java 开发者的工程思维来管理提示词版本。

---

这是《Java 程序员的 AI 进阶之路》系列第 03 篇。调通 API 只是起点，真正的挑战在后面。