Agent 开发实战（九）：工程化部署 + 端到端案例——一个生信分析 Agent 收尾

前面八篇分别解决了数据接入、工具、记忆、流式、RAG、多 Agent、可观测性。本篇把它们整合成一个能部署上线的完整 Agent，并以"生信分析 Agent"做端到端案例收尾——它能查样本库、读 VCF 变异、搜实验文档，最后产出一份带引用的分析报告。同时给出 Docker 化、鉴权、限流的工程化骨架。

1. 端到端场景

用户："帮我看下 2026-07 批次的测序样本，统计变异类型分布，并对照《测序质控 SOP》看看有没有该拦截的样本。"

Agent 的链路：

query_table(samples, …) → 拿到 7 月批次样本列表
vcf_summary(sample_A.vcf.gz) → 变异类型与样本数
search_documents("测序质控 SOP 拦截标准") → 召回质控条款
多 Agent 汇总 → 产出"变异统计 + 是否触发拦截 + 引用条款"的报告

这正是第（二）篇多源 + 第（六）篇 RAG + 第（七）篇多 Agent + 第（八）篇日志的合体。

2. 工程化：目录与装配

agent-service/
├── connectors/      # db / sheet / doc / bio（前几篇已实现）
├── tools/           # registry + executor（第二篇）
├── memory/          # 短期/长期/压缩（第三篇）
├── server.py        # FastAPI：SSE 流式（第四篇）
├── agent.py         # run_agent / run_multiagent（第六篇）
├── observ.py        # 日志 + 成本（第七篇）
├── auth.py          # API Key 鉴权 + 限流
├── Dockerfile
├── docker-compose.yml
└── .env.example

核心装配（节选）：

# server.py
from fastapi import FastAPI, Request, Header, HTTPException
from fastapi.responses import StreamingResponse
import time
from tools.registry import registry
from tools.executor import execute_one
from memory.compress import compress_tool_result
from observ import log, trace_id

app = FastAPI()

@app.post("/api/agent/ask")
def ask(req: dict, request: Request, x_api_key: str = Header(None)):
    if x_api_key != os.getenv("API_KEY"):
        raise HTTPException(401)
    return StreamingResponse(agent_stream(req["question"]),
                             media_type="text/event-stream")

3. 鉴权与限流

# auth.py —— 极简令牌 + 内存限流（生产换 Redis）
from collections import defaultdict
import time

RATE = defaultdict(list)

def rate_limit(ip: str, limit=20, window=60):
    now = time.time()
    RATE[ip] = [t for t in RATE[ip] if now - t < window]
    if len(RATE[ip]) >= limit:
        raise HTTPException(429, "too many requests")
    RATE[ip].append(now)

对外暴露的 Agent 必须加鉴权，否则它就是个数据泄露口。即使内网，也建议至少一层 API Key。

4. Docker 化（生信工具要进镜像）

生信连接器依赖 bcftools，必须在镜像里装好：

# Dockerfile
FROM python:3.11-slim
RUN apt-get update && apt-get install -y --no-install-recommends \
    bcftools samtools && rm -rf /var/lib/apt/lists/*
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
services:
  agent:
    build: .
    environment:
      - DB_HOST=db
      - DB_USER_READONLY=${DB_USER_READONLY}
      - DB_PASSWORD=${DB_PASSWORD}
      - LLM_API_KEY=${LLM_API_KEY}
      - API_KEY=${API_KEY}
    ports: ["8000:8000"]
    depends_on: [redis]
  redis:
    image: redis:7

# .env.example
DB_HOST=localhost
DB_USER_READONLY=agent_ro
DB_PASSWORD=******
LLM_API_KEY=sk-******
LLM_BASE_URL=https://your-compatible-endpoint
LLM_MODEL=gpt-4o-mini
API_KEY=server-secret
SHEET_DIR=/data/sheets
DOC_DIR=/data/docs
BIO_DIR=/data/bio

5. 健康检查

@app.get("/healthz")
def healthz():
    return {"status": "ok"}

K8s / 容器编排靠它做存活探针。

6. 端到端跑一遍

curl -N -X POST http://localhost:8000/api/agent/ask \
  -H "Content-Type: application/json" -H "x-api-key: server-secret" \
  -d '{"question":"统计 2026-07 批次样本的变异类型分布，并对照质控 SOP 看是否该拦截"}'

你会看到 SSE 流：先推 tool 事件（查样本→读 VCF→搜 SOP），再逐 token 推报告。报告里每个结论都带 [n] 引用，可回跳原文档核实。

7. 上线前 checklist

只读 DB 账号 + 各源目录白名单（第（二）篇）
工具权限分级 + 缓存 + 超时（第（三）篇）
记忆隔离与脱敏（第（四）篇）
SSE 流式 + 中断（第（五）篇）
混合检索 + 引文（第（六）篇）
复杂任务多 Agent（第（七）篇）
日志 + 成本 + eval（第（八）篇）
鉴权 + 限流 + 健康检查 + Docker（本篇）

8. 系列收尾

八篇走下来，我们从一个"能查数据库的玩具"逐步长成了一个生产级 Agent 框架：

多源数据接入 → 2. 工具编排 → 3. 记忆管理 → 4. 流式前端 → 5. RAG 进阶 → 6. 多 Agent → 7. 可观测评估 → 8. 工程化部署。

每篇都可独立成章，合起来是一条完整的落地路径。最重要的不是某一项技术，而是那条贯穿始终的原则：

模型只决策，工具才执行；每加一层能力，先加一道护栏。

护栏先行，Agent 才敢上线；能力渐进，团队才跟得上。祝你把自己的网站/系统，也接上一个"说得清、查得准、管得住"的 AI 助手。

系列到此告一段落，但 Agent 的工程化没有终点。下一步可以做的：接入评测平台、做工具自动发现、把多 Agent 编排可视化。路还长，先把地基打牢。