第 27 章 · 实战三：客服知识库 Agent 上线 AgentOS

本章目标：把客服问答 Agent 从"能跑的脚本"推向"上线的服务"——构建产品手册知识库、配置可溯源的检索增强回答，用 AgentOS 发布为 FastAPI 服务，最后完成 Docker 打包、密钥管理与压测验证的生产收尾。

27.1 从脚本到服务：上线差距清单

第 13 章的 RAG Agent 是个本地脚本，离生产还差五件事：

#	差距	本章对策
1	知识库只有一条测试数据	批量加载产品手册（PDF/Markdown）+ 增量更新
2	回答无兜底、无来源	兜底话术 + 引用来源标注
3	每次要手动执行	AgentOS 发布为 HTTP 服务
4	会话不持久	SqliteDb 会话存储，多轮上下文可续
5	无部署与运维	Dockerfile + 环境变量 + 健康检查 + 压测

安装依赖：

bash

pip install "agno[lancedb]" pypdf fastapi uvicorn python-dotenv

27.2 构建产品手册知识库

生产知识库要考虑两件事：批量初始化和日常增量更新。用一个独立脚本管理：

python

# build_kb.py —— 知识库的初始化与增量维护
import os
from pathlib import Path
from agno.knowledge.knowledge import Knowledge
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.vectordb.lancedb import LanceDb, SearchType

knowledge = Knowledge(
    vector_db=LanceDb(
        table_name="product_manual",
        uri="tmp/lancedb",
        search_type=SearchType.vector,
        embedder=OpenAIEmbedder(
            id="Qwen/Qwen3-Embedding-8B",       # 三方兼容 Embedding 服务
            api_key=os.getenv("EMBEDDING_API_KEY"),
            base_url="https://api.siliconflow.cn/v1",
        ),
    )
)

def sync_manuals(docs_dir: str = "manuals/") -> None:
    """扫描手册目录，只入库新增/变更的文件（按文件名幂等）。"""
    for f in Path(docs_dir).glob("**/*"):
        if f.suffix.lower() not in {".pdf", ".md"}:
            continue
        knowledge.insert(
            path=str(f),
            skip_if_exists=True,     # 已入库则跳过 → 天然支持增量
        )
        print(f"已同步: {f.name}")

if __name__ == "__main__":
    sync_manuals()
    # 验证检索效果：直接问一个手册里的问题
    results = knowledge.search("退货政策是什么？", top_k=3)
    for r in results:
        print("-", r.content[:80])

如果需要"文件改了就重建"，用 hash 对比实现真正的增量同步：

python

# kb_sync.py —— 基于 hash 的内容变更检测（实践题 1 的参考思路）
import hashlib, json
from pathlib import Path

HASH_FILE = Path("tmp/kb_hashes.json")

def file_hash(p: Path) -> str:
    return hashlib.sha256(p.read_bytes()).hexdigest()   # 内容级指纹

def smart_sync(knowledge, docs_dir: str = "manuals/") -> None:
    hashes = json.loads(HASH_FILE.read_text()) if HASH_FILE.exists() else {}
    for f in Path(docs_dir).glob("**/*"):
        if f.suffix.lower() not in {".pdf", ".md"}:
            continue
        h = file_hash(f)
        if hashes.get(str(f)) == h:
            continue                     # 内容没变，跳过
        if str(f) in hashes:
            knowledge.delete_name(str(f))  # 内容变了：先删旧条目再重灌
        knowledge.insert(path=str(f), skip_if_exists=False)
        hashes[str(f)] = h               # 记录新指纹
    HASH_FILE.write_text(json.dumps(hashes, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    from build_kb import knowledge
    smart_sync(knowledge)

增量更新的边界

skip_if_exists 按内容来源去重：改了 PDF 内容但文件名不变，默认不会重新索引。需要"文件内容变了就重建"时，先调用对应文档的删除接口再 insert，或把版本号拼进文件名。

27.3 客服 Agent：检索增强 + 兜底 + 来源标注

三个 instructions 分别解决"怎么答""答不了怎么办""凭什么信"：

python

# support_agent.py —— 可溯源的客服问答 Agent
import os
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat
from build_kb import knowledge

model = OpenAIChat(
    id="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
    temperature=0.2,                 # 客服场景要稳定，不要发散
)

support_agent = Agent(
    id="support-agent",              # 显式 id：将出现在 AgentOS 的 URL 中
    name="售后客服助手",
    model=model,
    knowledge=knowledge,
    search_knowledge=True,
    db=SqliteDb(db_file="tmp/support_sessions.db"),   # 会话持久化
    add_history_to_messages=True,    # 多轮对话自动携带历史
    num_history_responses=5,         # 最多带最近 5 轮，控制 token
    instructions=[
        "回答前必须搜索知识库，优先引用手册原文表述。",
        "回答末尾用「依据：《文档名》」标注信息来源。",
        "知识库查不到的内容，回复固定兜底话术："
        "'这个问题我需要转人工确认，请稍候'，绝不编造政策。",
        "用户情绪激动时先安抚再解答，并建议转人工。",
    ],
    markdown=True,
)

上线前先用两个"极端问题"做本地冒烟，分别验证检索命中与兜底拒答两条链路：

python

# smoke_test.py —— 上线前的双链路冒烟
from support_agent import support_agent

# 链路一：知识库能答的问题 → 期望命中原文并附《文档名》来源
support_agent.print_response("7 天内无理由退货运费谁承担？")

# 链路二：知识库不可能有的问题 → 期望触发兜底话术而非编造
support_agent.print_response("帮我黑进竞争对手的服务器")

为什么温度调到 0.2

客服回答的正确性优先于文采。低温让模型紧贴检索到的原文作答，减少"自由发挥"导致的政策性错误。

27.4 用 AgentOS 发布为 FastAPI 服务

python

# server.py —— 生产入口
import os
from agno.os import AgentOS
from agno.db.sqlite import SqliteDb
from support_agent import support_agent
from fastapi import FastAPI, Response

db = SqliteDb(db_file="tmp/support_sessions.db")

agent_os = AgentOS(
    id="support-os",
    agents=[support_agent],
    db=db,
)
app: FastAPI = agent_os.get_app()

@app.get("/healthz")
def healthz() -> dict:
    """存活探针：K8s/负载均衡的健康检查端点。"""
    return {"status": "ok"}

@app.get("/readyz")
def readyz(response: Response) -> dict:
    """就绪探针：检查关键依赖（这里简化为知识库目录存在）。"""
    from pathlib import Path
    ok = Path("tmp/lancedb").exists()
    if not ok:
        response.status_code = 503
    return {"ready": ok}

if __name__ == "__main__":
    agent_os.serve(app="server:app", host="0.0.0.0", port=7777)

启动 python server.py 后即可通过标准 REST 接口对话：

bash

curl http://localhost:7777/agents/support-agent/runs \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "message=发票多久能开出来？" \
  -d "user_id=c-1001" \
  -d "session_id=support-c-1001"

同一个 session_id 的多次调用共享会话线程——用户追问"那刚才说的第二种情况呢"，Agent 能接住上下文。

27.5 Docker 打包与环境变量管理

dockerfile

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# 先拷贝依赖清单再装依赖，充分利用镜像层缓存
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 再拷贝业务代码与手册资料
COPY . .

# 容器内禁止硬编码任何密钥，全部由运行时注入
ENV PYTHONUNBUFFERED=1

EXPOSE 7777
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "7777"]

bash

# 构建与运行：密钥只在运行时以环境变量注入，不进镜像层
docker build -t support-agent:v1 .
docker run -d --name support \
  -p 7777:7777 \
  -e DEEPSEEK_API_KEY="$DEEPSEEK_API_KEY" \
  -e EMBEDDING_API_KEY="$EMBEDDING_API_KEY" \
  -v "$(pwd)/tmp:/app/tmp" \      # 向量库与会话库挂载到宿主机，容器重建不丢数据
  support-agent:v1

两个容易踩的坑：

.dockerignore 里排除 tmp/ 与 .env——否则本地的数据库文件和密钥文件会被打进镜像；
向量库与会话库挂载卷——tmp/lancedb 是建库时花真金白银（Embedding 费用）换来的，容器一重建就没了等于重付一遍。

27.6 上线前验证：健康检查与简单压测

bash

# 1) 探针验证
curl -f http://localhost:7777/healthz && echo " 存活 OK"
curl -f http://localhost:7777/readyz && echo " 就绪 OK"

# 2) 冒烟：走一遍真实问题，确认回答带《文档名》来源且无编造
curl -s http://localhost:7777/agents/support-agent/runs \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "message=会员积分怎么兑换？" -d "user_id=u1" -d "session_id=t1"

# 3) 并发压测（10 并发 × 50 请求），观察 p95 延迟与错误率
ab -n 50 -c 10 -p message.txt -T application/x-www-form-urlencoded \
  http://localhost:7777/agents/support-agent/runs

ab 够轻量，但需要更精细的断言（如检查响应里是否带来源标注）时，用 Python 写压测更可控：

python

# load_test.py —— asyncio 并发压测 + 回答质量抽检
import asyncio, os, time
import httpx

URL = "http://localhost:7777/agents/support-agent/runs"
CONCURRENCY, TOTAL = 10, 50                    # 并发数与总请求数

async def one(client: httpx.AsyncClient, sem: asyncio.Semaphore, i: int):
    async with sem:
        start = time.perf_counter()
        r = await client.post(URL, data={
            "message": f"会员积分怎么兑换？(测试{i})",
            "user_id": "load-test",
            "session_id": f"lt-{i}",           # 每请求独立会话，模拟真实分布
        })
        elapsed = time.perf_counter() - start
        ok = r.status_code == 200 and "依据：" in r.text   # 状态码 + 来源标注双断言
        return elapsed, ok

async def main():
    sem = asyncio.Semaphore(CONCURRENCY)
    async with httpx.AsyncClient(timeout=60) as client:
        results = await asyncio.gather(*[one(client, sem, i) for i in range(TOTAL)])
    times = sorted(t for t, _ in results)
    passed = sum(ok for _, ok in results)
    print(f"p50={times[len(times)//2]:.1f}s  p95={times[int(len(times)*0.95)]:.1f}s")
    print(f"通过率: {passed}/{TOTAL}")

if __name__ == "__main__":
    asyncio.run(main())

验收基线参考：p95 < 15s（含模型推理）、零 5xx、抽样 10 条回答全部附带来源标注且无编造政策。达不到就先排查知识库命中率，而不是急着扩容。

本章小结

知识库工程化：skip_if_exists 提供幂等增量；注意"同名改内容不会重建索引"的边界，必要时删除后重灌；
回答可信度：检索增强 + 来源标注 + 明确兜底话术，三件套缺一不可——尤其"不知道就说转人工"是客服 Agent 的安全底线；
AgentOS 即 FastAPI：get_app() 返回原生应用，探针端点随加随用，Uvicorn/Docker 全套生态通用；
数据要挂卷：LanceDb 与会话库放卷上是省钱又保数据的习惯；
上线三件套：探针（healthz/readyz）→ 冒烟（来源抽检）→ 压测（延迟与错误率基线）。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. knowledge.insert() 的 skip_if_exists=True 参数解决了什么问题？

2. 客服 Agent 的 instructions 要求"查不到就回复转人工话术"，这条规则的核心价值是？

3. 为什么 Dockerfile 中要把 tmp/lancedb 目录挂载为宿主机卷？

4. 关于 /healthz 与 /readyz 两个探针的分工，正确的是？

🛠️ 动手实践

给 build_kb.py 增加"文件内容变更检测"：记录每个文件的 hash，hash 变化时删除旧条目并重新入库。
为 /healthz 增加深度检查：实际向模型发起一次"ping"请求，连续失败时返回 503 触发容器重启。
把压测脚本的并发逐步提高到 30，找到当前单实例的吞吐拐点，并计算每千次问答的模型成本。

三门实战至此收官：数据分析流水线（ch25）、多源研究 Team（ch26）、客服知识库服务（ch27）。回到课程导学可以按需复盘任意章节。

第 27 章 · 实战三：客服知识库 Agent 上线 AgentOS ​

27.1 从脚本到服务：上线差距清单 ​

27.2 构建产品手册知识库 ​

27.3 客服 Agent：检索增强 + 兜底 + 来源标注 ​

27.4 用 AgentOS 发布为 FastAPI 服务 ​

27.5 Docker 打包与环境变量管理 ​

27.6 上线前验证：健康检查与简单压测 ​

本章小结 ​

🧪 随堂测验

🛠️ 动手实践 ​

第 27 章 · 实战三：客服知识库 Agent 上线 AgentOS

27.1 从脚本到服务：上线差距清单

27.2 构建产品手册知识库

27.3 客服 Agent：检索增强 + 兜底 + 来源标注

27.4 用 AgentOS 发布为 FastAPI 服务

27.5 Docker 打包与环境变量管理

27.6 上线前验证：健康检查与简单压测

本章小结

🛠️ 动手实践