把 NeoHorse-1 接进现有系统，真正需要先定下来的不是模型参数，而是中间那层薄接口：业务代码只描述“要做什么”，不负责拼模型请求、不直接拿密钥、也不自己判断网络超时。这层接口通常只做四件事——统一入参、统一鉴权、统一超时重试、统一错误翻译。把这四件事放在一个模块里，业务侧后续换模型、换地址、调超时，改动面都能收敛到一个文件，而不是散落到几十个调用点。

接口层建议先定最小契约：业务只传任务和上下文，接口返回统一信封（是否成功、内容、错误码、是否可重试、trace_id）。鉴权头从环境变量读取并按需做日志脱敏；超时拆成连接、首包、总超时三级，重试限制在 2 到 3 次并只对可重放请求生效。错误按参数、鉴权、限流、服务端、网络五类映射后再返回业务。落地后用一次人为注入的延迟或断开验证整条链路，确认日志、重试和降级提示符合预期；具体阈值需要结合你的网络环境与模型响应特征确认。

定义接口层对外暴露的最小契约

契约的目标是让业务代码看不到鉴权、重试、序列化这些细节。一个够用的签名可以先写成这样，参数名按你的语言习惯替换：

def run_task(task: str, context: dict | None = None, *, trace_id: str,
             timeout: float | None = None) -> dict:
    """返回统一信封，不抛业务异常"""
    ...

输入字段建议只保留三项：task 是必填的字符串，说明这次要模型做什么；context 是可选字典，只放模型确实需要的事实，例如用户问题、历史摘要、格式约束，不要把整个业务对象塞进去；trace_id 由调用方生成并在日志中透传。输出统一走一个信封，字段固定为：ok（布尔）、output（成功时的内容）、error_code（失败时的分类码）、error_message（给日志看的原始信息）、retryable（布尔）、trace_id。失败返回示例：

{"ok": False, "output": None,
 "error_code": "UPSTREAM_TIMEOUT",
 "error_message": "total timeout after 60.0s",
 "retryable": True, "trace_id": "t-8f3a"}

这里刻意不写具体接口路径和模型名：路径属于部署配置，放在环境变量或配置文件里更合适，契约层只声明字段和语义。

封装鉴权头和基础请求

密钥散落在业务代码里是最容易出问题的写法。建议只在接口层模块内读取一次环境变量，业务代码完全接触不到：

import os
BASE_URL = os.environ["NEOHORSE_BASE_URL"]
API_KEY = os.environ.get("NEOHORSE_API_KEY", "")

def build_headers(trace_id: str) -> dict:
    return {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
        "X-Trace-Id": trace_id,
    }

请求头里的 Bearer 前缀、内容类型、追踪头都是占位写法，实际字段名以你接入的服务端约定为准。日志脱敏建议做成一个固定函数，任何打印密钥或响应头的地方都先过一遍：

def mask(secret: str) -> str:
    if not secret:
        return "<empty>"
    return secret[:4] + "***" + secret[-4:]

启动时可以打一行 logger.info("auth key = %s", mask(API_KEY))，这样既能在排障时确认密钥已加载，又不会把完整值写进日志文件。需要确认的是：环境变量缺失时应该尽早失败并给出明确提示，而不是带着空密钥去发请求。

设置分级超时和有限重试

把超时写成一个值是最常见的坑。建议拆成三级：连接超时、首包超时（等待第一个字节或首个数据块）、总超时。模型类调用首包可能明显慢于普通接口，三者分开配置后，排障时才能判断是连不上、还是服务端在算：

TIMEOUTS = {
    "connect": 3.0,
    "first_byte": 15.0,
    "total": 60.0,
}
RETRY = {
    "max_attempts": 3,
    "base_delay": 0.5,
    "factor": 2.0,
    "max_delay": 8.0,
    "jitter": True,
}

重试骨架按“尝试次数 + 指数退避 + 抖动”实现，达到 max_attempts 就停止，不要写成 while True。幂等性要提前确认：只读的推理请求通常可以安全重放；带副作用的调用要么加一个由调用方生成的 client_request_id 让下游去重，要么干脆不重试。抖动的作用是避免多个请求在同一时刻一起重试，属于常见做法，具体参数需要结合你的并发规模确认。

把底层错误映射成业务错误

上层不需要知道 HTTP 状态码，只需要知道“该不该改输入、该不该等、该不该找运维”。建议先定义一张固定的分类表：

捕获时按异常类型分流，避免一个 except Exception 吃掉所有信息：

try:
    resp = do_request(...)
except (ConnectTimeout, ReadTimeout):
    return fail("UPSTREAM_TIMEOUT", retryable=True)
except ConnectionError:
    return fail("UPSTREAM_ERROR", retryable=True)
except ValueError as e:
    return fail("INVALID_INPUT", retryable=False, detail=str(e))

返回给业务的提示建议保持稳定，不要把原始堆栈透出到用户界面：可重试的任务统一提示“任务未完成，请稍后重试”，参数类错误提示“输入不符合要求，请检查后重试”，鉴权类错误提示“服务配置异常，请联系管理员”。原始信息只写日志，用 trace_id 关联。

用一次模拟故障验证整条链路

配置写完不等于链路可用。可以用两种低成本方式各验一次。注入延迟：把 first_byte 临时调到 1 秒，或者让本地一个假服务故意慢几秒再返回，观察是否触发重试、最终是否返回 UPSTREAM_TIMEOUT。注入断开：把 BASE_URL 指向一个不可达地址，或用防火墙规则丢弃目标端口，观察连接超时是否在预期时间内触发，而不是一直挂着。

验证时重点看日志里的四样东西：每次尝试的序号、单次耗时、错误分类、是否决定重试，以及最终返回的信封里 trace_id 能否和业务侧日志对上。回退逻辑也要一起验：当配置项缺失或读取失败时，应使用代码里的默认超时常量继续运行；当重试次数耗尽仍然失败时，返回带 retryable=True 的失败信封，由业务决定是排队重试还是直接降级提示，而不是在接口层无限等待。