11. 开源 Agent 全家桶:Hermes 3 + 向量库的离线 RAG-Agent

上一篇我们有了本地能跑的 Hermes 3 Agent。但一个只能调用几个 mock 工具的 Agent 离"真正能用"还有距离。真实场景里,最常见的 Agent 形态是能查询你自己文档的 RAG-Agent:

你给它一堆 PDF、Markdown、Wiki 导出
它能根据用户问题检索相关片段,综合信息给答案
所有运行都在本地,数据不离开你的机器

这一篇把这套全本地 Agent 栈拼出来。完成后你会得到一个完整可用的企业内部助手的最小原型。

技术栈选型

模型层:Hermes 3 8B——上一篇讲过,工具调用稳、中文也能用、Ollama 一键起。 嵌入层:bge-m3——北智源的多语言嵌入模型,中英都强,支持 8192 token 的长文本,Ollama 可直接跑。 向量库:Chroma——最轻量的本地向量库,pip install 完事,不需要额外服务。 编排层:手写 Agent——不上 LangChain,让每一步可见、可改。

全部组件都是开源,全部在本地运行。总依赖不超过 5 个 Python 包。

环境准备

ollama pull hermes3:8b
ollama pull bge-m3

pip install openai chromadb pypdf markdown-it-py

pypdf 用来解析 PDF,markdown-it-py 用来处理 Markdown。根据你的文档类型,换成 python-docx、beautifulsoup4 等即可。

文档处理:从文件到向量

RAG 的第一步永远是把文档切成合适的块。块太大,检索时的上下文精度下降;块太小,单个块信息不完整。经验值是 500~800 token 每块,重叠 50~100 token。

from pathlib import Path
from pypdf import PdfReader

def read_pdf(path: Path) -> str:
    reader = PdfReader(path)
    return "\n\n".join(page.extract_text() or "" for page in reader.pages)

def read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8")

LOADERS = {
    ".pdf": read_pdf,
    ".md": read_text,
    ".txt": read_text,
}

def load_documents(doc_dir: str):
    docs = []
    for path in Path(doc_dir).rglob("*"):
        if path.suffix.lower() in LOADERS:
            text = LOADERS[path.suffix.lower()](path)
            docs.append({"path": str(path), "text": text})
    return docs

def chunk_text(text: str, size: int = 600, overlap: int = 80):
    chunks = []
    i = 0
    while i < len(text):
        chunks.append(text[i:i+size])
        i += size - overlap
    return chunks

按字符切是简单粗暴的做法。更讲究的做法是按段落边界切——遇到 \n\n 就尝试在那里切开。对 Markdown 文档,可以按 heading 切,保留文档的逻辑结构。但作为 MVP,按字符切已经够用。

用 bge-m3 生成嵌入

from openai import OpenAI

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

def embed(texts: list[str]) -> list[list[float]]:
    resp = client.embeddings.create(model="bge-m3", input=texts)
    return [item.embedding for item in resp.data]

bge-m3 通过 Ollama 的 embeddings 端点暴露。一次可以批量嵌入多个文本,比单次快。bge-m3 输出 1024 维向量,适合存 Chroma。

索引和检索

import chromadb

client_db = chromadb.PersistentClient(path="./rag_db")
collection = client_db.get_or_create_collection(
    name="docs",
    metadata={"hnsw:space": "cosine"},
)

def index_documents(doc_dir: str):
    docs = load_documents(doc_dir)
    for doc in docs:
        chunks = chunk_text(doc["text"])
        if not chunks:
            continue
        ids = [f"{doc['path']}#{i}" for i in range(len(chunks))]
        embeddings = embed(chunks)
        metadatas = [{"path": doc["path"], "chunk_idx": i} for i in range(len(chunks))]
        collection.add(
            ids=ids,
            documents=chunks,
            embeddings=embeddings,
            metadatas=metadatas,
        )
    print(f"indexed {collection.count()} chunks")

def retrieve(query: str, k: int = 5) -> list[dict]:
    q_emb = embed([query])[0]
    res = collection.query(query_embeddings=[q_emb], n_results=k)
    return [
        {"text": doc, "path": meta["path"], "score": 1 - dist}
        for doc, meta, dist in zip(res["documents"][0], res["metadatas"][0], res["distances"][0])
    ]

retrieve 返回前 K 个最相关的 chunk,附带来源路径。这是 Agent 后面要用的原材料。

把检索做成工具

真正有意思的部分来了。不要把检索做成"预先注入 prompt",而是做成 Agent 可以调用的工具。

区别是什么?传统 RAG 的做法是:用户提问 → 检索 top-K → 全部塞进 prompt → 模型答题。这叫 naive RAG,简单但有几个短板:

单次检索,命中率不高时就凉了
不能做多轮追加检索(用户问"A 和 B 哪个更好",要分别查 A 和 B)
模型不能判断"这个问题根本不需要查文档"

把检索做成工具后,模型自己决定"是否检索、用什么 query 检索、是否要多次检索":

def rag_search(query: str, k: int = 5) -> str:
    """Agent 可调用的检索工具"""
    results = retrieve(query, k=k)
    if not results:
        return "没有找到相关内容"
    parts = []
    for i, r in enumerate(results):
        parts.append(f"[{i+1}] 来源: {r['path']} (相关度 {r['score']:.2f})\n{r['text']}")
    return "\n\n---\n\n".join(parts)

TOOL_SCHEMAS = [{
    "type": "function",
    "function": {
        "name": "rag_search",
        "description": "在企业知识库中搜索相关文档片段。适合需要查内部资料、规章制度、项目文档的问题。",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "搜索关键词或问题"},
                "k": {"type": "integer", "default": 5, "description": "返回多少个片段"},
            },
            "required": ["query"],
        },
    },
}]

TOOLS = {"rag_search": rag_search}

关键在 description——告诉模型什么时候该用这个工具。写得好,模型在遇到"介绍一下我们公司的 OKR 制度"时会主动调用;在遇到"1+1 等于多少"时会直接回答,不浪费一次检索。

Agent 主循环

拼装起来:

def run_rag_agent(question: str, max_steps: int = 6):
    messages = [
        {"role": "system", "content": """你是一个企业知识助手。
对于任何涉及公司政策、内部流程、项目资料的问题,你应该先调用 rag_search 工具查询文档。
对于一般性常识、寒暄,直接回答不需要查文档。
回答时,如果引用了文档内容,要标注来源(即工具返回的路径)。"""},
        {"role": "user", "content": question},
    ]

    for _ in range(max_steps):
        resp = client.chat.completions.create(
            model="hermes3:8b",
            messages=messages,
            tools=TOOL_SCHEMAS,
            tool_choice="auto",
            temperature=0.0,
        )
        msg = resp.choices[0].message
        messages.append(msg.model_dump(exclude_none=True))

        if not msg.tool_calls:
            return msg.content

        for tc in msg.tool_calls:
            args = json.loads(tc.function.arguments)
            result = TOOLS[tc.function.name](**args)
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": str(result)[:3000],
            })

    return "超出最大步数"

# 使用
index_documents("./my_docs")
print(run_rag_agent("我们公司的报销流程是什么?"))

这一段串起来的整条链路:

Agent 收到问题
判断需要查文档,调 rag_search(query="报销流程")
检索 top-5 片段
Agent 看完觉得信息不够,再调一次 rag_search(query="报销金额限额")
两次信息综合,生成带来源标注的答案

这就是一个Agentic RAG。相比 naive RAG,它的优势在同一个问题里能做多次渐进检索,对复杂问题的回答质量高很多。

几个生产常见的扩展

这个骨架上可以叠的东西还有很多,每个都能独立写一篇:

混合检索——不只是向量检索,同时做 BM25(关键词)检索,把两者结果融合。对"精确匹配"类问题(代码、API 名、产品编号)特别重要。rank-bm25 或 chromadb 的 hybrid 搜索可用。

Reranker——检索召回 20 个,用一个 Reranker 模型重新排序取 top-5。Cohere 的 Rerank 或 bge-reranker-v2-m3 都是开源选项。检索精度提升明显,代价是多一次推理。

结构化过滤——按文档类型、时间、作者过滤。Chroma 支持 where 子句,在 metadata 上做精确过滤,结合向量检索能大幅缩小搜索空间。

查询改写——用户的问题经常含糊,让 Agent 先把问题改写成更适合检索的形式(expansion、分解)再检索。LangChain 里叫 MultiQuery。

引用校验——Agent 输出答案后,再跑一遍 LLM,检查"这个答案里的每个事实,是不是真的能在检索的文档里找到"。hallucination 显著下降。

这些功能叠上去,你就从 MVP 走到了企业级 RAG-Agent。重点是先跑通最简版本,再按真实需求加——不要一上来就堆十个组件。

和闭源 RAG 的真实差距

实话实说,一个 hermes3:8b + bge-m3 的本地 RAG,和用 GPT-4 + OpenAI embedding 的 RAG,在回答质量上有差距。主要体现在:

综合和推理——GPT-4 从多个文档片段里综合出非显性的答案的能力更强
refusal 的尺度——闭源模型更"知道什么时候说不知道",开源模型容易强答

但在简单事实问答上差距很小,对大多数"帮我查公司制度"这类需求够用。而且:数据不出你的机器、没有 API 限流、单次成本接近零、可以自己微调模型让它更懂你的文档。

真实的生产决策里,很多团队选开源 RAG 不是因为它质量高,而是因为数据合规是硬要求。有了 Hermes 3 这类模型后,这个选择不再意味着"质量妥协",只是"质量略有差距"。

小结

这一篇把前十篇讲过的东西汇到一起:本地模型、工具调用、上下文管理、检索,组成了一个能跑的 Agentic RAG。你应该能感觉到Agent 工程本质上是"组件拼装"——每个零件都不难,难的是拼装得有效率、有可观测性、有可靠性。

下一篇回到 Agent 架构,讲多 Agent 协作——Orchestrator-Worker、Debate、Handoff,以及第一篇开头就提过的那个反直觉事实:多 Agent 在很多时候不是更强,反而更弱。怎么避开那些坑,怎么判断什么时候多 Agent 真的值得。