模型接入的第一版通常很短：创建客户端、拼一段消息、打印返回文本。它能验证凭据和网络，却很难直接进入工具链。原因并不在于调用代码不够复杂，而在于几个工程问题尚未被明确：端点是否可替换、密钥如何注入、长响应如何持续呈现、瞬时失败如何处理、日志如何关联一次任务，以及模型输出异常时业务该如何降级。
这些问题在命令行问答、代码生成辅助、批处理摘要和内部工作流中都会出现。尤其当团队需要切换模型、切换接入方，或分别使用开发与生产环境时，将地址、模型名和调用参数散落在业务代码中，会使调整成本迅速升高。
本文采用一个小型但完整的 Python 客户端作为例子。它不假设某个服务必然支持某种模型、流式协议或参数；实际接入前，应以目标服务当前文档为准。HaerAPI（https://www.haerapi.com）可作为需要评估模型接入渠道时的一个候选入口，但接入实现仍应围绕可配置性、审计与故障处理设计。
先建立稳定的调用边界
一个可维护的调用层至少应把四类变化隔离开：
- 连接配置：Base URL、API Key、超时和模型名来自部署环境，而不是源代码。
- 请求语义：系统提示、用户输入、温度等由业务层传入，但需做长度和类型校验。
- 传输策略：重试只处理可能恢复的传输或服务端故障，不应重放所有请求。
- 结果语义：调用层返回文本或事件；业务层决定是否保存、展示、执行或要求人工确认。
这种划分的关键是避免“模型客户端就是业务逻辑”。例如，生成 SQL、修改配置、调用外部工具等高风险动作，不能仅因模型返回了一段看似正确的文本就自动执行。调用层负责取得结果和保留证据，业务层负责权限、校验与提交。
配置：让密钥留在运行环境中
下面使用环境变量保存配置。示例中的变量名只是应用约定，不代表任何平台的固定字段。若所选服务采用 OpenAI 兼容接口，可将其兼容地址填入 LLM_BASE_URL；若接口格式不同，应保留本文的配置与错误处理思路，并按文档替换请求实现。
export LLM_API_KEY='replace-at-runtime'
export LLM_BASE_URL='https://your-endpoint.example/v1'
export LLM_MODEL='your-model-id'
export LLM_TIMEOUT_SECONDS='45'
在容器或 CI 中，应通过平台的 Secret 注入机制提供这些变量。不要把 .env、终端历史、完整请求头或密钥写入仓库和日志。开发环境若使用 .env 文件，也应将其加入忽略规则，并提供不含真实值的 .env.example。
可以先用一个配置对象集中校验必填项：
from dataclasses import dataclass
import os
@dataclass(frozen=True)
class Settings:
api_key: str
base_url: str
model: str
timeout_seconds: float
@classmethod
def from_env(cls) -> "Settings":
api_key = os.environ.get("LLM_API_KEY")
base_url = os.environ.get("LLM_BASE_URL")
model = os.environ.get("LLM_MODEL")
if not all([api_key, base_url, model]):
raise RuntimeError("LLM_API_KEY、LLM_BASE_URL、LLM_MODEL 必须配置")
return cls(
api_key=api_key,
base_url=base_url.rstrip("/"),
model=model,
timeout_seconds=float(os.environ.get("LLM_TIMEOUT_SECONDS", "45")),
)
实现：流式输出不等于跳过错误处理
流式响应适合交互式命令行，因为用户可以较早看到结果。但流的中途断开意味着“已经得到部分文本”，调用方必须明确这个部分文本能否使用。对于摘要展示，部分结果或许仍有价值；对于结构化指令、文件修改或自动化执行，则应视为失败并停止后续动作。
以下示例使用 openai Python SDK 的常见兼容调用形式。安装包版本、构造参数和流事件字段可能随 SDK 演进而变化，生产使用前请锁定已验证的依赖版本，并核对服务端兼容说明。
python -m pip install openai
import logging
import time
import uuid
from openai import APIConnectionError, APIStatusError, OpenAI, RateLimitError
from settings import Settings
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
RETRYABLE_STATUS = {
408, 409, 429, 500, 502, 503, 504}
def is_retryable(error: Exception) -> bool:
if isinstance(error, (APIConnectionError, RateLimitError)):
return True
return isinstance(error, APIStatusError) and error.status_code in RETRYABLE_STATUS
def stream_answer(user_text: str) -> str:
if not user_text.strip():
raise ValueError("输入不能为空")
cfg = Settings.from_env()
client = OpenAI(api_key=cfg.api_key, base_url=cfg.base_url, timeout=cfg.timeout_seconds)
request_id = str(uuid.uuid4())
messages = [
{
"role": "system", "content": "请给出准确、简洁的技术说明；不确定时明确说明条件。"},
{
"role": "user", "content": user_text},
]
for attempt in range(3):
parts: list[str] = []
try:
logging.info("model_request request_id=%s attempt=%s model=%s", request_id, attempt + 1, cfg.model)
stream = client.chat.completions.create(
model=cfg.model,
messages=messages,
stream=True,
)
for event in stream:
delta = event.choices[0].delta.content if event.choices else None
if delta:
print(delta, end="", flush=True)
parts.append(delta)
print()
return "".join(parts)
except Exception as error:
logging.warning("model_request_failed request_id=%s type=%s", request_id, type(error).__name__)
if not is_retryable(error) or attempt == 2:
raise
time.sleep(2 ** attempt)
raise RuntimeError("unreachable")
if __name__ == "__main__":
stream_answer("解释为什么网络请求需要设置超时，并列出实现要点。")
这里的重试次数和退避时间只是保守示例，不是通用最优值。若一次请求可能触发计费、创建工单或影响外部状态，仅凭网络异常无法判断服务端是否已经处理成功。此类场景应使用业务幂等键、服务端查询接口或状态机确认结果，不能简单重发。
将调用接入工具前的四个检查
第一，设定输入边界。限制单次文本大小、附件类型和可接受的字符编码；对来自用户或外部系统的内容，明确其只是数据，不应覆盖系统指令或直接取得工具权限。
第二，设定输出边界。若下游需要 JSON，不要只要求“返回 JSON”，而要用 JSON Schema 或 Pydantic 做解析与字段校验。解析失败时记录原始响应的受控摘要，返回可处理错误，不要把不合法文本直接传给数据库或 Shell。
第三，设定超时边界。连接超时、读取超时和整个任务的截止时间是不同概念。SDK 若只暴露统一超时，也应在外层任务系统设置总截止时间，避免队列被少量慢请求长期占用。
第四，设定审计边界。日志应包含请求 ID、模型标识、耗时、重试次数、结果状态和经过脱敏的业务上下文。不要默认记录完整提示词和响应，因为其中可能包含客户数据、代码或凭据。需要保留原文时，应先获得相应的数据处理授权，并设置访问控制与保留期限。
常见问题
为什么设置了 Base URL 仍然请求失败？
常见原因是地址路径与 SDK 预期不一致、企业代理拦截 TLS、环境变量未进入实际进程，或接入方并非兼容该 SDK 的协议。先在启动时记录经过脱敏后的主机名和模型名，再根据目标服务文档核对 URL 路径、认证头和请求格式。不要在排障日志中打印 API Key。
为什么 429 或 5xx 不应该无限重试？
无限重试会放大拥塞，并使任务延迟失去上限。针对可恢复错误使用有限次数、指数退避和随机抖动；超过预算后交给队列延迟执行或返回明确的可重试状态。对 4xx 参数错误、认证失败和内容校验失败，应直接失败并修复请求，而不是重试。
流式内容已经打印，异常后能否当作成功？
取决于业务契约。聊天展示可以把它标注为“响应中断，内容不完整”；要求完整结构、引用、代码块闭合或后续自动执行的任务，应丢弃该结果或进入人工复核。实现时可在内存累计片段，只有收到正常结束并通过校验后才写入正式记录。
如何避免模型切换影响业务？
将模型 ID、端点和能力开关配置化，并为关键任务建立小型回归样本集。样本应覆盖正常输入、超长输入、空输入、敏感字段、结构化输出和失败降级。这里验证的是业务契约，例如“必须产生可解析的对象”，而不是试图证明某个模型在所有问题上都更好。
总结
模型接入的可靠性来自明确边界，而不是更长的提示词或更多的重试。把配置从代码中移出，把流式输出视为可能中断的传输，把重试限制在可恢复且幂等的操作，并让结构化校验、权限控制和审计留在业务系统中，才能让本地脚本逐步演进为可运维的工具能力。选择具体 API 或中转服务时，最后应以其当前接口文档、数据处理规则、模型可用性和组织合规要求完成验证。
