ai agent--- 后端概念补充：Docker Compose、ElasticSearch、IK、BM25等

在开发 `agent` 的时候遇到两个数据库：`mysql 和 milvus`。

如果说数据库是业务的基石，负责持久化存储原始业务数据，比如 MySQL 存用户信息，milvus存储公司文档。核心要求是稳健、不丢失。

那么数据库和业务代码之间就需要一个中间件去跑腿，把数据库里面的数据快速且安全地在两方之间进行搬运，用来弥补数据库和业务代码的不足。这个中间件就是Elasticsearch 和 Redis。

而 redis 专门做缓存、es 做全文检索、milvus 做语义检索、、bullmq 做消息队列，是用于专门的用途，各司其职、专精专用，它们不是原始数据，丢了也不影响数据完整性。

检索补足：MySQL 不擅长全文模糊搜索，我们就引入 Elasticsearch 专门做高性能检索
性能不足：核心数据库读写磁盘太慢，我们就用 Redis 这种内存级中间件来做高速缓存
异步补足：业务逻辑处理太耗时，我们就用 RabbitMQ 或 BullMQ 这类消息队列中间件来做任务缓冲和解耦。

一.Docker Compose是什么?

Docker Compose = 用一份配置文件，定义并管理一组 Docker 容器。

Docker：管单个容器
Docker Compose：管多个容器之间的关系

一个项目要运行起来需要数据库，后端服务，前端服务等诸多应用，在docker里面，如果你一个一个run 那么，他很容易出错。

Docker Compose 把这些需要启动的应用对应的端口号，位置等信息，全部写进一个文件里面进行管理。只要运行这个文件，系统里面对应的应用就能按照一定的顺序，挨个启动起来。

Docker Compose的好处

✅ 一键启动整个系统

✅ 自动创建网络（容器之间可直接用服务名通信）

✅ 管理启动顺序（depends_on）

✅ 数据持久化（Volume）

✅ 环境变量集中管理

✅ 本地开发 / 测试环境一致性

docker-compose.yml

在项目里面你创建一个文件：docker-compose.yml，写上类似下面的代码，你可以在package.json里面配置具体的运行命令，然后用npm运行即可。

<span>version</span>: <span>"3.9"</span>

<span>services</span>:
  <span>web</span>:
    <span>image</span>: my-<span>app</span>:latest
    <span>ports</span>:
      - <span>"8080:8080"</span>
    <span>depends_on</span>:
      - es

  <span>es</span>:
    <span>image</span>: docker.<span>elastic</span>.<span>co</span>/elasticsearch/<span>elasticsearch</span>:<span>8.13</span><span>.0</span>
    <span>environment</span>:
      - discovery.<span>type</span>=single-node
    <span>ports</span>:
      - <span>"9200:9200"</span>
    <span>volumes</span>:
      - <span>es_data</span>:<span>/usr/</span>share/elasticsearch/data

<span>volumes</span>:
  <span>es_data</span>:

为什么生产环境最终要上 Kubernetes？

核心原因一句话：Docker Compose 解决的是“怎么把多个容器跑起来”，Kubernetes 解决的是“怎么让这些容器在生产环境里长期、稳定、自动地跑下去”。

对比 docker 和 Kubernets之间的区别。

Kubernetes 持续做一件事：“让集群的实际状态，无限逼近你期望的状态。”

K8s 是 Kubernetes 的缩写

Docker Compose 是“把系统跑起来”，Kubernetes 是“让系统在生产环境里活着、活好、活稳”。当服务数量、流量、可用性要求超过单机能力时，Kubernetes 就是自然选择。

二.ElasticSearch是什么?

Elasticsearch（简称 ES）是一个开源的分布式搜索与分析引擎，最核心的能力是：让你在海量数据里快速全文检索、过滤、聚合统计，并且能水平扩展、高可用。它底层基于 Apache Lucene，但把 Lucene 的复杂性包装成了易用的 REST API。

ES 是 Elasticsearch 的缩写，说白了他就是一个独立的，端口号是 9200 的 Web 服务。nodejs业务代码要想用它就发送htpp请求就好了。

Elasticsearch = 分布式的、实时的、支持全文检索的数据存储 + 分析系统。

1.Elasticsearch的特点

像 数据库（能存数据、能查数据）
但更擅长 模糊搜索、全文检索、相关性排序
同时又像一个 实时分析引擎（聚合、统计、监控）

2.Elasticsearch 能做什么

由于普通 MySQL 使用的是正向索引：以一行为单位存储完整数据，检索文本内容时，需要逐行遍历、逐个字段匹配内容。数据量越大、文本越长，模糊 / 全文搜索就越慢，性能极差，并不适合大范围关键词检索。为了解决这个问题，就产生了 ES。

Elasticsearch 采用倒排索引机制，会自动对 text 类型字段进行分词处理，拆解为一个个独立词条，再以「词条」为核心，反向关联所有包含该词条的文档。

"倒排"是相对"正排"说的——正排是"按文档找词"（这是文档里有哪些词），倒排是"按词找文档"（这个词在哪些文档里）。

ES 是一个独立运行的服务，跟你的 Node.js等后端代码是彻底解耦的两回事。所以不管后端语言怎么换，都不影响ES的运行。

如何理解后端业务代码、ES、数据库之间的关系？

你把 ES 想成一个专门干搜索的"外包团队" ：

你（Node.js）只需要把资料交给它，再向它问问题
它内部怎么建目录、怎么归档、怎么查档案，你完全不用管
你们之间只有一个对接窗口（9200 端口的 HTTP 接口）

所以说，ES 装在 Docker 里独立运行，它产生的那一堆倒排索引、正排索引、原始 JSON，全部由 ES 自己管理，跟 Node.js 后端代码没有任何关系。 。

3.ES流水线组成部分

在ES流水线上：IK 分词器、倒排索引、BM25 是理解 ES 的"铁三角"，也是这条流水线上的三个工位：

举个例子：你在搜索框输入"如何训练猫咪" → IK 分词器把它切成"如何 / 训练 / 猫咪" → 拿着这三个词同时去倒排索引查目录，各自拿到一批文档，合并去重 → 拿到一批候选后，BM25 在收集的过程中就按"词频会腻、长文打折、稀有词值钱"给每条算分，分数高的先收 → 收够就停，最终按分数从高到低排给你。

上面说的“倒排索引查目录”这个目录是存数据的时候是 Elasticsearch 自己建的,那张表是在你存数据那一刻就生成好的，之后一直躺在磁盘上。当你存数据的时候，Elasticsearch就已经把这条信息的关键信息，搜录到他的目录里面了，目的就是为了你在搜索的时候能够快速搜出来。

案例流程图如下：

4.如何使用ES

4.1 安装docker，

4.2 拉取ES

docker pull <span>elasticsearch</span>:<span>8.15</span><span>.0</span>

4.3 运行ES

docker run -d -p <span>9200</span>:<span>9200</span> --name es \
  -e <span>"discovery.type=single-node"</span> \
  -e <span>"xpack.security.enabled=false"</span> \
  <span>elasticsearch</span>:<span>8.15</span><span>.0</span>

验证是否起来了

curl <span>localhost</span>:<span>9200</span>

4.4 安装 IK 插件

因为 ES 对中文的分词效果不好，他会将“连衣裙”分成“连”“衣”“裙”，一个汉字一组，所以用 IK 插件。

docker exec -it es bin/elasticsearch-plugin install \
  <span>https</span>:<span>//release.infinilabs.com/analysis-ik/stable/elasticsearch-analysis-ik-8.15.0.zip</span>
docker restart es

4.5创建索引

curl -X <span>PUT</span> <span>"localhost:9200/articles"</span> -H <span>"Content-Type: application/json"</span> -d <span>'{
  "mappings": {
    "properties": {
      "title":   { "type": "text", "analyzer": "ik_max_word" },
      "content": { "type": "text", "analyzer": "ik_max_word" },
      "author":  { "type": "keyword" }
    }
  }
}'</span>

4.6 在nodejs里面使用

npm install @elastic/elasticsearch

<span>import</span> { <span>Client</span> } <span>from</span> <span>'@elastic/elasticsearch'</span>

<span>const</span> es = <span>new</span> <span>Client</span>({
  <span>node</span>: <span>'http://localhost:9200'</span>,   <span>// ES 的地址</span>
  <span>// 如果 ES 设了密码：</span>
  <span>// auth: { username: 'elastic', password: 'xxx' }</span>
})

<span>await</span> es.<span>ping</span>()   <span>// 探活，能通就说明对接成功</span>

4.7 在nodejs里面进行增删改查操作

BM25是ES内部自己做的，当ES查询返回数据的时候，他就会把BM25返回的评分值返回出来。

<span>//添加</span>
<span>await</span> es.<span>index</span>({
  <span>index</span>: <span>'articles'</span>,
  <span>id</span>: <span>'1'</span>,                         <span>// 可选，不传 ES 自动生成</span>
  <span>document</span>: {
    <span>title</span>: <span>'如何训练猫咪'</span>,
    <span>content</span>: <span>'训练猫咪需要耐心，正向强化比惩罚更有效。'</span>,
    <span>author</span>: <span>'小明'</span>
  }
})

<span>await</span> es.<span>indices</span>.<span>refresh</span>({ <span>index</span>: <span>'articles'</span> })  <span>// 强制刷新，让刚存的能立刻搜到</span>

<span>//搜索</span>
<span>const</span> { hits } = <span>await</span> es.<span>search</span>({
  <span>index</span>: <span>'articles'</span>,
  <span>query</span>: {
    <span>match</span>: { <span>content</span>: <span>'如何训练猫咪'</span> }   <span>// 会自动分词，再去倒排表查</span>
  },
  <span>size</span>: <span>10</span>,                             <span>// 只要前 10 条</span>
  <span>highlight</span>: { <span>fields</span>: { <span>content</span>: {} } } <span>// 让关键词在结果里高亮</span>
})

hits.<span>hits</span>.<span>forEach</span>(<span><span>hit</span> =></span> {
  <span>console</span>.<span>log</span>(hit.<span>_score</span>, hit.<span>_source</span>.<span>title</span>)  <span>// _score 就是 BM25 算出来的分</span>
})

5.ES的数据存在哪里？

ES运行在docker里面，他的数据存在docker内部的/usr/share/elasticsearch/data

如果docker重启后，ES的数据会怎么样？

操作	数据是否保留	原因
`docker restart es`（重启）	✅ 不丢	同一个容器，内部文件系统原封不动
`docker stop` / `start`	✅ 不丢	同上
`docker rm es` 后重新 `run`	❌ 丢了	旧容器被删，新容器是全新的，data 目录跟着没了
宿主机重启	✅ 不丢	只要容器没被删除

ES 数据默认写在容器内 /usr/share/elasticsearch/data，容器重启不丢、删除就丢。生产环境必须用 **-v 卷名:/usr/share/elasticsearch/data** 挂个数据卷，把数据落到宿主机，这样哪怕容器整个删了重建，数据也还在。

三. 嵌入模型

在实际的大模型应用里面,RAG的语义检索并不好用，搜出来的结果差强人意，为了弥补这个不足，我们用ElasticSearch 关键词检索来配合，打出一套组合拳。这就是混合检索框架。

解释嵌入模型

首先需要搞清楚的是：老版本的milvus里面存储的是向量和文档id，不存储文档，文档是在mysql里面存储的。也就是说我通过余弦相似度拿到文档id以后，去mysql里面搜id对应的文档，然后把文档给大模型思考。

现在的新版本milvus3.0支持text类型，可以往里面存储文档，所以你的文档是可以放在text里面的，这样就少了一步通过id去mysql里面搜索文档的过程。

一般企业应用里面用的都是将文档存在mysql里面，这样的好处是减轻milvus的压力，方便文档更新。

在一般案例里面你可以将文档直接存到milvus里面，方便学习，但是当数据量很大的时候，向量数据库的体积增大，读写的开销就会变大。

嵌入模型是 Embedding，他是做语义理解的。

嵌入模型在大模型和milvus之间会用到两次,一次是保存数据的时候，一次是milvus查询的时候。

向milvus保存数据的步骤

1.将文件转化成纯文本

利用下面工具将各种形式的文件提取成纯文本。

pnpm add pdf-parse mammoth xlsx unzipper

<span>// document-parser.service.ts</span>
<span>import</span> { <span>Injectable</span>, <span>Logger</span> } <span>from</span> <span>'@nestjs/common'</span>;
<span>import</span> * <span>as</span> pdfParse <span>from</span> <span>'pdf-parse'</span>;
<span>import</span> * <span>as</span> mammoth <span>from</span> <span>'mammoth'</span>;
<span>import</span> * <span>as</span> xlsx <span>from</span> <span>'xlsx'</span>;

@<span>Injectable</span>()
<span>export</span> <span>class</span> <span>DocumentParserService</span> {
  private readonly logger = <span>new</span> <span>Logger</span>(<span>DocumentParserService</span>.<span>name</span>);

  <span>async</span> <span>parse</span>(<span>buffer</span>: <span>Buffer</span>, <span>mimeType</span>: string): <span>Promise</span><string> {
    <span>try</span> {
      <span>switch</span> (mimeType) {
        <span>case</span> <span>'application/pdf'</span>:
          <span>return</span> <span>this</span>.<span>parsePdf</span>(buffer);

        <span>case</span> <span>'application/vnd.openxmlformats-officedocument.wordprocessingml.document'</span>:
          <span>return</span> <span>this</span>.<span>parseDocx</span>(buffer);

        <span>case</span> <span>'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'</span>:
          <span>return</span> <span>this</span>.<span>parseXlsx</span>(buffer);

        <span>case</span> <span>'text/html'</span>:
          <span>return</span> <span>this</span>.<span>parseHtml</span>(buffer);

        <span>case</span> <span>'text/markdown'</span>:
        <span>case</span> <span>'text/plain'</span>:
          <span>return</span> buffer.<span>toString</span>(<span>'utf-8'</span>);

        <span>default</span>:
          <span>throw</span> <span>new</span> <span>Error</span>(<span>`不支持的文件类型: <span>${mimeType}</span>`</span>);
      }
    } <span>catch</span> (err) {
      <span>this</span>.<span>logger</span>.<span>error</span>(<span>`解析失败 [<span>${mimeType}</span>]: <span>${err.message}</span>`</span>);
      <span>throw</span> err;
    }
  }

  private <span>async</span> <span>parsePdf</span>(<span>buffer</span>: <span>Buffer</span>): <span>Promise</span><string> {
    <span>const</span> result = <span>await</span> <span>pdfParse</span>(buffer);
    <span>return</span> result.<span>text</span>;
  }

  private <span>async</span> <span>parseDocx</span>(<span>buffer</span>: <span>Buffer</span>): <span>Promise</span><string> {
    <span>const</span> { <span>value</span>: html } = <span>await</span> mammoth.<span>convertToHtml</span>({ buffer });
    <span>return</span> html
      .<span>replace</span>(<span>/<table[\s\S]*?<\/table>/g</span>, <span>'\n[表格]\n'</span>)
      .<span>replace</span>(<span>/<[^>]+>/g</span>, <span>'\n'</span>)
      .<span>replace</span>(<span>/\n{2,}/g</span>, <span>'\n'</span>)
      .<span>trim</span>();
  }

  private <span>parseXlsx</span>(<span>buffer</span>: <span>Buffer</span>): string {
    <span>const</span> wb = xlsx.<span>read</span>(buffer);
    <span>return</span> wb.<span>SheetNames</span>
      .<span>map</span>(<span><span>name</span> =></span> {
        <span>const</span> sheet = wb.<span>Sheets</span>[name];
        <span>return</span> <span>`=== <span>${name}</span> ===\n`</span> + xlsx.<span>utils</span>.<span>sheet_to_csv</span>(sheet);
      })
      .<span>join</span>(<span>'\n'</span>);
  }

  private <span>parseHtml</span>(<span>buffer</span>: <span>Buffer</span>): string {
    <span>return</span> buffer
      .<span>toString</span>(<span>'utf-8'</span>)
      .<span>replace</span>(<span>/<script[\s\S]*?<\/script>/g</span>, <span>''</span>)
      .<span>replace</span>(<span>/<style[\s\S]*?<\/style>/g</span>, <span>''</span>)
      .<span>replace</span>(<span>/<[^>]+>/g</span>, <span>'\n'</span>)
      .<span>replace</span>(<span>/&nbsp;/g</span>, <span>' '</span>)
      .<span>replace</span>(<span>/\s{2,}/g</span>, <span>' '</span>)
      .<span>trim</span>();
  }
}

2.切片

利用工具给文本切片，一般会切成500字一块，前后50字和上面切片重复，以免对不上号。

将ids更新到mysql的事情也是在这里做的。

pnpm add @langchain/textsplitters

<span>// chunker.service.ts</span>
<span>import</span> { <span>Injectable</span> } <span>from</span> <span>'@nestjs/common'</span>;
<span>import</span> { <span>RecursiveCharacterTextSplitter</span> } <span>from</span> <span>'@langchain/textsplitters'</span>;

@<span>Injectable</span>()
<span>export</span> <span>class</span> <span>ChunkerService</span> {
  <span>// 图上说的“500字一块，重叠50字”正好对应这两个参数</span>
  private readonly splitter = <span>new</span> <span>RecursiveCharacterTextSplitter</span>({
    <span>chunkSize</span>: <span>500</span>,      <span>// 每块目标字符数（≈字数）</span>
    <span>chunkOverlap</span>: <span>50</span>,    <span>// 相邻块重叠字符数，防止句子被劈断</span>
  });

  <span>async</span> <span>split</span>(<span>text</span>: string): <span>Promise</span><string[]> {
    <span>return</span> <span>this</span>.<span>splitter</span>.<span>splitText</span>(text);
  }
}

3.用嵌入模型生成向量

调嵌入模型的embedding_model.encode，将文本块生成向量值，id，还有章节信息，他们都是数组。

文章片段的权限，对应的章节等信息都是在这一节处理好，保存到milvus里面的。

<span>// 逐块向量化</span>
<span>const</span> vectors = [];
<span>for</span> (<span>const</span> chunk <span>of</span> chunks) {
  <span>const</span> vector = <span>await</span> embedding_model.<span>encode</span>(chunk.<span>text</span>);
  vectors.<span>push</span>(vector);
}

<span>// 组装 ids 与 payloads</span>
<span>const</span> ids = [];
<span>const</span> payloads = [];
<span>for</span> (<span>let</span> i = <span>0</span>; i < chunks.<span>length</span>; i++) {
  <span>const</span> chunk = chunks[i];

  ids.<span>push</span>(<span>`<span>${chunk.document_id}</span>_<span>${i}</span>`</span>);
  payloads.<span>push</span>({
    <span>document_id</span>: chunk.<span>document_id</span>, <span>// 自己带的</span>
    <span>page</span>:         chunk.<span>page</span>,        <span>// 解析时记下的</span>
    <span>chapter</span>:      chunk.<span>chapter</span>,    <span>// 解析时记下的</span>
    <span>updated_at</span>:   <span>now</span>(),            <span>// 代码现打的</span>
    <span>permission</span>:   <span>get_acl</span>(chunk),   <span>// 查权限系统</span>
  });
}

4.向 Milvus写入数据

第三步骤生成了三个数组，他们的下标都是一样的，所以直接塞进insert里面即可。

<span>// 写入 Milvus，</span>
milvus.<span>insert</span>([ids, vectors, payloads]);

他们会一组一组地进入数据。

查询milvus的步骤

我们还没有学习重排模型可以直接跳过。流程就变成了，用户输入问题，嵌入模型将问题转化成向量，利用milvus的api去向量数据库里面搜索对应的值。他会返回一组符合条件的 id 数组，我们根据id数据去mysql拿对应的文档。拿到文档都给大模型处理就好了

1.向量化问题

利用嵌入模型将问题向量化

<span>const</span> queryVector = <span>await</span> embedding_model.<span>encode</span>(<span>"2024年华东区销量冠军是哪款？"</span>);
<span>// 输出: [0.12, -0.45, 0.88, ...]  1536 维</span>

2.利用milvus的api搜索相似值

Milvus 里的真实动作：不是“搜索”，是“比距离”，利用相似性给出对应文档的id

<span>const</span> searchResult = <span>await</span> client.<span>search</span>({
  <span>collection_name</span>: <span>COLLECTION_NAME</span>,
  <span>data</span>: [queryVector],        <span>// 注意是数组，支持一次搜多个向量</span>
  <span>limit</span>: <span>20</span>,                  <span>// 只要前 20 条</span>
  <span>output_fields</span>: [<span>'document_id'</span>, <span>'page'</span>, <span>'chapter'</span>, <span>'permission'</span>],
  <span>filter</span>: <span>'permission == "public"'</span>,  <span>// 可以先过滤，比如只搜有权限的</span>
});

返回的数据如下：

searchResult = [
  {
    <span>id</span>: <span>"doc_001_17"</span>,      <span>// 命中的是 doc_001 的第 17 块</span>
    <span>score</span>: <span>0.87</span>,            <span>// 相似度分数，越接近 1 越像</span>
    <span>document_id</span>: <span>"doc_001"</span>,
    <span>page</span>: <span>23</span>,
    <span>chapter</span>: <span>"第三章 > 第二节"</span>,
    <span>permission</span>: <span>"public"</span>
  },
]

3.拿文档

<span>const</span> ids = searchResult.<span>map</span>(<span><span>r</span> =></span> r.<span>id</span>);

<span>// 去mysql里面拿文档</span>
<span>const</span> contents = <span>await</span> db.<span>query</span>(
  <span>'SELECT id, text, title FROM document_chunks WHERE id = ANY($1)'</span>,
  [ids]
);

四.重排模型

重排模型是什么

重排的意义是嵌入模型漏掉了正确答案，重排负责把它捡回来

嵌入模型（Bi-Encoder） ：问题和文档各走各的编码器，互不看对方，最后比距离
重排模型（Cross-Encoder） ：把问题跟文档拼成一条输入，一起送进同一个编码器，模型能注意到两者的每个词之间的关联

重排模型 = 把问题和每条候选拼一起精读打分。 它比嵌入模型准得多，但因为它必须逐条配对计算、所以慢且贵，只能放在嵌入模型粗筛之后当「最后一公里」用——把 Top 100 里的正确答案顶到 Top 5。

重排是拿着问题和嵌入模型搜出来的答案一个一个地再次核对，提高准确性。

重排模型是输入用户问题 + 一段文档，输出一个相关度分数

专门用来给 RAG 做去噪、筛选、重新排顺序，体量小、推理快、成本极低

为什么要用重排模型

混合召回（向量 + 关键词）会带来大量冗余信息
大模型上下文窗口有限，不能把所有文档都塞进去
噪声太多会让模型答非所问、逻辑混乱、幻觉增加
先过滤、再精简，才能让回答更精准

重排模型怎么用

<span>// 伪代码，以 Cohere / Jina 的 API 为例</span>
<span>async</span> <span>function</span> <span>rerank</span>(<span>query, candidates</span>) {
  <span>const</span> res = <span>await</span> <span>fetch</span>(<span>RERANK_ENDPOINT</span>, {
    <span>method</span>: <span>'POST'</span>,
    <span>headers</span>: { <span>'Content-Type'</span>: <span>'application/json'</span>, <span>'Authorization'</span>: <span>`Bearer <span>${API_KEY}</span>`</span> },
    <span>body</span>: <span>JSON</span>.<span>stringify</span>({
      <span>model</span>: <span>'bge-reranker-v2-m3'</span>,
      <span>query</span>: query,
      <span>documents</span>: candidates.<span>map</span>(<span><span>c</span> =></span> c.<span>text</span>),
      <span>top_n</span>: <span>5</span>,           <span>// 只留最相关的 5 条</span>
    }),
  });

  <span>const</span> data = <span>await</span> res.<span>json</span>();
  <span>// 返回：[{ index: 2, score: 0.93 }, { index: 0, score: 0.81 }, ...]</span>
  <span>return</span> data.<span>results</span>
    .<span>sort</span>(<span>(<span>a, b</span>) =></span> b.<span>score</span> - a.<span>score</span>)
    .<span>map</span>(<span><span>r</span> =></span> candidates[r.<span>index</span>]);  <span>// 按新分数重排</span>
}

<span>// 在检索链里的位置</span>
<span>const</span> queryVector = <span>await</span> embedder.<span>encode</span>(query);                    <span>// ① 问题向量化</span>
<span>const</span> candidates  = <span>await</span> milvus.<span>search</span>(queryVector, { <span>limit</span>: <span>50</span> }); <span>// ② 粗筛 50 条</span>
<span>const</span> contents    = <span>await</span> db.<span>findMany</span>(candidates.<span>map</span>(<span><span>c</span> =></span> c.<span>id</span>));   <span>// ③ ID 反查原文</span>
<span>const</span> top5        = <span>await</span> <span>rerank</span>(query, contents);                  <span>// ④ 精排</span>
<span>const</span> answer      = <span>await</span> llm.<span>generate</span>(query, top5);                <span>// ⑤ 生成</span>

那嵌入模型给多少条数据，要重排模型重排后答案最合理呢？

常见重排模型怎么选

bge-reranker-v2-m3：BAAI 出品，多语言（含中文），效果好，可本地部署
jina-reranker-v2：Jina AI，支持长文本
Cohere Rerank：API 调用，省事但要付费
GTE-Reranker：阿里达摩院，中文场景表现不错

千问重排

获取apiKey

一般我们用阿里的重排模型：bailian.console.aliyun.com/cn-beijing/…

点击重排模型，然后下拉点击获取apiKey

curl测试

利用官网里面给的curl测试命令，测试通不通，在Windows shell里面运行。将命令行里面“你的apikey”替换成你实际的apikey


curl.<span>exe</span> --location <span>'https://ws-klbc6qsfh6f1w9k1.cn-beijing.maas.aliyuncs.com/api/v1/services/rerank/text-rerank/text-rerank'</span> <span>`
--header "Authorization: Bearer 你的apikey " `</span>
--header <span>'Content-Type: application/json'</span> <span>`
--data '{
    "model": "qwen3.7-text-rerank",
    "input": {
        "query": "什么是文本排序模型",
        "documents": [
            "文本排序模型广泛用于搜索引擎和推荐系统中，它们根据文本相关性对候选文本进行排序",
            "量子计算是计算科学的一个前沿领域",
            "预训练语言模型的发展给文本排序模型带来了新的进展"
        ]
    },
    "parameters": {
        "top_n": 5,
        "return_documents": true
    }
}'

</span>

nodejs测试

安装包

pnpm install @langchain/core @langchain/openai @langchain/langgraph @langchain/community dotenv zod @zilliz/milvus2-sdk-node

.env配置

<span>OPENAI_API_KEY</span>=sk-xx
<span>OPENAI_BASE_URL</span>=<span>https</span>:<span>//dashscope.aliyuncs.com/compatible-mode/v1</span>
<span>RERANK_URL</span>=<span>https</span>:<span>//dashscope.aliyuncs.com/api/v1/services/rerank/text-rerank/text-rerank</span>
<span>MODEL_NAME</span>=qwen-plus
<span>RERANK_MODEL</span>=qwen3-rerank

重排

封装重排的方法

<span>import</span> <span>"dotenv/config"</span>;
<span>import</span> { <span>BaseDocumentCompressor</span> } <span>from</span> <span>"@langchain/core/retrievers/document_compressors"</span>;

<span>export</span> <span>class</span> <span>DashScopeRerank</span> <span>extends</span> <span>BaseDocumentCompressor</span> {

  <span>constructor</span>(<span>{ apiKey, model = <span>"qwen3-rerank"</span>, topN = <span>3</span>, baseUrl } = {}</span>) {
    <span>super</span>();
    <span>this</span>.<span>apiKey</span> = apiKey;
    <span>this</span>.<span>model</span> = model;
    <span>this</span>.<span>topN</span> = topN;
    <span>this</span>.<span>baseUrl</span> = baseUrl ?? process.<span>env</span>.<span>RERANK_URL</span>;
  }

  <span>async</span> <span>compressDocuments</span>(<span>documents, query, _callbacks</span>) {
    <span>const</span> res = <span>await</span> <span>fetch</span>(<span>this</span>.<span>baseUrl</span>, {
      <span>method</span>: <span>"POST"</span>,
      <span>headers</span>: {
        <span>Authorization</span>: <span>`Bearer <span>${<span>this</span>.apiKey}</span>`</span>,
        <span>"Content-Type"</span>: <span>"application/json"</span>,
      },
      <span>body</span>: <span>JSON</span>.<span>stringify</span>({
        <span>model</span>: <span>this</span>.<span>model</span>,
        <span>input</span>: {
          query,
          <span>documents</span>: documents.<span>map</span>(<span>(<span>d</span>) =></span> d.<span>pageContent</span>),
        },
        <span>parameters</span>: {
          <span>return_documents</span>: <span>false</span>,
          <span>top_n</span>: <span>this</span>.<span>topN</span>,
        },
      }),
    });

    <span>const</span> json = <span>await</span> res.<span>json</span>();
    <span>if</span> (!res.<span>ok</span>) {
      <span>throw</span> <span>new</span> <span>Error</span>(
        <span>`DashScope rerank <span>${res.status}</span>: <span>${<span>JSON</span>.stringify(json)}</span>`</span>,
      );
    }

    <span>const</span> results = json?.<span>output</span>?.<span>results</span>;
    <span>if</span> (!<span>Array</span>.<span>isArray</span>(results)) {
      <span>throw</span> <span>new</span> <span>Error</span>(<span>`unexpected rerank response: <span>${<span>JSON</span>.stringify(json)}</span>`</span>);
    }

    <span>return</span> results.<span>map</span>(<span>(<span>item</span>) =></span> documents[item.<span>index</span>]);
  }
}

测试重排

<span>import</span> <span>"dotenv/config"</span>;
<span>import</span> { <span>Document</span> } <span>from</span> <span>"@langchain/core/documents"</span>;
<span>import</span> { <span>DashScopeRerank</span> } <span>from</span> <span>"./rerank.mjs"</span>;

<span>async</span> <span>function</span> <span>main</span>(<span></span>) {
    <span>const</span> apiKey = process.<span>env</span>.<span>OPENAI_API_KEY</span>;

    <span>const</span> compressor = <span>new</span> <span>DashScopeRerank</span>({ apiKey, <span>topN</span>: <span>3</span> });

    <span>const</span> query = <span>"什么是文本排序模型"</span>;
    <span>const</span> docs = [
        <span>new</span> <span>Document</span>({
            <span>pageContent</span>:
                <span>"预训练语言模型的发展给文本排序模型带来了新的进展"</span>,
        }),
        <span>new</span> <span>Document</span>({
            <span>pageContent</span>: <span>"量子计算是计算科学的一个前沿领域"</span>,
        }),
        <span>new</span> <span>Document</span>({
            <span>pageContent</span>: <span>"文本排序模型广泛用于搜索引擎和推荐系统中…"</span>,
        }),
    ];

    <span>const</span> ranked = <span>await</span> compressor.<span>compressDocuments</span>(docs, query);
    <span>console</span>.<span>log</span>(<span>"重排后顺序（pageContent）："</span>);
    <span>for</span> (<span>const</span> d <span>of</span> ranked) {
        <span>console</span>.<span>log</span>(<span>"-"</span>, d.<span>pageContent</span>);
    }
}

<span>main</span>()

测试如下

五.多路召回

多路召回包含以下方式：

向量检索（Milvus）
关键词检索（ES）
过滤召回（权限、时间、类型）
精确匹配（编号、型号）
知识图谱检索

最常见的混合检索就是ES + Milvus 两路并行是多路召回中，最简单的一种方式。

所以说多路召回包含混合检索，但不等于混合检索。

只用语义检索 → 向量检索

只用关键词检索 → ES 搜索

两路一起跑 → 各有各的命中，再合并去重，就能同时覆盖「同义词」和「精确型号」。

「多路召回」是架构层面的事，就是用不同的方式把切当数据全部搜索出来。

「重排」是召回之后的事情，就是把召回的数据按照特定的打分机制重新排序。从而精简大模型的思考范围，以免答非所问

多路召回负责数据的广度：尽量把正确答案捞进候选池
重排负责精度：把正确答案顶到最前面

一句话总结：多路召回就是用向量、关键词、过滤等多种方法各搜一遍，合并去重，再用 RRF 或加权融合统一排序，最后交给重排精排。 核心思想是别把宝押在单一方法上。

多路召回的好处

既能找得全、又能找得准，过滤掉无关杂音，减少大模型瞎编，生产用完全没问题。

多路召回用「多一点延迟」换「几乎不漏掉正确答案」，本质是拿冗余换较高的准确性。

1.加权求和

<span>const</span> merged = <span>new</span> <span>Map</span>();

<span>// 第一路：向量，分数 0~1</span>
<span>for</span> (<span>const</span> hit <span>of</span> vectorHits) {
  merged.<span>set</span>(hit.<span>id</span>, { ...hit, <span>score</span>: hit.<span>score</span> * <span>0.6</span> });
}

<span>// 第二路：关键词，BM25 分数可能上千</span>
<span>for</span> (<span>const</span> hit <span>of</span> bm25Hits) {
  <span>const</span> normalized = hit.<span>score</span> / <span>MAX_BM25</span>;  <span>// 先归一化到 0~1</span>
  <span>if</span> (merged.<span>has</span>(hit.<span>id</span>)) {
    merged.<span>get</span>(hit.<span>id</span>).<span>score</span> += normalized * <span>0.4</span>; <span>// 加权累加</span>
  } <span>else</span> {
    merged.<span>set</span>(hit.<span>id</span>, { ...hit, <span>score</span>: normalized * <span>0.4</span> });
  }
}

<span>// 去重合并后按总分排序</span>
<span>const</span> combined = [...merged.<span>values</span>()].<span>sort</span>(<span>(<span>a, b</span>) =></span> b.<span>score</span> - a.<span>score</span>).<span>slice</span>(<span>0</span>, <span>100</span>);

权重 0.6 / 0.4 是经验值，具体按你的数据调整。

2. RRF（倒数排名融合）

RRF 的好处是完全不关心各路分数的量级，只看谁排得靠前，工业界用得更多。

<span>// RRF 公式：score = Σ 1 / (k + rank)，k 通常取 60</span>
<span>function</span> <span>rrf</span>(<span>hits, k = <span>60</span></span>) {
  <span>return</span> hits.<span>reduce</span>(<span>(<span>acc, hit, rank</span>) =></span> {
    acc[hit.<span>id</span>] = (acc[hit.<span>id</span>] || <span>0</span>) + <span>1</span> / (k + rank + <span>1</span>);
    <span>return</span> acc;
  }, {});
}

<span>const</span> fused = {};
<span>for</span> (<span>const</span> id <span>in</span> <span>rrf</span>(vectorHits)) fused[id] = (fused[id] || <span>0</span>) + <span>rrf</span>(vectorHits)[id];
<span>for</span> (<span>const</span> id <span>in</span> <span>rrf</span>(bm25Hits))   fused[id] = (fused[id] || <span>0</span>) + <span>rrf</span>(bm25Hits)[id];

六. 多问题改写 + 混合检索 + 重排模型

多问题改写 + 混合检索 + 重排模型是当前企业级落地标准的完善版 RAG 方案。

先多路召回，然后合并去重，之后重排关键信息，最后将评分高的数据给大模型，给出最准确的答案。

安装包

pnpm install @langchain/core @langchain/openai @langchain/langgraph @langchain/community dotenv zod @zilliz/milvus2-sdk-node

.env配置

<span>OPENAI_API_KEY</span>=sk-xx
<span>OPENAI_BASE_URL</span>=<span>https</span>:<span>//dashscope.aliyuncs.com/compatible-mode/v1</span>
<span>RERANK_URL</span>=<span>https</span>:<span>//dashscope.aliyuncs.com/api/v1/services/rerank/text-rerank/text-rerank</span>
<span>MODEL_NAME</span>=qwen-plus
<span>RERANK_MODEL</span>=qwen3-rerank

查询扩展

就是在我们问问题的时候，对方的回答恰好对上了，可是因为你问的不对，却没有匹配到这个刚刚好的解决方案，怎么办？

比如我问：iPhone 15 Pro 充不进电怎么办？

就是将问题从不同的角度拆分成多个问题，比如下图，我们问的是：iPhone 15 Pro 充不进电怎么办？ 大模型帮我转化成了三个比较合理的问题。这样能够匹配到正确答案的几率就会增大。这就是所谓的查询扩展。

查询扩展的目的就是提高命中率。

查询扩展又叫多问题改写

原问：iPhone <span>15</span> <span>Pro</span> 充不进电怎么办
   ├─ 角度<span>1</span>（换说法）：iPhone <span>15</span> <span>Pro</span> 无法充电的解决方法
   ├─ 角度<span>2</span>（换角度）：iPhone <span>15</span> <span>Pro</span> 充电无反应排查步骤
   └─ 角度<span>3</span>（加限定）：iPhone <span>15</span> <span>Pro</span> 充电口故障或电池问题

大模型的prompt里面需要有一些字样：可改写说法、换提问角度、或略加限定词；专有名词、型号、订单号等必须保留原样。

代码如下：

<span>/**
 * 用大模型根据用户问题生成恰好 3 条不同角度的检索问句；每条问句各自走 ES / Milvus，最后合并去重。
 */</span>
<span>import</span> { <span>ChatPromptTemplate</span> } <span>from</span> <span>"@langchain/core/prompts"</span>;
<span>import</span> * <span>as</span> z <span>from</span> <span>"zod"</span>;

<span>export</span> <span>const</span> <span>QueryAugmentationSchema</span> = z.<span>object</span>({
  <span>queries</span>: z
    .<span>array</span>(z.<span>string</span>())
    .<span>length</span>(<span>3</span>)
    .<span>describe</span>(
      <span>"恰好 3 条中文检索问句：不同角度改写或扩写；保留订单号、品牌等字面信息；不要编造事实"</span>,
    ),
});

<span>const</span> <span>AUGMENT_PROMPT</span> = <span>ChatPromptTemplate</span>.<span>fromMessages</span>([
  [
    <span>"system"</span>,
    <span>`用户会给出一句中文问题。请另外写出恰好 3 条检索用的问句（与原意一致、角度尽量不同），便于搜索引擎或向量库分别召回：
可改写说法、换提问角度、或略加限定词；专有名词、型号、订单号等必须保留原样。
只输出结构化字段 queries（长度为 3 的字符串数组）。`</span>,
  ],
  [<span>"human"</span>, <span>"{query}"</span>],
]);

<span>function</span> <span>normalizeThreeQueries</span>(<span>original, list</span>) {
  <span>const</span> out = (list ?? [])
    .<span>map</span>(<span>(<span>s</span>) =></span> (<span>typeof</span> s === <span>"string"</span> ? s.<span>trim</span>() : <span>""</span>))
    .<span>filter</span>(<span>Boolean</span>);
  <span>while</span> (out.<span>length</span> < <span>3</span>) out.<span>push</span>(original);
  <span>return</span> out.<span>slice</span>(<span>0</span>, <span>3</span>);
}


<span>export</span> <span>async</span> <span>function</span> <span>augmentQuery</span>(<span>chatModel, query</span>) {
  <span>const</span> structured = chatModel.<span>withStructuredOutput</span>(<span>QueryAugmentationSchema</span>);
  <span>const</span> chain = <span>AUGMENT_PROMPT</span>.<span>pipe</span>(structured);
  <span>try</span> {
    <span>const</span> raw = <span>await</span> chain.<span>invoke</span>({ query });
    <span>return</span> { <span>queries</span>: <span>normalizeThreeQueries</span>(query, raw.<span>queries</span>) };
  } <span>catch</span> {
    <span>return</span> { <span>queries</span>: <span>normalizeThreeQueries</span>(query, []) };
  }
}

<span>/** 原始问题在前，其后接 LLM 生成的问句；不做去重，顺序固定；每条各跑一次 ES、Milvus */</span>
<span>export</span> <span>function</span> <span>retrievalQueryStrings</span>(<span>original, augmentation</span>) {
  <span>return</span> [original, ...(augmentation?.<span>queries</span> ?? [])]
    .<span>map</span>(<span>(<span>s</span>) =></span> (<span>typeof</span> s === <span>"string"</span> ? s.<span>trim</span>() : <span>""</span>))
    .<span>filter</span>(<span>Boolean</span>);
}

实现多路召回和重排

<span>/**
 * 混合检索：LLM 重写为 3 条多角度问句 → 每条问句分别 ES + Milvus → 全量合并去重 → Rerank → LLM 作答。
 * LangGraph：START → query_augment → es_recall ∥ milvus_recall → merge → rerank → generate_answer → END。
 */</span>
<span>import</span> <span>"dotenv/config"</span>;
<span>import</span> { <span>Client</span> } <span>from</span> <span>"@elastic/elasticsearch"</span>;
<span>import</span> { <span>Document</span> } <span>from</span> <span>"@langchain/core/documents"</span>;
<span>import</span> { <span>ChatPromptTemplate</span> } <span>from</span> <span>"@langchain/core/prompts"</span>;
<span>import</span> { <span>Milvus</span> } <span>from</span> <span>"@langchain/community/vectorstores/milvus"</span>;
<span>import</span> { <span>ChatOpenAI</span>, <span>OpenAIEmbeddings</span> } <span>from</span> <span>"@langchain/openai"</span>;
<span>import</span> { <span>Annotation</span>, <span>END</span>, <span>START</span>, <span>StateGraph</span> } <span>from</span> <span>"@langchain/langgraph"</span>;
<span>import</span> { <span>DashScopeRerank</span> } <span>from</span> <span>"../rerank/dashscope-rerank.mjs"</span>;
<span>import</span> {
  augmentQuery,
  retrievalQueryStrings,
} <span>from</span> <span>"./query-augment.mjs"</span>;

<span>const</span> <span>INDEX</span> = <span>"life_notes"</span>;

<span>const</span> <span>HybridRetrievalState</span> = <span>Annotation</span>.<span>Root</span>({
  <span>query</span>: <span>Annotation</span>(),
  <span>queryAugmentation</span>: <span>Annotation</span>(),
  <span>esHits</span>: <span>Annotation</span>(),
  <span>milvusHits</span>: <span>Annotation</span>(),
  <span>merged</span>: <span>Annotation</span>(),
  <span>topDocuments</span>: <span>Annotation</span>(),
  <span>answer</span>: <span>Annotation</span>(),
});

<span>function</span> <span>docFromEsHit</span>(<span>hit</span>) {
  <span>const</span> s = hit.<span>_source</span> ?? {};
  <span>const</span> text = [s.<span>note_title</span> ?? s.<span>title</span>, s.<span>note_body</span> ?? s.<span>content</span>]
    .<span>filter</span>(<span>Boolean</span>)
    .<span>join</span>(<span>"\n"</span>);
  <span>return</span> <span>new</span> <span>Document</span>({
    <span>pageContent</span>: text,
    <span>metadata</span>: { <span>id</span>: hit.<span>_id</span>, <span>source</span>: <span>"es"</span>, ...s },
  });
}

<span>/** ES 与 Milvus 结果拼接后仅按 metadata.id 去重，保留首次出现（通常 ES 在前） */</span>
<span>function</span> <span>merge</span>(<span>esDocs, milvusDocs</span>) {
  <span>const</span> combined = [...(esDocs ?? []), ...(milvusDocs ?? [])].<span>filter</span>(
    <span>(<span>d</span>) =></span> d?.<span>pageContent</span>,
  );
  <span>return</span> <span>dedupeDocsById</span>(combined);
}

<span>/** 去重键仅为 metadata.id（trim 后非空）；无 id 丢弃，不按正文去重；保留首次出现顺序 */</span>
<span>function</span> <span>dedupeDocsById</span>(<span>docs</span>) {
  <span>const</span> seen = <span>new</span> <span>Set</span>();
  <span>const</span> out = [];
  <span>for</span> (<span>const</span> d <span>of</span> docs ?? []) {
    <span>if</span> (!d?.<span>pageContent</span>) <span>continue</span>;
    <span>const</span> id =
      d.<span>metadata</span>?.<span>id</span> != <span>null</span> ? <span>String</span>(d.<span>metadata</span>.<span>id</span>).<span>trim</span>() : <span>""</span>;
    <span>if</span> (!id) <span>continue</span>;
    <span>if</span> (seen.<span>has</span>(id)) <span>continue</span>;
    seen.<span>add</span>(id);
    out.<span>push</span>(d);
  }
  <span>return</span> out;
}

<span>function</span> <span>printDocs</span>(<span>label, docs</span>) {
  <span>console</span>.<span>log</span>(<span>`\n=== <span>${label}</span> (<span>${docs?.length ?? <span>0</span>}</span> 条) ===`</span>);
  <span>for</span> (<span>let</span> i = <span>0</span>; i < (docs ?? []).<span>length</span>; i++) {
    <span>const</span> d = docs[i];
    <span>const</span> preview = (d.<span>pageContent</span> ?? <span>""</span>).<span>slice</span>(<span>0</span>, <span>200</span>).<span>replace</span>(<span>/\n/g</span>, <span>" "</span>);
    <span>console</span>.<span>log</span>(<span>`[<span>${i}</span>] <span>${preview}</span><span>${d.pageContent?.length > <span>200</span> ? <span>"…"</span> : <span>""</span>}</span>`</span>);
    <span>console</span>.<span>log</span>(<span>`    metadata:`</span>, d.<span>metadata</span> ?? {});
  }
}

<span>/** 打印 LLM 生成的多角度检索问句及逐条检索列表 */</span>
<span>function</span> <span>printQueryRewrite</span>(<span>original, augmentation</span>) {
  <span>const</span> qs = augmentation?.<span>queries</span> ?? [];
  <span>const</span> forRetrieval = <span>retrievalQueryStrings</span>(original, augmentation);

  <span>console</span>.<span>log</span>(<span>`\n--- 查询扩展（LLM 生成 <span>${qs.length}</span> 条检索问句）---`</span>);
  <span>console</span>.<span>log</span>(<span>"原始 query:"</span>, original ?? <span>""</span>);
  <span>for</span> (<span>let</span> i = <span>0</span>; i < qs.<span>length</span>; i++) <span>console</span>.<span>log</span>(<span>`  [<span>${i + <span>1</span>}</span>] <span>${qs[i] ?? <span>""</span>}</span>`</span>);
  <span>console</span>.<span>log</span>(
    <span>`\n逐条 ES + Milvus（共 <span>${forRetrieval.length}</span> 条检索串，含原始问题）:`</span>,
  );
  <span>for</span> (<span>let</span> i = <span>0</span>; i < forRetrieval.<span>length</span>; i++) {
    <span>console</span>.<span>log</span>(<span>`  [<span>${i + <span>1</span>}</span>] <span>${forRetrieval[i] ?? <span>""</span>}</span>`</span>);
  }
}

<span>function</span> <span>stringifyMessageContent</span>(<span>content</span>) {
  <span>if</span> (<span>typeof</span> content === <span>"string"</span>) <span>return</span> content;
  <span>if</span> (!<span>Array</span>.<span>isArray</span>(content)) <span>return</span> <span>String</span>(content ?? <span>""</span>);
  <span>return</span> content
    .<span>map</span>(<span>(<span>c</span>) =></span>
      <span>typeof</span> c === <span>"string"</span> ? c : <span>typeof</span> c?.<span>text</span> === <span>"string"</span> ? c.<span>text</span> : <span>""</span>,
    )
    .<span>join</span>(<span>""</span>);
}

<span>function</span> <span>formatDocsAsContext</span>(<span>docs</span>) {
  <span>return</span> (docs ?? [])
    .<span>map</span>(<span>(<span>d, i</span>) =></span> {
      <span>const</span> meta = d.<span>metadata</span> ?? {};
      <span>const</span> src = meta.<span>source</span> ?? <span>""</span>;
      <span>const</span> id = meta.<span>id</span> != <span>null</span> ? <span>String</span>(meta.<span>id</span>) : <span>""</span>;
      <span>const</span> head = id ? <span>`[<span>${i + <span>1</span>}</span>] id=<span>${id}</span><span>${src ? <span>` source=<span>${src}</span>`</span> : <span>""</span>}</span>`</span> : <span>`[<span>${i + <span>1</span>}</span>]`</span>;
      <span>return</span> <span>`<span>${head}</span>\n<span>${d.pageContent ?? <span>""</span>}</span>`</span>;
    })
    .<span>join</span>(<span>"\n\n---\n\n"</span>);
}

<span>const</span> <span>ANSWER_PROMPT</span> = <span>ChatPromptTemplate</span>.<span>fromMessages</span>([
  [
    <span>"system"</span>,
    <span>`你是阅读用户「生活笔记」知识库并作答的助手。
规则：
- 只根据下方「检索片段」推断答案；片段里没有的信息不要编造。
- 若片段不足以回答，明确说明「笔记里未提到」，并可给出一句保守建议。
- 回答简洁有条理，可使用简短列表；口吻自然中文。`</span>,
  ],
  [
    <span>"human"</span>,
    <span>`用户问题：{query}

检索片段：
{context}`</span>,
  ],
]);

<span>const</span> <span>NO_CONTEXT_PROMPT</span> = <span>ChatPromptTemplate</span>.<span>fromMessages</span>([
  [
    <span>"system"</span>,
    <span>`你是阅读用户「生活笔记」知识库并作答的助手。当前没有检索到任何片段。
请用一两句话说明无法从笔记中回答，并礼貌询问用户是否换个说法或补充关键词。`</span>,
  ],
  [<span>"human"</span>, <span>"用户问题：{query}"</span>],
]);

<span>export</span> <span>function</span> <span>compileHybridRetrievalGraph</span>(<span>esClient, milvus, reranker, chatModel</span>) {
  <span>const</span> <span>ES_K</span> = <span>15</span>;
  <span>const</span> <span>MILVUS_K</span> = <span>15</span>;

  <span>return</span> <span>new</span> <span>StateGraph</span>(<span>HybridRetrievalState</span>)
    .<span>addNode</span>(<span>"query_augment"</span>, <span>async</span> (state) => ({
      <span>queryAugmentation</span>: <span>await</span> <span>augmentQuery</span>(chatModel, state.<span>query</span> ?? <span>""</span>),
    }))
    .<span>addNode</span>(<span>"es_recall"</span>, <span>async</span> (state) => {
      <span>const</span> qs = <span>retrievalQueryStrings</span>(state.<span>query</span>, state.<span>queryAugmentation</span>);
      <span>const</span> n = <span>Math</span>.<span>max</span>(<span>1</span>, qs.<span>length</span>);
      <span>const</span> kEach = <span>Math</span>.<span>max</span>(<span>2</span>, <span>Math</span>.<span>ceil</span>(<span>ES_K</span> / n));
      <span>const</span> batches = <span>await</span> <span>Promise</span>.<span>all</span>(
        qs.<span>map</span>(<span>(<span>q</span>) =></span>
          esClient.<span>search</span>({
            <span>index</span>: <span>INDEX</span>,
            <span>size</span>: kEach,
            <span>query</span>: {
              <span>multi_match</span>: {
                <span>query</span>: q,
                <span>fields</span>: [<span>"note_title^2"</span>, <span>"note_body"</span>, <span>"title"</span>, <span>"content"</span>],
                <span>type</span>: <span>"best_fields"</span>,
                <span>analyzer</span>: <span>"ik_smart"</span>,
              },
            },
          }),
        ),
      );
      <span>const</span> flat = batches.<span>flatMap</span>(<span>(<span>res</span>) =></span>
        (res.<span>hits</span>?.<span>hits</span> ?? []).<span>map</span>(docFromEsHit),
      );
      <span>return</span> { <span>esHits</span>: <span>dedupeDocsById</span>(flat) };
    })
    .<span>addNode</span>(<span>"milvus_recall"</span>, <span>async</span> (state) => {
      <span>const</span> qs = <span>retrievalQueryStrings</span>(state.<span>query</span>, state.<span>queryAugmentation</span>);
      <span>const</span> n = <span>Math</span>.<span>max</span>(<span>1</span>, qs.<span>length</span>);
      <span>const</span> kEach = <span>Math</span>.<span>max</span>(<span>2</span>, <span>Math</span>.<span>ceil</span>(<span>MILVUS_K</span> / n));
      <span>const</span> batches = <span>await</span> <span>Promise</span>.<span>all</span>(
        qs.<span>map</span>(<span>(<span>q</span>) =></span> milvus.<span>similaritySearch</span>(q, kEach)),
      );
      <span>const</span> flat = batches.<span>flat</span>();
      <span>return</span> { <span>milvusHits</span>: <span>dedupeDocsById</span>(flat) };
    })
    .<span>addNode</span>(<span>"merge"</span>, <span>async</span> (state) => ({
      <span>merged</span>: <span>merge</span>(state.<span>esHits</span>, state.<span>milvusHits</span>),
    }))
    .<span>addNode</span>(<span>"rerank"</span>, <span>async</span> (state) => {
      <span>const</span> merged = state.<span>merged</span> ?? [];
      <span>if</span> (!merged.<span>length</span>) <span>return</span> { <span>topDocuments</span>: [] };
      <span>const</span> topDocuments = <span>await</span> reranker.<span>compressDocuments</span>(merged, state.<span>query</span>);
      <span>return</span> { topDocuments };
    })
    .<span>addNode</span>(<span>"generate_answer"</span>, <span>async</span> (state) => {
      <span>const</span> query = state.<span>query</span> ?? <span>""</span>;
      <span>const</span> docs = state.<span>topDocuments</span> ?? [];
      <span>if</span> (!docs.<span>length</span>) {
        <span>const</span> chain = <span>NO_CONTEXT_PROMPT</span>.<span>pipe</span>(chatModel);
        <span>const</span> msg = <span>await</span> chain.<span>invoke</span>({ query });
        <span>return</span> { <span>answer</span>: <span>stringifyMessageContent</span>(msg.<span>content</span>).<span>trim</span>() };
      }
      <span>const</span> chain = <span>ANSWER_PROMPT</span>.<span>pipe</span>(chatModel);
      <span>const</span> msg = <span>await</span> chain.<span>invoke</span>({
        query,
        <span>context</span>: <span>formatDocsAsContext</span>(docs),
      });
      <span>return</span> { <span>answer</span>: <span>stringifyMessageContent</span>(msg.<span>content</span>).<span>trim</span>() };
    })
    .<span>addEdge</span>(<span>START</span>, <span>"query_augment"</span>)
    .<span>addEdge</span>(<span>"query_augment"</span>, <span>"es_recall"</span>)
    .<span>addEdge</span>(<span>"query_augment"</span>, <span>"milvus_recall"</span>)
    .<span>addEdge</span>([<span>"es_recall"</span>, <span>"milvus_recall"</span>], <span>"merge"</span>)
    .<span>addEdge</span>(<span>"merge"</span>, <span>"rerank"</span>)
    .<span>addEdge</span>(<span>"rerank"</span>, <span>"generate_answer"</span>)
    .<span>addEdge</span>(<span>"generate_answer"</span>, <span>END</span>)
    .<span>compile</span>();
}

<span>const</span> esClient = <span>new</span> <span>Client</span>({ <span>node</span>: <span>"http://localhost:9200"</span> });
<span>const</span> embeddings = <span>new</span> <span>OpenAIEmbeddings</span>({
  <span>model</span>: <span>"text-embedding-v3"</span>,
  <span>apiKey</span>: process.<span>env</span>.<span>OPENAI_API_KEY</span>,
  <span>configuration</span>: {
    <span>baseURL</span>: <span>"https://dashscope.aliyuncs.com/compatible-mode/v1"</span>,
  },
});
<span>const</span> milvus = <span>await</span> <span>Milvus</span>.<span>fromExistingCollection</span>(embeddings, {
  <span>url</span>: <span>"http://localhost:19530"</span>,
  <span>collectionName</span>: <span>INDEX</span>,
  <span>textField</span>: <span>"doc_text"</span>,
  <span>vectorField</span>: <span>"embedding"</span>,
});
<span>const</span> reranker = <span>new</span> <span>DashScopeRerank</span>({
  <span>apiKey</span>: process.<span>env</span>.<span>OPENAI_API_KEY</span>,
  <span>model</span>: <span>"qwen3-rerank"</span>,
  <span>topN</span>: <span>3</span>,
  <span>baseUrl</span>:
    <span>"https://dashscope.aliyuncs.com/api/v1/services/rerank/text-rerank/text-rerank"</span>,
});

<span>const</span> chatModel = <span>new</span> <span>ChatOpenAI</span>({
  <span>model</span>: process.<span>env</span>.<span>LLM_MODEL_NAME</span> ?? <span>"qwen-turbo"</span>,
  <span>apiKey</span>: process.<span>env</span>.<span>OPENAI_API_KEY</span>,
  <span>temperature</span>: <span>0.2</span>,
  <span>configuration</span>: {
    <span>baseURL</span>:
      process.<span>env</span>.<span>OPENAI_BASE_URL</span>
  },
});

<span>/** 示例用户 query（字符串列表） */</span>
<span>const</span> <span>SAMPLE_QUERIES</span> = [
  <span>// "PO-20250409-K9 滤芯订单",</span>
  <span>"家里无线老是断断续续的咋整啊"</span>,
  <span>// "那个黑凉粉粉怎么冲不结块",</span>
  <span>// "明火炖太久汤汁又黏又涩，起锅前要怎么处理才不腻",</span>
];

<span>const</span> graph = <span>compileHybridRetrievalGraph</span>(esClient, milvus, reranker, chatModel);

<span>const</span> drawable = <span>await</span> graph.<span>getGraphAsync</span>();
<span>console</span>.<span>log</span>(drawable.<span>drawMermaid</span>());
<span>console</span>.<span>log</span>();

<span>for</span> (<span>const</span> query <span>of</span> <span>SAMPLE_QUERIES</span>) {
  <span>console</span>.<span>log</span>(<span>`query: <span>${query}</span>`</span>);

  <span>const</span> state = <span>await</span> graph.<span>invoke</span>({ query });

  <span>printQueryRewrite</span>(state.<span>query</span>, state.<span>queryAugmentation</span>);
  <span>console</span>.<span>log</span>(<span>"\n（原始 JSON）"</span>, <span>JSON</span>.<span>stringify</span>(state.<span>queryAugmentation</span>));

  <span>printDocs</span>(<span>"Elasticsearch 检索"</span>, state.<span>esHits</span>);
  <span>printDocs</span>(<span>"Milvus 检索"</span>, state.<span>milvusHits</span>);
  <span>printDocs</span>(<span>"重排后保留"</span>, state.<span>topDocuments</span> ?? []);

  <span>console</span>.<span>log</span>(<span>"\n=== 大模型生成回答 ===\n"</span>);
  <span>console</span>.<span>log</span>(state.<span>answer</span> ?? <span>""</span>);
}

ai agent--- 后端概念补充：Docker Compose、ElasticSearch、IK、BM25等

一.Docker Compose是什么?

Docker Compose的好处

docker-compose.yml

二.ElasticSearch是什么?

1.Elasticsearch的特点

2.Elasticsearch 能做什么

3.ES流水线组成部分

4.如何使用ES

4.1 安装docker，

4.2 拉取ES

4.3 运行ES

4.4 安装 IK 插件

4.5创建索引

4.6 在nodejs里面使用

4.7 在nodejs里面进行增删改查操作

5.ES的数据存在哪里？

三. 嵌入模型

解释嵌入模型

向milvus保存数据的步骤

1.将文件转化成纯文本

2.切片

3.用嵌入模型生成向量

4.向 Milvus写入数据

查询milvus的步骤

1.向量化问题

2.利用milvus的api搜索相似值

3.拿文档

四.重排模型

重排模型是什么

为什么要用重排模型

重排模型怎么用

常见重排模型怎么选

千问重排

获取apiKey

curl测试

nodejs测试

安装包

.env配置

重排

封装重排的方法

测试重排

测试如下

五.多路召回

多路召回的好处

1.加权求和

2. RRF（倒数排名融合）

六. 多问题改写 + 混合检索 + 重排模型

安装包

.env配置

查询扩展

相关推荐