kb_qa · 知识库问答(RAG 全流程)

本页收录可运行示例，配套本文档「指南 02」(指南 02)。 代码为随框架交付的示例，已一并收录到本站供直接对照使用。

说明与运行

示例: 知识库问答(KB 检索 → LLM 上下文 → 回答). 演示 retrieval 能力包全流程: 1. 导入: 文档拆分 → embedding → 入库(MemoryVectorStore) 2. 检索: KnowledgeBaseSearch 节点检索 3. 回答: 检索文本注入 ChatLLM context 运行: cd agentframework PYTHONPATH=src python examples/kb_qa.py

运行前需在已安装 agentframework 的 Python ≥ 3.11 环境；示例多为自带本地模型，无需真实 API Key；需要外部依赖（如 chromadb / 数据库驱动）会在说明中注明。

完整源码

"""示例: 知识库问答(KB 检索 → LLM 上下文 → 回答).

演示 retrieval 能力包全流程:
1. 导入: 文档拆分 → embedding → 入库(MemoryVectorStore)
2. 检索: KnowledgeBaseSearch 节点检索
3. 回答: 检索文本注入 ChatLLM context

运行:
    cd agentframework
    PYTHONPATH=src python examples/kb_qa.py
"""

from agentframework import Workflow, serve
from agentframework.nodes import Answer, ChatLLM, KnowledgeBaseSearch
from agentframework.retrieval import (
    HashEmbedder,
    MemoryVectorStore,
    RecursiveCharacterSplitter,
    create_retriever,
    ingest_document,
)

# 示例知识库内容
KB_TEXT = """FastGPT 是一个 AI Agent 构建平台。

它通过 Flow 提供开箱即用的数据处理、模型调用能力和可视化工作流编排。

FastGPT 支持多种知识库检索方式，包括向量检索、混合检索和 Rerank 深度检索。

FastGPT 使用 NextJS 构建，后端基于 MongoDB 和 PostgreSQL 向量数据库。"""


class _EchoModel:
    """演示模型: 从参考上下文消息中提取检索到的内容."""

    def invoke(self, messages):
        for m in messages:
            content = str(getattr(m, "content", ""))
            if content.startswith("以下是参考资料"):
                # 提取参考内容的最后一段引用行
                lines = [ln for ln in content.split("\n") if ln.strip()]
                return f"根据知识库: {lines[-1].strip()[:40]}"
        return "未找到参考资料"

    def stream(self, messages):
        from langchain_core.messages import AIMessage

        yield AIMessage(content=self.invoke(messages))


def build_kb_qa() -> Workflow:
    """构建知识库问答工作流."""
    # 1. 初始化存储 + embedding + 导入
    store = MemoryVectorStore()
    embedder = HashEmbedder(dimension=32)
    ingest_document(
        KB_TEXT,
        store=store,
        embedder=embedder,
        source="fastgpt-manual",
        splitter=RecursiveCharacterSplitter(chunk_size=50, chunk_overlap=10),
    )
    print(f"知识库已导入: {store.count()} 个块")

    retriever = create_retriever(store, embedder, top_k=2)

    # 2. 编排工作流
    wf = Workflow(name="kb_qa")
    wf.add_node(
        KnowledgeBaseSearch(
            name="kb",
            retriever=retriever,
        )
    )
    wf.add_node(
        ChatLLM(
            name="llm",
            chat_model=_EchoModel(),
            system="你是知识库问答助手, 请基于参考资料回答",
            context=wf.ref("kb.text"),
            output=False,  # 严格收口: 由 Answer 收口(骨架设计 5.2)
        )
    )
    wf.add_node(Answer(name="ans", text=wf.ref("llm.reply")))
    wf.add_edge("kb", "llm")
    wf.add_edge("llm", "ans")
    return wf


if __name__ == "__main__":
    wf = build_kb_qa()
    app = wf.compile()
    print("已编译工作流:", wf.name)
    serve({"app-kb": app})

---

回到示例索引。