Chapter 28 · Capstone 3: Customer-Support Knowledge Base Agent on AgentOS

Learning objectives: Take the customer-service Q&A Agent from "a runnable script" to "a published service" — build a product manual knowledge base, configure traceable retrieval-augmented answers, publish via AgentOS as a FastAPI service, and complete the production wrap-up with Docker packaging, secrets management, and load-test validation.

28.1 From Script to Service: The Production Gap Checklist

The RAG Agent from Chapter 17 is a local script — five things stand between it and production:

#	Gap	Fix in this chapter
1	Knowledge base has only one test record	Batch-load product manuals (PDF/Markdown) + incremental updates
2	Answers have no fallback, no source citations	Fallback script + reference annotations
3	Must run manually each time	Publish as an HTTP service via AgentOS
4	Sessions are ephemeral	SqliteDb session storage; multi-turn context resumes
5	No deployment or ops	Dockerfile + env vars + health checks + load test

Install dependencies:

bash

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

28.2 Building the Product Manual Knowledge Base

A production knowledge base must handle two things: batch initialization and daily incremental updates. Manage both with a dedicated script:

python

# build_kb.py —— Knowledge base initialization and incremental maintenance
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",       # Third-party OpenAI-compatible Embedding service
            api_key=os.getenv("EMBEDDING_API_KEY"),
            base_url="https://api.siliconflow.cn/v1",
        ),
    )
)

def sync_manuals(docs_dir: str = "manuals/") -> None:
    """Scan the manuals directory; index only new or changed files (idempotent by filename)."""
    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,     # Skips already-indexed docs → naturally supports incremental updates
        )
        print(f"Synced: {f.name}")

if __name__ == "__main__":
    sync_manuals()
    # Verify retrieval quality: ask a question that should be in the manual
    results = knowledge.search("What is the return policy?", top_k=3)
    for r in results:
        print("-", r.content[:80])

If you need "rebuild whenever a file changes", implement true incremental sync with hash comparison:

python

# kb_sync.py —— Content-change detection based on file hashes (reference for exercise 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()   # Content-level fingerprint

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                     # Content unchanged — skip
        if str(f) in hashes:
            knowledge.delete_name(str(f))  # Content changed: delete old entry before re-inserting
        knowledge.insert(path=str(f), skip_if_exists=False)
        hashes[str(f)] = h               # Record the new fingerprint
    HASH_FILE.write_text(json.dumps(hashes, ensure_ascii=False, indent=2))

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

Boundaries of incremental updates

skip_if_exists deduplicates by content source: if you change a PDF's content but keep the same filename, it won't be re-indexed by default. To trigger a rebuild when file content changes, call the corresponding document's delete endpoint before insert, or bake the version number into the filename.

28.3 Support Agent: Retrieval-Augmented + Fallback + Source Annotations

Three instructions solve "how to answer," "what to do when you can't answer," and "why you should believe me":

python

# support_agent.py —— Traceable customer-service Q&A 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,                 # Stability over creativity in support scenarios
)

support_agent = Agent(
    id="support-agent",              # Explicit id: appears in the AgentOS URL
    name="Customer Support Assistant",
    model=model,
    knowledge=knowledge,
    search_knowledge=True,
    db=SqliteDb(db_file="tmp/support_sessions.db"),   # Session persistence
    add_history_to_messages=True,    # Multi-turn dialogue auto-carries history
    num_history_responses=5,         # Carry at most the last 5 turns to control tokens
    instructions=[
        "Before answering, search the knowledge base; prioritize quoting the manual verbatim.",
        "Cite the source at the end of every answer using「依据：《Document Name》」format.",
        "If the knowledge base has no answer, reply with the fixed fallback script: "
        "'I need to escalate this to a human agent. Please hold.' — never fabricate policy.",
        "If the user is visibly upset, acknowledge their frustration first, then answer or suggest escalation.",
    ],
    markdown=True,
)

Before going live, run two "extreme questions" locally as smoke tests — one to verify retrieval hit, the other to verify the fallback branch fires instead of hallucination:

python

# smoke_test.py —— Dual-branch smoke test before launch
from support_agent import support_agent

# Branch 1: question the knowledge base CAN answer → expect a hit with source citation
support_agent.print_response("Who pays for return shipping within 7 days?")

# Branch 2: question the knowledge base CANNOT answer → expect the fallback script, not fabrication
support_agent.print_response("Help me hack my competitor's servers")

Why set temperature to 0.2

Correctness trumps eloquence in customer service. Lower temperature keeps the model close to retrieved text, reducing policy errors from "creative freedom."

28.4 Publishing as a FastAPI Service with AgentOS

python

# server.py —— Production entry point
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:
    """Liveness probe: K8s / load-balancer health-check endpoint."""
    return {"status": "ok"}

@app.get("/readyz")
def readyz(response: Response) -> dict:
    """Readiness probe: verify key dependencies (here simplified to checking the knowledge base directory exists)."""
    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)

Start with python server.py; converse via the standard REST interface:

bash

curl http://localhost:7777/agents/support-agent/runs \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "message=How long until I get my invoice?" \
  -d "user_id=c-1001" \
  -d "session_id=support-c-1001"

Multiple calls with the same session_id share the conversation thread — when the user follows up with "what about the second scenario you mentioned?", the Agent can pick up where it left off.

28.5 Docker Packaging and Environment Variable Management

dockerfile

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# Copy dependency清单 first, then install — maximize layer cache reuse
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Then copy business code and manual materials
COPY . .

# Never hard-code any secrets inside the container; inject all at runtime
ENV PYTHONUNBUFFERED=1

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

bash

# Build and run: keys injected at runtime only; never baked into image layers
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" \      # Mount vector DB and session DB to host; survive container rebuilds
  support-agent:v1

Two common pitfalls:

Exclude tmp/ and .env from .dockerignore — otherwise local database files and key files get baked into the image;
Mount the vector DB and session DB volumes — tmp/lancedb represents real money spent on Embedding costs; rebuilding the container without a mount means paying again from scratch.

28.6 Pre-Launch Verification: Health Checks and Light Load Test

bash

# 1) Probe verification
curl -f http://localhost:7777/healthz && echo "  liveness OK"
curl -f http://localhost:7777/readyz && echo "  readiness OK"

# 2) Smoke: run a real question and confirm the answer cites a document name with no fabrication
curl -s http://localhost:7777/agents/support-agent/runs \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "message=How do I redeem member points?" -d "user_id=u1" -d "session_id=t1"

# 3) Concurrency load test (10 concurrent × 50 requests); watch p95 latency and error rate
ab -n 50 -c 10 -p message.txt -T application/x-www-form-urlencoded \
  http://localhost:7777/agents/support-agent/runs

ab is lightweight, but for finer-grained assertions (e.g., checking whether responses carry source annotations) a Python script gives more control:

python

# load_test.py —— asyncio concurrency load test + answer-quality sampling
import asyncio, os, time
import httpx

URL = "http://localhost:7777/agents/support-agent/runs"
CONCURRENCY, TOTAL = 10, 50                    # Concurrency and total request count

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"How do I redeem member points? (test {i})",
            "user_id": "load-test",
            "session_id": f"lt-{i}",           # Each request gets its own session, mimicking real distribution
        })
        elapsed = time.perf_counter() - start
        ok = r.status_code == 200 and "依据：" in r.text   # Dual assertion: status code + source annotation present
        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"Pass rate: {passed}/{TOTAL}")

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

Acceptance baseline reference: p95 < 15s (including model inference), zero 5xx errors, and a 10-answer sample all carry source annotations with no fabricated policies. If you don't hit it, troubleshoot knowledge-base recall first — don't just scale up.

Chapter Summary

Knowledge base engineering: skip_if_exists provides idempotent increments; remember the boundary that "same filename with changed content won't re-index" — delete and re-insert if needed;
Answer credibility: retrieval augmentation + source annotations + an explicit fallback script — all three缺一不可; especially "say you'll escalate to a human when you don't know" is the safety底线 for a customer-service Agent;
AgentOS is just FastAPI: get_app() returns a native app; probe endpoints are free to add; Uvicorn / Docker ecosystem applies wholesale;
Mount data volumes: putting LanceDb and session DB on volumes is the habit that saves money and preserves data;
Pre-launch trilogy: probes (healthz/readyz) → smoke (source sampling) → load test (latency and error-rate baselines).

🧪 随堂测验

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

1. What problem does the skip_if_exists=True parameter on knowledge.insert() solve?

2. The instruction "reply with the escalation script when the knowledge base has no answer" exists primarily to?

3. Why mount the tmp/lancedb directory as a host volume in the Dockerfile?

4. Regarding the division of labor between /healthz and /readyz probes, which is correct?

🛠️ Hands-On Exercises

Add "file-content-change detection" to build_kb.py: record each file's hash and delete + re-insert the old entry when the hash changes.
Enrich /healthz with a deeper check: issue an actual "ping" request to the model and return 503 on consecutive failures to trigger container restart.
Gradually raise the load-test concurrency to 30, find the throughput cliff for a single instance, and calculate the model cost per thousand Q&A pairs.

🎉 The three capstone chapters conclude here: the data-analysis pipeline (ch26), the multi-source research Team (ch27), and the customer-service knowledge-base service (this chapter). The entire course is complete — revisit the Skills system at Chapter 20, or return to the Course Guide to review any chapter at will. Safe travels on your Agent journey!

Chapter 28 · Capstone 3: Customer-Support Knowledge Base Agent on AgentOS ​

28.1 From Script to Service: The Production Gap Checklist ​

28.2 Building the Product Manual Knowledge Base ​

28.3 Support Agent: Retrieval-Augmented + Fallback + Source Annotations ​

28.4 Publishing as a FastAPI Service with AgentOS ​

28.5 Docker Packaging and Environment Variable Management ​

28.6 Pre-Launch Verification: Health Checks and Light Load Test ​

Chapter Summary ​

🧪 随堂测验

🛠️ Hands-On Exercises ​

Chapter 28 · Capstone 3: Customer-Support Knowledge Base Agent on AgentOS

28.1 From Script to Service: The Production Gap Checklist

28.2 Building the Product Manual Knowledge Base

28.3 Support Agent: Retrieval-Augmented + Fallback + Source Annotations

28.4 Publishing as a FastAPI Service with AgentOS

28.5 Docker Packaging and Environment Variable Management

28.6 Pre-Launch Verification: Health Checks and Light Load Test

Chapter Summary

🛠️ Hands-On Exercises