RAGRetrievalService 参考

RAGRetrievalService 负责从向量库中检索相关文本块，并生成可用于 LLM 的上下文。

概览

检索服务提供以下能力：

将查询转为向量
在 Qdrant 中检索相似文本块（可启用双读集合）
向量召回后可选执行 rerank 阶段
按 Agent 挂载的知识源过滤
回表查询文本块内容
格式化上下文

配置

该服务继承其依赖的嵌入配置：

参数	类型	默认值	说明
`embedding.backend_type`	str	`api`	查询 embedding 后端类型（`api` / `local`）
`embedding.provider`	str	`sentence_transformers`	查询 embedding provider
`embedding.model`	str	`sentence-transformers/all-MiniLM-L6-v2`	查询嵌入模型
`embedding.dim`	int	384	向量维度

可选重排配置：

参数	类型	默认值	说明
`rerank.enabled`	bool	false	是否启用检索后重排
`rerank.model`	str	`qwen3-rerank`	重排模型（仅支持 `rerank_models` 类别）
`rerank.top_k`	int	20	进入重排阶段的候选上限

检索参数在调用时传入，不做全局配置。

API 参考

类：`RAGRetrievalService`

class RAGRetrievalService:
    """Service for retrieving relevant context for RAG."""

    def __init__(
        self,
        *,
        qdrant_service: Optional[QdrantVectorService] = None,
        embedding_service: Optional[EmbeddingService] = None,
    ) -> None:
        """Initialize the RAG retrieval service.

        Args:
            qdrant_service: Qdrant service instance
            embedding_service: Embedding service instance
        """

方法：`retrieve_context`

检索与查询相关的文本块。

def retrieve_context(
    self,
    query: str,
    agent_id: str,
    *,
    top_k: int = 5,
    score_threshold: Optional[float] = 0.3,
) -> list[RetrievedChunk]:
    """Retrieve relevant document chunks for a query.

    Args:
        query: User query to find relevant context for
        agent_id: Agent identifier to filter by mounted sources
        top_k: Maximum number of chunks to retrieve
        score_threshold: Minimum similarity score threshold

    Returns:
        List of relevant chunks sorted by score
    """

返回值：list[RetrievedChunk]

每个 RetrievedChunk 包含：

chunk_id：文本块 ID
source_id：知识源 ID
document_name：原始文件名
chunk_index：块序号
content：文本内容
score：相似度分数
metadata：额外元数据

方法：`format_context`

将检索结果格式化为上下文字符串。

def format_context(
    self,
    chunks: list[RetrievedChunk],
    *,
    max_context_length: int = 4000,
    include_metadata: bool = True,
) -> str:
    """Format retrieved chunks into a context string for the LLM.

    Args:
        chunks: Retrieved chunks to format
        max_context_length: Maximum total context length
        include_metadata: Whether to include source metadata

    Returns:
        Formatted context string
    """

方法：`retrieve_and_format`

一次调用完成检索与格式化。

def retrieve_and_format(
    self,
    query: str,
    agent_id: str,
    *,
    top_k: int = 5,
    score_threshold: Optional[float] = 0.3,
    max_context_length: int = 4000,
) -> str:
    """Retrieve and format context in one call.

    Args:
        query: User query
        agent_id: Agent identifier
        top_k: Maximum chunks to retrieve
        score_threshold: Minimum similarity score
        max_context_length: Maximum context length

    Returns:
        Formatted context string, empty if no relevant chunks
    """

工厂函数：`get_rag_service`

def get_rag_service() -> RAGRetrievalService:
    """Get or create the singleton RAG retrieval service instance."""

使用示例

基础检索

from ai_service.services.rag_retrieval import get_rag_service

rag_service = get_rag_service()
chunks = rag_service.retrieve_context(
    query="What is the refund policy?",
    agent_id="agent-123",
    top_k=5,
    score_threshold=0.3
)

for chunk in chunks:
    print(f"Score: {chunk.score:.3f}")
    print(f"Source: {chunk.document_name}")
    print(f"Content: {chunk.content[:100]}...")
    print()

检索并格式化

context = rag_service.retrieve_and_format(
    query="How do I track my order?",
    agent_id="agent-123",
    top_k=3,
    max_context_length=2000
)

if context:
    print("Retrieved context:")
    print(context)
else:
    print("No relevant context found")

自定义格式化

chunks = rag_service.retrieve_context(
    query="What are the shipping options?",
    agent_id="agent-123"
)

context = rag_service.format_context(
    chunks,
    max_context_length=3000,
    include_metadata=False
)

调整检索参数

# 更多结果，低阈值
chunks = rag_service.retrieve_context(
    query="pricing information",
    agent_id="agent-123",
    top_k=10,
    score_threshold=0.2
)

# 更少结果，高阈值
chunks = rag_service.retrieve_context(
    query="specific technical detail",
    agent_id="agent-123",
    top_k=3,
    score_threshold=0.5
)

检索参数说明

top_k

控制返回的最大文本块数量：

小 1-3：更精准
中 5-7：平衡推荐
大 10-20：覆盖更广

score_threshold

相似度阈值，过滤低相关结果：

低 0.1-0.2：更高召回
中 0.3-0.4：平衡推荐
高 0.5-0.7：更高精度

max_context_length

限制最终上下文长度：

短 1000-2000：快速
中 3000-4000：平衡推荐
长 5000-8000：更全面

上下文格式

含元数据

[Source: user_guide.pdf, Chunk 3, Relevance: 0.85]
Our refund policy allows returns within 30 days...

[Source: faq.pdf, Chunk 12, Relevance: 0.72]
To request a refund, please contact support...

不含元数据

Our refund policy allows returns within 30 days...

To request a refund, please contact support...

知识源过滤机制

服务会自动使用 Agent 挂载的知识源进行过滤：

chunks = rag_service.retrieve_context(
    query="...",
    agent_id="agent-123"
)

工作流程：

查询 Agent 的已挂载 source_id
用 source_id 过滤 Qdrant 检索
仅返回已挂载来源的结果

性能特性

阶段	耗时	备注
查询嵌入	0.1-0.3s	取决于模型
向量检索	0.01-0.1s	取决于集合规模
文本回表	0.01-0.05s	PostgreSQL 查询
格式化	<0.01s	字符串处理
总计	0.2-0.5s	常见范围

最佳实践

参数调优

chunks = rag_service.retrieve_context(
    query=query,
    agent_id=agent_id,
    top_k=5,
    score_threshold=0.3
)

结果太少：降低阈值或提高 top_k
结果不相关：提高阈值或降低 top_k

空结果处理

chunks = rag_service.retrieve_context(query, agent_id)

if not chunks:
    # 可选择：不带上下文回答或降低阈值重试
    pass

上下文长度控制

estimated_length = sum(len(c.content) for c in chunks)
if estimated_length > max_context_length:
    # 可选择降低 top_k 或增加 max_context_length
    pass

常见问题

问题：没有检索结果

原因：

Agent 未挂载知识源
相似度阈值过高
查询语义与语料偏差大

解决：

from ai_service.storage.models import get_active_source_ids_for_agent
from ai_service.utils.database import SessionLocal

db = SessionLocal()
sources = get_active_source_ids_for_agent(db, agent_id="agent-123")
print(f"Mounted sources: {sources}")
db.close()

chunks = rag_service.retrieve_context(
    query=query,
    agent_id=agent_id,
    score_threshold=0.1
)

问题：结果不相关

解决：提高 score_threshold 或降低 top_k。

问题：检索变慢

原因：集合过大、连接池不足或模型未预热

解决：预热嵌入模型、检查 Qdrant 与数据库性能。