综合实战：知识库问答智能体

把前面几章攒的能力组装成第一个完整项目：基于本地 Markdown/文本文档的知识库问答 Agent——离线资料先分块入库，用户提问时由 Agent 检索相关片段、依据片段作答并标注来源。整体链路：数据准备（分块）→ 嵌入入库（向量库）→ 检索工具（top-k + 来源）→ LangGraph 问答 Agent（检索增强 + 引用 + 流式）→ FastAPI 暴露。项目按文件拆分，便于替换任意一环。

目录与主链路

kb_agent/
├── data/          # 放你要喂的 .md / .txt
├── embed.py       # 文本 -> 向量
├── ingest.py      # 读文档 -> 分块 -> 向量化 -> 落库
├── retriever.py   # 向量检索，返回 top-k + 来源
├── agent.py       # LangGraph：检索工具 + 带引用回答
└── app.py         # FastAPI /chat，直接套第 28 章模板

embed.py：向量化（沿用第 13 章约定）

# embed.py
import os, zlib

def _debug(text: str, dim: int = 256) -> list:
    """离线演示用伪向量：无语义、仅验证流程，正式环境勿用。"""
    v = [0.0] * dim
    for i in range(len(text) - 1):
        v[zlib.crc32(text[i:i + 2].encode("utf-8")) % dim] += 1.0
    norm = sum(x * x for x in v) ** 0.5 or 1.0
    return [x / norm for x in v]

def get_embedding(text: str) -> list:
    mode = os.environ.get("EMBED_MODE", "api")     # api / debug
    if mode == "debug":
        return _debug(text)
    from openai import OpenAI                      # OpenAI 兼容 embedding 接口
    client = OpenAI(api_key=os.environ.get("EMBED_API_KEY", os.environ.get("LLM_API_KEY")),
                    base_url=os.environ.get("EMBED_BASE_URL", os.environ.get("LLM_BASE_URL")))
    return client.embeddings.create(
        model=os.environ.get("EMBED_MODEL", "text-embedding-3-small"),
        input=text).data[0].embedding

注意：deepseek 等纯对话服务商不提供 embedding，演示用 EMBED_MODE=debug，正式接入通义/OpenAI/Ollama 的 embedding 即可，代码不变。

ingest.py：md/txt 分块并入库

# ingest.py —— 用法：EMBED_MODE=debug python ingest.py
import glob, json, os, pathlib
from embed import get_embedding

def chunk_md(text: str, max_len: int = 300, overlap: int = 30) -> list:
    """极简分块：先按 '#' 标题拆，再把大段按字符窗口切并留重叠。"""
    parts, buf = [], ""
    for line in text.splitlines():
        buf += line + "\n"
        if line.startswith("#") and buf.strip() and len(buf) > max_len // 2:
            parts.append(buf); buf = ""
    parts.append(buf)
    chunks = []
    for p in parts:
        p = p.strip()
        while len(p) > max_len:
            chunks.append(p[:max_len]); p = p[max_len - overlap:]
        if p:
            chunks.append(p)
    return chunks

store_path = "kb_store.json"

def ingest(doc_dir: str = "data") -> None:
    store = json.load(open(store_path, encoding="utf-8")) if os.path.exists(store_path) else []
    have = {s["id"] for s in store}
    for fp in glob.glob(os.path.join(doc_dir, "*.md")) + glob.glob(os.path.join(doc_dir, "*.txt")):
        text = pathlib.Path(fp).read_text(encoding="utf-8")
        for i, chunk in enumerate(chunk_md(text)):
            sid = f"{os.path.basename(fp)}#{i}"
            if sid in have:
                continue
            store.append({"id": sid, "source": os.path.basename(fp),
                          "text": chunk, "vec": get_embedding(chunk)})
    json.dump(store, open(store_path, "w", encoding="utf-8"), ensure_ascii=False)
    print(f"入库完成，共 {len(store)} 个片段")

分块是 RAG 质量的第一道关卡：生产建议用 RecursiveCharacterTextSplitter 之类按语义切分，块长 300~800 字、保留重叠，太长召回了"半篇文章"、太短信息割裂。

retriever.py：检索工具（top-k + 来源）

# retriever.py
import json
from embed import get_embedding

def _load():
    return json.load(open("kb_store.json", encoding="utf-8"))

def _cos(a, b):
    return (sum(x * y for x, y in zip(a, b))
            / ((sum(x*x for x in a) ** 0.5 or 1.0) * (sum(y*y for y in b) ** 0.5 or 1.0)))

def retrieve(query: str, top_k: int = 3) -> list:
    qv = get_embedding(query)
    hits = sorted(((_cos(qv, s["vec"]), s) for s in _load()),
                  key=lambda t: t[0], reverse=True)[:top_k]
    return [{"score": round(sc, 3), "source": s["source"], "text": s["text"]}
            for sc, s in hits]

agent.py：LangGraph 检索问答（带引用 + 流式）

第 18 章的结构直接复用：把检索包成工具，让模型按需调用；系统提示里强制"只依据检索结果回答并给出来源文件名"：

# agent.py —— 结构同第 18 章，仅换工具与提示词
import os
from typing import Annotated, TypedDict
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from retriever import retrieve

SYSTEM = ("你是知识库问答助手。只依据检索到的片段回答；引用格式【文件名】；"
          "检索不到就明说不知道，禁止编造。")

@tool
def search_kb(query: str) -> str:
    """从本地知识库检索与 query 相关的片段，返回带相似度与来源的文本。"""
    return "\n".join(f"[{h['score']}|{h['source']}] {h['text']}"
                     for h in retrieve(query, top_k=3))

model = ChatOpenAI(model=os.getenv("LLM_MODEL", "deepseek-chat"),
                   api_key=os.getenv("LLM_API_KEY"),
                   base_url=os.getenv("LLM_BASE_URL")).bind_tools([search_kb])

class State(TypedDict):
    messages: Annotated[list, add_messages]

def agent(state):
    return {"messages": [model.invoke(state["messages"])]}

def should_continue(state):
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else END

builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode([search_kb]))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
builder.add_edge("tools", "agent")
graph = builder.compile()          # 与第 28 章同样的 invoke / stream 用法

def ask(question: str) -> str:
    out = graph.invoke({"messages": [SystemMessage(SYSTEM), HumanMessage(question)]})
    return next((m.content for m in reversed(out["messages"]) if m.content), "")

app.py：暴露成服务

与第 28 章完全一致，仅把 import 换成 from agent import graph，/chat 用 invoke、/chat/stream 把 graph.stream(stream_mode="messages") 逐 token 包成 SSE——重复样板不再贴出，完整代码按本章各文件拼装即可。

运行效果示例

# 1) 喂文档（把瑞林工具站帮助文档放进 data/）
EMBED_MODE=debug python ingest.py
# 输出：入库完成，共 23 个片段

# 2) 提问（模型需能联网；先本地验证检索）
python -c "import retriever; [print(h) for h in retriever.retrieve('怎么充值会员', 3)]"
# 输出：[{'score': 0.31, 'source': 'help.md', 'text': '充值会员：…'}, …]（debug 得分仅供参考）

# 3) 问答
python -c "from agent import ask; print(ask('会员能退款吗？'))"
# 输出（示意）：按知识库说明，会员开通后不支持退款（见【help.md】）。

小结：知识库问答智能体 = 分块(ingest) + 向量入库 + top-k 检索工具 + LangGraph"检索-作答-引用"循环 + FastAPI 出口；每层独立成文件，换 embedding、换向量库（chromadb/FAISS）、换文档源都只动对应文件——这也是综合项目应有的拆分方式。