第 23 章 · 与外部 API 及服务集成

# ❌ 这一行可能永久挂住
import requests
resp = requests.get("https://api.example.com/data")

# ✅ 超时必须显式指定，而且分两段
import httpx

resp = httpx.get(
    "https://api.example.com/data",
    timeout=httpx.Timeout(connect=5.0, read=30.0),   # 连接 5s，读取 30s
)

# clinic/agent_http.py（节选）
class HttpClient:
    """带重试、退避、限流与审计的 HTTP 客户端。

    设计原则：调用方只需要关心"要什么数据"，
    超时/重试/限流的细节全部封装在这里。
    """

    def __init__(self, base_url: str, token: str | None = None,
                 connect_timeout: float = 5.0, read_timeout: float = 30.0,
                 max_retries: int = 3, backoff_base: float = 1.5,
                 rate_limit_per_sec: float = 5.0,
                 audit_path: Path | None = None):
        ...

# ❌ 铁律 1：绝不硬编码
API_KEY = "sk-abc123..."                     # 提交到 git 就完了

# ❌ 铁律 2：绝不写进日志
logger.info(f"调用 API，token={token}")       # 日志会被收集、会被看到

# ❌ 铁律 3：绝不放进 Agent 的上下文
messages.append({"role": "system", "content": f"你的 token 是 {token}"})
#    ↑ 灾难：token 会进入模型 API 请求，也进入 state 序列化文件

# ✅ 正确做法
import os

def get_token() -> str:
    """从环境变量读取。本地用 .env（已 gitignore），生产用密钥管理服务。"""
    tok = os.environ.get("CDISC_API_TOKEN")
    if not tok:
        raise RuntimeError(
            "未设置环境变量 CDISC_API_TOKEN。\n"
            "本地开发：在 .env 里写 CDISC_API_TOKEN=xxx（.env 已在 .gitignore）\n"
            "生产环境：使用密钥管理服务注入环境变量"
        )
    return tok

SENSITIVE_KEYS = ("token", "key", "secret", "password", "authorization", "cookie")

def _mask(v: Any, prefix_keep: int = 4) -> str:
    """打码。保留前几位是为了**认出这是哪把钥匙**，不是为了好看。"""
    if v is None or v == "":
        return "***"
    s = str(v)
    scheme, sep, cred = s.partition(" ")
    if sep and cred and len(scheme) <= 12:          # ★ 形如 "Bearer <credential>"
        return f"{scheme} {cred[:prefix_keep]}***"
    return f"{s[:prefix_keep]}***" if len(s) > prefix_keep else "***"

RETRYABLE_STATUS = {408, 425, 429, 500, 502, 503, 504}

# ❌ 危险：POST 失败就重试 → 可能创建两份
for _ in range(3):
    try:
        post("/api/submissions", json=payload)
        break
    except TimeoutError:
        continue        # 第一次其实成功了，只是响应超时 → 现在有两份提交

# ✅ 带幂等键，服务端负责去重
post("/api/submissions", json=payload,
     headers={"Idempotency-Key": stable_key(payload)})

def stable_key(payload: dict) -> str:
    body = json.dumps(payload, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(body.encode()).hexdigest()[:32]

def _backoff(self, attempt: int, retry_after: float | None = None) -> float:
    """计算等待秒数。尊重服务端的 Retry-After。"""
    if retry_after is not None:
        return min(retry_after, 60.0)                 # 服务端说了算
    base = self.backoff_base ** attempt               # 1.5, 2.25, 3.375...
    jitter = random.uniform(0, base * 0.3)            # ★ 抖动
    return min(base + jitter, 30.0)

class RateLimiter:
    """简单的令牌桶。避免把对方服务打挂，也避免自己上黑名单。"""

    def __init__(self, per_sec: float, burst: int | None = None):
        self.interval = 1.0 / per_sec
        self.burst = burst or max(1, int(per_sec))
        self._tokens = float(self.burst)
        self._last = time.monotonic()
        self._lock = threading.Lock()

    def acquire(self) -> None:
        with self._lock:
            now = time.monotonic()
            self._tokens = min(self.burst, self._tokens + (now - self._last) / self.interval)
            self._last = now
            if self._tokens < 1:
                time.sleep((1 - self._tokens) * self.interval)
                self._tokens = 0
            else:
                self._tokens -= 1

def classify(status: int | None, exc: Exception | None) -> tuple[str, bool]:
    """返回 (错误类型, 是否可重试)。"""
    if exc is not None:
        if isinstance(exc, (httpx.ConnectTimeout, httpx.ReadTimeout,
                            httpx.ConnectError, httpx.ReadError)):
            return "transient", True
        return "internal", False
    if status in (400, 422):
        return "validation", False        # 请求本身有问题，重试无用
    if status in (401, 403):
        return "permission", False        # ★ 终止并告警，不要自动重试
    if status == 404:
        return "not_found", False
    if status == 429:
        return "transient", True          # 限流，退避后重试
    if 500 <= status < 600:
        return "transient", True
    return "internal", False

class CachedFallback:
    """外部数据取不到时，回退到上次的缓存，并明确标注数据时间。"""

    def get(self, key: str) -> tuple[Any, str]:
        try:
            data = self.client.fetch(key)
            self.cache[key] = (data, time.time())
            return data, f"实时获取（{_now()})"
        except TransientError:
            if key in self.cache:
                data, ts = self.cache[key]
                age = (time.time() - ts) / 3600
                return data, f"⚠️ 使用缓存数据（获取于 {age:.1f} 小时前）"
            raise

# ❌ 直接信任
data = resp.json()
df = pd.DataFrame(data["results"])          # KeyError: 'results'
df["age"] = df["age"].astype(int)           # ValueError: 有 "N/A"

# ✅ 显式校验
def parse_subjects(payload: dict) -> list[dict]:
    if not isinstance(payload, dict) or "results" not in payload:
        raise ContractError(
            f"响应缺少 results 字段。实际顶层键：{list(payload)[:10]}"
        )
    rows = payload["results"]
    if not isinstance(rows, list):
        raise ContractError(f"results 应为数组，实际为 {type(rows).__name__}")

    out = []
    for i, r in enumerate(rows):
        if not isinstance(r, dict) or "subjectId" not in r:
            raise ContractError(f"第 {i} 条记录缺少 subjectId：{str(r)[:120]}")
        out.append({
            "subject_id": str(r["subjectId"]).strip(),
            "age": pd.to_numeric(r.get("age"), errors="coerce"),   # 宽容转数值
            "sex": str(r.get("sex", "")).strip().upper() or None,
        })
    return out

resp = client.get("/api/v2/subjects", headers={"API-Version": "2024-06-01"})

@dataclass
class Watermark:
    """记录上次成功拉取到哪，避免重复拉、避免漏拉。"""
    last_id: str | None = None
    last_ts: str | None = None
    updated_at: float = 0.0

    def save(self, path: Path) -> None:
        path.write_text(json.dumps(asdict(self), ensure_ascii=False), encoding="utf-8")

    @classmethod
    def load(cls, path: Path) -> "Watermark":
        if not path.exists():
            return cls()
        return cls(**json.loads(path.read_text(encoding="utf-8")))

# ❌ 灾难：把可识别信息发给了外部 LLM
prompt = f"""
请分析受试者 {subj.subject_name}（出生日期 {subj.dob}，
身份证号 {subj.id_number}）的实验室结果：
{lab_results}
"""

# ✅ 只发研究编号 + 必要的分析字段
prompt = f"""
请分析受试者 {row.USUBJID} 在访视 {row.VISIT} 的实验室结果：
{deidentified_labs}
"""

def build_llm_payload(df: pd.DataFrame, need: list[str]) -> str:
    """只把需要的列发给模型。多一列就多一分风险，也多一分 token。"""
    missing = [c for c in need if c not in df.columns]
    if missing:
        raise ValueError(f"缺少必需列：{missing}")
    return df[need].to_csv(index=False)

def log_llm_call(prompt: str, response: str, model: str,
                 audit_dir: Path = Path("outputs/audit")) -> None:
    """每次模型调用落一行。这是"可追溯"的最小实现。"""
    audit_dir.mkdir(parents=True, exist_ok=True)
    rec = {
        "ts": datetime.now().isoformat(timespec="seconds"),
        "model": model,
        "prompt_chars": len(prompt),
        "prompt_sha256": hashlib.sha256(prompt.encode()).hexdigest()[:16],
        "response_chars": len(response),
        "prompt_preview": prompt[:200],          # 只留预览，不存全文
    }
    with (audit_dir / f"llm_calls_{date.today():%Y%m}.jsonl").open(
            "a", encoding="utf-8") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")

# cases/case_http_demo.py（本章配套，见 clinic/agent_http.py）
from clinic.agent_http import HttpClient, MockTransport

# 离线演示：用 MockTransport，不真的发网络请求
client = HttpClient(
    base_url="https://ct.example.org",
    token="demo-token",
    transport=MockTransport({                      # ★ 可注入的传输层
        "/api/v2/codelists/AE": {
            "version": "2024-06-01",
            "items": [{"code": "HEADACHE", "term": "Headache"}, ...],
        },
    }),
    # ★ 让"等多久"可断言：不真的睡，也不真的随机
    sleep=lambda s: waits.append(s),
    rand=lambda: 0.5,
)

resp = client.get("/api/v2/codelists/AE")      # 返回统一封装的 Response
data = resp.json()                             # 显式解析，非法 JSON 会抛 ContractError
print(f"取到 {len(data['items'])} 条术语，版本 {data['version']}")

# 审计日志（自动脱敏）
client.dump_audit("outputs/http/audit_snapshot.json")

class MockTransport:
    """离线传输层：不打网络，按路径返回预置数据。"""
    def __init__(self, routes=None, fail_paths=None, fail_times=None):
        # routes     正常返回
        # fail_paths 永久失败（int 状态码 / Exception / 带响应头的 Response）
        # fail_times 前 N 次失败、之后成功 —— 专门用来验证重试
        ...

    def request(self, method, url, **kw) -> Response:
        ...