Bonus-2 · RAG / Agent 全景实战手册

import chromadb
from openai import OpenAI
import time

DOCS = [
    {"id": "k8s-pod",   "text": "Pod 是 K8s 最小调度单元,包含 1 个或多个容器共享 network namespace 和 volume,生命周期一致。"},
    {"id": "k8s-deploy","text": "Deployment 通过 ReplicaSet 间接管理 Pod 副本数,支持滚动升级(RollingUpdate)和 Recreate 两种策略。"},
    {"id": "k8s-svc",   "text": "Service 通过 selector 把 Pod 抽象成稳定的 ClusterIP / NodePort / LoadBalancer 服务,后端 Pod 死掉自动剔除。"},
    {"id": "k8s-pvc",   "text": "PVC 是用户对存储的声明式申请,绑定 PV 后通过 CSI 驱动挂载到 Pod。"},
    {"id": "k8s-hpa",   "text": "HPA 基于 metrics-server 提供的 CPU/Memory 或 custom metric 自动扩缩 Deployment 副本数。"},
    {"id": "vllm-pa",   "text": "vLLM 的 PagedAttention 把 KV cache 分页管理,显存碎片减少 4×,30 并发请求下吞吐提升 24×。"},
    {"id": "vllm-cb",   "text": "Continuous Batching 让 vLLM 在 iteration level 调度,新请求随时加入,不需要等 batch 凑齐。"},
    {"id": "mig",       "text": "NVIDIA MIG (Multi-Instance GPU) 在 A100/A800/H100 上把单卡硬件切成最多 7 个独立 instance,显存和 SM 都隔离。"},
]

client = chromadb.PersistentClient(path="/opt/bonus2/chroma_data")
collection = client.get_or_create_collection(name="k8s_facts")
collection.upsert(ids=[d["id"] for d in DOCS], documents=[d["text"] for d in DOCS])

llm = OpenAI(base_url="http://10.43.165.182:8000/v1", api_key="not-needed")

def rag_chat(question, top_k=3):
    res = collection.query(query_texts=[question], n_results=top_k)
    docs = res["documents"][0]; ids = res["ids"][0]
    context = "\n".join(f"[{i+1}] ({did}) {d}" for i, (did, d) in enumerate(zip(ids, docs)))
    prompt = f"请基于以下知识回答(末尾标注源 [1][2][3]):\n知识:\n{context}\n\n问题: {question}\n\n答复:"
    t0 = time.time()
    resp = llm.chat.completions.create(
        model="qwen2.5-3b",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=200, temperature=0.3,
    )
    return resp.choices[0].message.content, time.time()-t0, ids

def tool_calculator(expr):     # safe eval, 限制在 math 模块
def tool_datetime(_):          # 当前时间
def tool_kb_lookup(key):       # 小型 KB: vllm / mig / hpa
TOOLS = {"calculator": ..., "datetime": ..., "kb_lookup": ...}

ACTION_RE = re.compile(r"Action:\s*(\w+)\(([^)]*)\)")
FINAL_RE  = re.compile(r"Final Answer:\s*(.+)", re.DOTALL)

while step < max_steps:
    resp = llm.chat.completions.create(..., stop=["Observation:"])
    out = resp.choices[0].message.content
    if FINAL_RE.search(out): return final
    m = ACTION_RE.search(out)
    obs = TOOLS[m.group(1)](m.group(2))
    messages.append({"role": "assistant", "content": out})
    messages.append({"role": "user",      "content": f"Observation: {obs}"})