智能软件工程 · 案例实战：AI 原生架构 + RAG 检索增强（组件边界 / 数据流 / 最小工作流）

为什么讲架构 + RAG：前面 19 篇里，Agent 侧靠 pr03、模型侧靠 pr11/12 都有落点，但「系统怎么分层、边界在哪、知识怎么进模型」一直没被系统回答。本篇补齐两块：① AI 原生架构的组件边界与数据流；② 常被漏讲的 RAG 检索增强，并为统一 traces.csv 新增 retrieval_hit 与 citation 两列。

阅读路径：§2 架构分层与边界，§3 质量属性权衡，§4 组件职责划分，§5 RAG 流水线与失败模式，§6 Agent 工作流与状态管理，§7 端到端数据流，§8 可运行的最小 RAG + 最小工作流代码，§9 度量与失败归因。

1. 案例覆盖的知识要点

一句话结论：核心落点是架构分层 + AI 原生设计；RAG 是本篇系统补齐的一块，它同时喂给 pr08 度量、pr09 失败、pr11-12 模型层。

2. 从传统架构到 AI 原生架构：六层与边界

传统软件的分层（表现层 / 业务逻辑 / 数据访问）默认「逻辑是确定的、数据是结构化的」。AI 原生系统要在这之上/之间插入三个新层：编排（Agent/工作流）、模型（推理能力）、知识（外部记忆/RAG）。一图对照：

三条边界一句话记住：确定性逻辑留业务层（能写成代码就别交给模型）；能力交给模型层（语言理解、生成、判断）；可变事实交给知识层（会更新、要溯源的，放检索库而非塞进权重）。把事实硬编进 Prompt 或权重，正是维护性灾难与幻觉的源头。

3. AI 系统的质量属性：四条彼此拉扯的曲线

传统质量属性（性能、可用性、安全性）仍在，但 AI 系统多了几个必须联合考量、也最容易踩坑的量：

关键论断：AI 原生架构的质量属性不能各自最优化，它们构成一条帕累托前沿（与 pr02 §7、pr11 的「质量↔成本」一脉相承）。架构决策的本质，是为具体业务在前沿上选一个可辩护的工作点，并用度量证明它是权衡后的选择，而非偶然结果。

4. 组件职责划分：模型 / 工具 / 知识 / 业务

最容易犯的错，是「什么都丢给大模型」。给一张职责划分决策表——拿到一个需求，先问「它的性质」，再决定落在哪层：

反面教材：让 LLM「口算」折扣、让 Prompt 记价格表——前者违反「精确计算归工具」，后者违反「可变事实归知识」。这两类误置是幻觉与维护地狱的主要来源，也直接对应 §9 的失败编码。

与 pr16 呼应：MCP 是工具接入的「协议」，节点式与 Agent 自主是「编排范式」——工具层用不用 MCP，和编排层选节点式还是 Agent，是两个正交决策，别混为一谈。

5. RAG 检索增强：把系列缺的那块补上

前面 19 篇把模型、工具、Agent 都讲了，唯独没讲「外部知识怎么进模型」。RAG（Retrieval-Augmented Generation）就是 §4 里「可变事实归知识层」的具体实现：不重训模型，而是在生成前检索相关文档片段，塞进上下文让模型「看着资料回答」，并要求给出引用。

5.1 每一步的工程要点

• 切块（chunking）：太大→噪声多、检索钝；太小→语义碎。重叠窗口（overlap）常设 10–20%。切块策略变更 = 数据侧规格漂移，须触发重评（接 pr19 §6）。
• Embedding：向量模型也有版本与领域适配（pr13 选型）。换 Embedding ⇒ 整库需重建，是版本清单要锁的组件。
• 检索：向量 top-k + 元数据过滤（权限、时间、类目）。可混合 BM25 关键词，缓解纯向量的精确匹配短板。
• 重排（rerank）：用交叉编码器对 top-k 二次排序，质量↑、延迟↑——又一处帕累托旋钮。
• 拼 Prompt：注入检索块 + 明确「只用给定资料，缺信息就说不知道」，是把 GD 护栏落到 Prompt 层。
• 引用：输出须标注来自哪个块/文档，供 §9「引用正确率」度量与用户溯源。

5.2 RAG 失败模式（对齐 pr09 失败编码）

为此，统一 traces.csv（pr02/pr11/pr12 已建）在 RAG 场景新增两列：retrieval_hit（检到的块是否含正确答案，0/1）与 citation_valid（引用是否指向支撑答案的块）。它们让 R-F1（检索问题）与 R-F2/F3（生成问题）可被分别统计，而不是把「答错」笼统归给模型。

6. Agent 工作流与状态管理

RAG 解决「知识从哪来」，Agent 编排解决「控制流怎么走」。一个知识库问答 Agent 的最小循环 = pr03 的 ReAct 骨架，外加 RAG 检索这一步与状态维护：

状态管理要点：把「已检索块、已尝试动作、当前步数、预算余量」显式放进状态对象，而不是全塞进对话历史。显式状态才能做预算封顶、终止判定与失败回放（无限回环是 Agent 最常见的可靠性事故，对应 pr06 多智能体的成本失控）。

与 pr16 的关系：这张图若用 ComfyUI / 固定 DAG 表达，就是「节点式编排」；若由模型每步自主决定下一步，就是「Agent 自主编排」。同一控制流，两种范式——选型看可预测性 vs 灵活性，与 pr06「何时值得拆角色」的权衡逻辑同构。

7. 端到端数据流：一次问答穿越了哪些边界

架构案例的价值，在于把「一个请求」画成穿过六层的完整轨迹，才看得见边界。一次「客服工单助手」（承接 pr19）问「上月物流延误怎么赔」：

1. 业务层：鉴权、限流，记录 request_id，设 SLA 预算（→ trace 起始）。
2. 编排层：识别需检索，初始化状态（steps=0, budget=…）。
3. 知识层：query→Embedding→向量检索（+元数据过滤权限/时效）→重排→top-k 块。此处产出 retrieval_hit 的判定素材。
4. 工具层：如需精确金额，调用规则引擎/计算器（不让模型口算，§4）。
5. 模型层：拼「检索块 + 问题 + 护栏指令」→ 生成接地答案 + citation。引用列在此生成。
6. 记忆层：把本轮问答与检索摘要写入会话/情节记忆，供多轮。
7. 业务层：脱敏、落 traces.csv（含 retrieval_hit / citation_valid），返回。

8. 实操：架构图 + 最小 RAG 工作流（可跑）

两个实操串起来：先画一张智能软件系统架构图（用 §2 的六层模板交付），再做 Agent 架构设计与最小工作流——用下面这段最小可跑 RAG 起步（模型/向量库以你实际选型为准，接口对齐 OpenAI 兼容端点）：

import json, math, csv, requests, time

# ---- 依赖：pip install requests numpy  （向量库先用最朴素的内存余弦，便于讲清原理）----
import numpy as np

EMBED_SERVER = "http://127.0.0.1:8001/v1"     # 任一 OpenAI 兼容端点
EMB_MODEL, GEN_MODEL = "your-embedding", "your-llm"   # 以官方部署为准

def embed(texts):
    r = requests.post(f"{EMBED_SERVER}/embeddings",
                      json={"model": EMB_MODEL, "input": texts}, timeout=60)
    return np.array([d["embedding"] for d in r.json["data"]], dtype="float32")

def chunk(text, size=300, overlap=60):        # §5.1：带重叠的定长切块
    step = size - overlap
    return [text[i:i+size] for i in range(0, len(text), step)]

def build_index(docs):
    # docs: [{"id","text","meta":{...}}]；块级索引 + 元数据
    chunks = []
    for d in docs:
        for j, c in enumerate(chunk(d["text"])):
            chunks.append({"cid": f'{d["id"]}#{j}', "text": c, "meta": d["meta"]})
    chunks_emb = embed([c["text"] for c in chunks])
    chunks_emb /= np.linalg.norm(chunks_emb, axis=1, keepdims=True) + 1e-9
    return chunks, chunks_emb

def retrieve(q_emb, chunks, chunks_emb, k=4, allowed=None):
    sims = chunks_emb @ q_emb
    order = np.argsort(-sims)
    out = []
    for i in order:
        if allowed and not allowed(chunks[i]["meta"]):   # §5.2 R-F5：越权/过期过滤
            continue
        out.append((chunks[i], float(sims[i])))
        if len(out) >= k:
            break
    return out

def answer(question, context_blocks):
    ctx = "\n\n".join(f'[{b["cid"]}] {b["text"]}' for b in context_blocks)
    prompt = (f"仅依据下列资料回答；资料没有则回答『无法确定』；"
              f"每个结论标注来源块 id。\n资料:\n{ctx}\n问题:{question}")
    r = requests.post(f"{EMBED_SERVER}/chat/completions", json={
        "model": GEN_MODEL, "temperature": 0,
        "messages":[{"role":"user","content":prompt}]}, timeout=120)
    return r.json["choices"][0]["message"]["content"]

# ---- 跑一条并落 trace（含新增两列）----
docs = json.load(open("kb.json", encoding="utf-8"))
chunks, chunks_emb = build_index(docs)

q = "上月物流延误怎么赔？"
q_emb = embed([q])[0]; q_emb /= (np.linalg.norm(q_emb) + 1e-9)
hits = retrieve(q_emb, chunks, chunks_emb,
                allowed=lambda m: m.get("cat") == "policy" and not m.get("stale"))
blocks = [h[0] for h in hits]
ans = answer(q, blocks)

gold = "policy-2#0"        # 示意占位：人工标注「含正确依据的块」
retrieval_hit = int(any(b["cid"] == gold for b in blocks))   # R-F1 信号
citation_valid = int(gold in ans)                            # R-F3/F4 信号（简化）
row = dict(q=q, topk=[b["cid"] for b in blocks], answer=ans,
           retrieval_hit=retrieval_hit, citation_valid=citation_valid,
           latency_s=round(time.time, 3))
with open("traces.csv", "a", newline="", encoding="utf-8") as f:
    w = csv.DictWriter(f, fieldnames=row); w.writeheader; w.writerow(row)
print(ans)

def agent_answer(q, chunks, chunks_emb, budget=3):
    state = {"steps": 0, "blocks": [], "done": False}
    q_emb = embed([q])[0]; q_emb /= (np.linalg.norm(q_emb) + 1e-9)
    while not state["done"] and state["steps"] < budget:      # 显式预算封顶
        state["steps"] += 1
        hits = retrieve(q_emb, chunks, chunks_emb, k=3)
        state["blocks"] += [h[0] for h in hits]
        # gate：检索是否充分（这里用最朴素的启发式，可换 Judge）
        state["done"] = len(state["blocks"]) >= 3
    if not state["blocks"]:
        return "无法确定（检索为空，触发 R-F1）", state
    return answer(q, state["blocks"][:3]), state

诚实提示：代码里的向量库是示意用朴素内存余弦（几百条能跑，生产应换 FAISS / 专用向量库并加持久化）；allowed 过滤、gold 标注、citation_valid 判定都是简化占位，落地时按真实数据分布与权限模型重做。模型名、上下文长度、能力以你所选模型的官方模型卡为准（pr13）。

9. 度量与失败归因：把架构质量变成数字

RAG + Agent 架构的质量，落进 pr08 的三层仪表盘，并与 pr09 失败编码打通：

关键归因法：先看 retrieval_hit 拆分问题域——hit=0 是检索/知识层的锅（回去调切块、Embedding、top-k、过滤），hit=1 却答错是生成层的锅（回去调 Prompt、护栏、模型）。这一步把「答得不准」从一句主观抱怨，变成两条可分别优化的工程指标——正是本系列「把缺陷变成可统计对象」主线在架构层的落地。

一页纸总结：AI 原生架构 = 在业务层之上加「编排 / 模型 / 工具 / 知识 / 记忆」六层，边界口诀「事实进知识、认知进模型、精确计算与副作用进工具、流程进编排」；质量属性是可靠性/延迟/成本/可观测/可替换的帕累托权衡；RAG 把可变事实经「切块→Embedding→检索→重排→拼Prompt→生成→引用」接进模型，用 retrieval_hit / citation_valid 把失败拆成检索侧与生成侧；Agent 用显式状态 + 充分性 gate + 预算封顶做最小工作流。两个实操 = 六层架构图 + 问答 Agent 最小工作流。向下游供 pr08 度量、pr09 失败、pr19 规格漂移的数据源。