超越简单嵌入，构建大模型智能体的生产级上下文检索系统

文章探讨了构建大模型智能体上下文检索系统的必要性，指出简单线性检索流程不适合生产环境。作者提出包含五层架构（索引、路由、查询构建、检索、生成）的解决方案，并介绍Airweave开源框架如何实现这一系统，支持多源数据连接、智能路由和混合搜索等功能。

想象一下，你的数据分散在多个来源中（Gmail、Drive、Notion 等）。

你会如何为其构建一个统一的查询引擎？

许多开发者仍然认为上下文检索是一个线性流程：

分块 → 嵌入 → 检索 → 生成

这对于简单的演示效果很好，但生产系统需要从根本上不同的东西。

为了更好地理解，请考虑这个查询：

“比较我们在芝加哥地区的第四季度销售业绩与利益相关者会议上制定的去年预测。”

这一类单一查询需要：

来自 SQL 数据库的销售数据
图关系（组织层级结构）
针对预测报告的向量搜索
基于时间的过滤（今年第四季度与去年相比）
权限检查（用于用户授权）

没有单一的嵌入查找可以处理这种复杂性！

要真正解决这个问题，你需要构建一个包含五个关键层的智能体上下文检索系统：

索引层不同的内容需要不同的索引：

文档的语义分块
嵌套内容的层次化索引
针对 Calendar、Slack 等来源的特殊索引

2. 路由层在检索之前，你需要智能路由来决定：

查询应该命中图数据库吗？
它需要结构化的 SQL 查询吗？
还是需要用于概念匹配的语义搜索？

3. 查询构建层原始查询可能需要：

被分解为子查询
被翻译成不同的查询语言（SQL, Cypher, 向量相似度）

4. 检索层

应用权限和访问检查
并行运行多个检索
根据相关性/新鲜度重排序

5. 生成层综合生成一个带有引用支持的响应

在第一个查询进入生产环境之前，这需要数月的工程开发。

这绝对是一个难题……

……但这正是 Google (Vertex AI)、Microsoft (Azure AI Search) 和 AWS (Amazon Q) 等公司构建其生产级智能体的方式。

如果你想看到实际应用，这种方法实际上已在Airweave中实现，这是一个近期流行的 100% 开源框架，为跨 30 多个应用程序和数据库的 AI 智能体提供上下文检索层。

它实现了上面讨论的上下文检索的整个生产堆栈，例如：

针对特定来源的索引
查询扩展
智能路由
多源检索
类似 Perplexity 的带引用支持的响应

例如，为了检测更新并启动重新同步，人们可能会进行时间戳比较。

但这并不能说明内容是否真的发生了变化（也许只是权限被更新了），你可能仍然不必要地重新嵌入了所有内容。

Airweave 通过实现特定于来源的哈希技术来处理这个问题，例如实体级哈希、文件内容哈希、基于游标的同步等。

作为一个要点，请记住，智能体的上下文检索是一个基础设施问题，而不是嵌入问题。

你需要从第一天就构建持续同步、智能分块和混合搜索。

你可以在这里找到 Airweave 仓库 → https://github.com/airweave-ai/airweave

此Repo上的演示视频，可以帮助你学习如何为智能体提供上下文检索层，从而能够实时搜索任何应用程序、数据库或文档存储。

它无缝连接到 Notion、Google Drive 和 SQL 数据库等工具，将其内容转换为可搜索的知识。

使用方式

前端

Access the UI at http://localhost:8080
Connect sources, configure syncs, and query data

API

Swagger docs: http://localhost:8001/docs
Create connections, trigger syncs, and search data

Python SDK

pip install airweave-sdk

from airweave import AirweaveSDK # Initialize client client = AirweaveSDK( api_key="YOUR_API_KEY", base_url="http://localhost:8001" ) # Create a collection collection = client.collections.create(name="My Collection") # Add a source connection source = client.source_connections.create( name="My Stripe Connection", short_name="stripe", readable_collection_id=collection.readable_id, authentication={ "credentials": {"api_key": "your_stripe_api_key"} } ) # Semantic search (default) results = client.collections.search( readable_id=collection.readable_id, query="Find recent failed payments" ) # Hybrid search (semantic + keyword) results = client.collections.search( readable_id=collection.readable_id, query="customer invoices Q4 2024", search_type="hybrid" ) # With query expansion and reranking results = client.collections.search( readable_id=collection.readable_id, query="technical documentation", enable_query_expansion=True, enable_reranking=True, top_k=20 ) # Search with recency bias (prioritize recent results) results = client.collections.search( readable_id=collection.readable_id, query="critical bugs", recency_bias=0.8, # 0.0 to 1.0, higher = more recent limit=10 ) # Get AI-generated answer instead of raw results answer = client.collections.search( readable_id=collection.readable_id, query="What are our customer refund policies?", response_type="completion", enable_reranking=True )

主要技术栈

前端: React/TypeScript with ShadCN
后端: FastAPI (Python)
数据库: PostgreSQL (metadata), Qdrant (vectors)
工作组件: Temporal (workflow orchestration), Redis (pub/sub)
部署: Docker Compose (dev), Kubernetes (prod)