CodeBuddy协作实战：构建基于多数据库的混合检索 RAG 问答系统开发日志
原创
CodeBuddy协作实战：构建基于多数据库的混合检索 RAG 问答系统开发日志
原创
熊猫钓鱼
修改于 2025-09-03 17:51:15
修改于 2025-09-03 17:51:15
代码可运行
运行总次数：0
代码可运行
在人工智能技术重塑软件开发流程的今天，AI编码助手不再仅仅是自动补全工具，而是能够深度参与系统架构设计、核心算法实现与复杂问题排查的协作伙伴。
本次开发日志将详细记录我如何利用CodeBuddy，从零开始构建一个融合了关系数据库、文档数据库与向量数据库的混合检索 RAG（检索增强生成）问答系统。面对传统关键词检索在语义理解上的固有局限，我通过AI辅助完成了技术选型、架构设计、核心代码实现以及性能调优的全过程。本文将深入探讨这一过程中遇到的挑战、解决方案以及AI协作带来的效率提升，为开发复杂数据密集型应用提供一份详细的实战参考。
回顾整个项目，我感觉最大的收获在于探索出一种与AI高效协同的新开发模式：开发者专注于核心业务逻辑、整体架构和关键决策，而将实现细节、样板代码和常见模式的设计交由AI高效完成。这种模式并非取代开发者，而是将其从繁重的重复劳动中解放出来，从而能更专注于创造和创新。随着AI技术的持续演进，这种深度协作模式必将成为软件开发的新常态，帮助我们应对日益复杂的技术挑战，构建出更智能、更强大的应用系统。
技术环境与协作工具
- 核心AI工具：CodeBuddy
- 项目环境：Python 3.10, FastAPI, PostgreSQL, MongoDB, Pinecone, OpenAI Embeddings
- 协作场景：从0到1构建支持混合检索的RAG（检索增强生成）问答系统
协作目标：解决传统检索的局限性
在开发企业知识库系统时，我遇到了一个关键问题：传统的基于关键词的搜索引擎无法理解用户查询的语义含义。比如用户搜索"员工请假流程"，但文档中只有"年假申请审批步骤"的内容，传统搜索无法有效匹配。
我的目标是构建一个能够：
- 理解用户查询的语义意图
- 同时支持精确的关键词匹配和模糊的语义搜索
- 结合结构化数据和非结构化数据的优势
- 提供准确且上下文相关的答案
第一阶段：架构设计中的AI协作
初始方案与AI咨询
我首先向CodeBuddy提出了架构设计需求：
我的提示词：
我需要设计一个混合检索系统，要求：
1. 使用关系数据库存储结构化元数据（文档ID、创建时间、所属部门等）
2. 使用文档数据库存储原始文档内容和可变属性
3. 使用向量数据库处理语义检索
4. 能够将三种数据库的检索结果进行融合排序
请给出一个可行的技术架构方案，并说明数据同步的策略。CodeBuddy的响应：
基于您的需求，我建议以下架构：
## 技术栈推荐
- **关系数据库**: PostgreSQL (支持JSONB和强大的事务能力)
- **文档数据库**: MongoDB (灵活的文档模型)
- **向量数据库**: Pinecone (托管服务，简化开发) 或 Milvus (开源)
- **应用框架**: FastAPI (异步支持好)
- **嵌入模型**: OpenAI text-embedding-ada-002
## 数据流设计
1. **文档入库流程**：
- 原始文档存入MongoDB，获取ObjectId
- 文档元数据存入PostgreSQL，包含MongoDB的文档引用ID
- 文档内容通过嵌入模型向量化后存入Pinecone，关联PostgreSQL中的文档ID
2. **查询流程**：
- 用户查询同时发起三种搜索：
a) PostgreSQL中的关键词匹配（标题、标签等）
b) MongoDB中的全文检索（可选）
c) Pinecone中的向量相似度搜索
- 对三种结果进行加权融合排序
## 数据同步策略
- 使用异步任务队列（Celery或RabbitMQ）处理向量化过程
- 为所有数据记录添加时间戳和版本标记
- 设计幂等的更新操作避免数据不一致这个建议为我提供了清晰的架构方向，特别是数据同步策略部分是我最初没有考虑周全的。
第二阶段：核心实现中的AI辅助
数据模型定义
在定义数据模型时，我使用CodeBuddy生成了基础的数据模型代码：
我的提示词：
请为上述架构创建Python数据模型：
1. PostgreSQL的文档元数据模型（使用SQLAlchemy）
2. MongoDB的文档内容模型（使用Pymongo）
3. Pinecone的向量存储结构
包含必要的字段：id、标题、内容、部门、创建时间、更新时间等。CodeBuddy生成的代码：
# PostgreSQL 元数据模型（ORM）
from sqlalchemy import Column, Integer, String, DateTime, JSON
from sqlalchemy.ext.declarative import declarative_base
import datetime
Base = declarative_base()
class DocumentMetadata(Base):
__tablename__ = 'document_metadata'
id = Column(Integer, primary_key=True)
title = Column(String(500), nullable=False)
department = Column(String(100), nullable=False)
tags = Column(JSON) # 存储标签数组
mongo_document_id = Column(String(24)) # 关联MongoDB的ID
pinecone_vector_id = Column(String(100)) # 关联Pinecone的ID
created_at = Column(DateTime, default=datetime.datetime.utcnow)
updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
is_active = Column(Integer, default=1)
# MongoDB 文档模型
class MongoDBDocument:
def __init__(self, content, format_type="markdown"):
self.content = content
self.format_type = format_type
self.created_at = datetime.datetime.utcnow()
def to_dict(self):
return {
"content": self.content,
"format_type": self.format_type,
"created_at": self.created_at
}
# Pinecone 向量模型
class VectorRecord:
def __init__(self, vector_id, values, metadata):
self.id = vector_id
self.values = values
self.metadata = metadata混合检索算法的实现
这是最复杂的部分，我需要实现一个将三种数据库结果融合的算法。我向CodeBuddy描述了具体需求：
我的提示词：
请实现一个混合检索算法类，包含以下方法：
1. 关键词搜索：在PostgreSQL中搜索标题和标签
2. 向量搜索：在Pinecone中搜索相似向量
3. 结果融合：对两种结果进行加权评分排序
4. 从MongoDB中获取最终内容
要求支持权重配置和分页参数。CodeBuddy提供的初始实现：
import logging
from typing import List, Dict, Any
from collections import defaultdict
class HybridSearchEngine:
def __init__(self, pg_db, mongo_db, pinecone_index, weights=None):
self.pg_db = pg_db
self.mongo_db = mongo_db
self.pinecone_index = pinecone_index
self.weights = weights or {
'keyword_title': 0.4,
'keyword_tag': 0.3,
'vector_similarity': 0.7
}
self.logger = logging.getLogger(__name__)
async def keyword_search(self, query: str, limit: int = 10) -> List[Dict]:
"""在PostgreSQL中执行关键词搜索"""
try:
# 构建SQL查询（使用SQLAlchemy核心）
search_stmt = f"""
SELECT id, title, department, tags,
ts_rank(to_tsvector('english', title), plainto_tsquery('english', :query)) as title_rank,
ts_rank(to_tsvector('english', array_to_string(tags, ' ')), plainto_tsquery('english', :query)) as tag_rank
FROM document_metadata
WHERE is_active = 1 AND (
to_tsvector('english', title) @@ plainto_tsquery('english', :query) OR
to_tsvector('english', array_to_string(tags, ' ')) @@ plainto_tsquery('english', :query)
)
ORDER BY (title_rank * {self.weights['keyword_title']} + tag_rank * {self.weights['keyword_tag']}) DESC
LIMIT :limit
"""
result = await self.pg_db.execute(search_stmt, {'query': query, 'limit': limit})
records = result.fetchall()
return [dict(record) for record in records]
except Exception as e:
self.logger.error(f"Keyword search failed: {str(e)}")
return []
async def vector_search(self, query_embedding: List[float], limit: int = 10) -> List[Dict]:
"""在Pinecone中执行向量搜索"""
try:
# Pinecone查询
results = self.pinecone_index.query(
vector=query_embedding,
top_k=limit * 2, # 多取一些用于后续过滤
include_metadata=True
)
# 处理结果
vector_results = []
for match in results.matches:
metadata = match.metadata or {}
vector_results.append({
'id': match.id,
'score': match.score,
'metadata': metadata
})
return vector_results
except Exception as e:
self.logger.error(f"Vector search failed: {str(e)}")
return []
async def hybrid_search(self, query: str, query_embedding: List[float], limit: int = 10) -> List[Dict]:
"""执行混合搜索并融合结果"""
# 并行执行两种搜索
keyword_results = await self.keyword_search(query, limit * 2)
vector_results = await self.vector_search(query_embedding, limit * 2)
# 结果融合
fused_results = self.fuse_results(keyword_results, vector_results, limit)
# 获取完整内容
final_results = await self.enrich_with_content(fused_results)
return final_results
def fuse_results(self, keyword_results, vector_results, limit: int) -> List[Dict]:
"""融合关键词和向量搜索结果"""
scored_documents = defaultdict(float)
doc_metadata = {}
# 处理关键词搜索结果
for idx, result in enumerate(keyword_results):
doc_id = result['id']
# 计算关键词搜索的标准化分数（0-1范围）
keyword_score = 1.0 / (idx + 1) # 排名倒数作为分数
scored_documents[doc_id] += keyword_score * 0.5 # 关键词权重
doc_metadata[doc_id] = result
# 处理向量搜索结果
for idx, result in enumerate(vector_results):
doc_id = result['metadata'].get('pg_id')
if not doc_id:
continue
# 向量相似度分数（Pinecone返回的score已经是0-1范围）
vector_score = result['score']
scored_documents[doc_id] += vector_score * 0.5 # 向量权重
# 补充元数据
if doc_id not in doc_metadata:
doc_metadata[doc_id] = result['metadata']
# 按融合分数排序
sorted_docs = sorted(
[(doc_id, score) for doc_id, score in scored_documents.items()],
key=lambda x: x[1],
reverse=True
)[:limit]
# 构建最终结果
final_results = []
for doc_id, score in sorted_docs:
result = doc_metadata.get(doc_id, {})
result['fusion_score'] = score
final_results.append(result)
return final_results
async def enrich_with_content(self, search_results: List[Dict]) -> List[Dict]:
"""从MongoDB获取完整内容"""
final_results = []
for result in search_results:
mongo_id = result.get('mongo_document_id')
if not mongo_id:
continue
try:
# 从MongoDB获取内容
content_doc = await self.mongo_db.documents.find_one({"_id": mongo_id})
if content_doc:
result['content'] = content_doc.get('content', '')
result['content_preview'] = content_doc.get('content', '')[:200] + '...' # 预览
final_results.append(result)
except Exception as e:
self.logger.error(f"Failed to fetch content from MongoDB: {str(e)}")
continue
return final_results这个初始实现已经很不错，但我发现了一些需要优化的地方，于是进行了进一步的迭代。
第三阶段：问题排查与优化
遇到的Bug：向量维度不匹配错误
在测试过程中，我遇到了一个棘手的错误：
Bug现象：
pinecone.core.client.exceptions.BadRequestError:
The dimension of the vector (1536) does not match the dimension of the index (512)排查步骤：
- 首先检查了Pinecone索引的配置，确认创建时设置的维度是1536
- 检查了OpenAI嵌入模型的输出维度，text-embedding-ada-002确实是1536维
- 发现在创建索引后，代码中有一处硬编码的测试数据使用了512维向量
CodeBuddy辅助排查：
我向CodeBuddy描述了这个问题：
我的提示词：
我在使用Pinecone时遇到维度不匹配错误：
- 索引配置维度：1536
- 模型输出维度：1536
- 但错误显示有512维的向量
请帮我分析可能的原因，并提供排查建议。CodeBuddy的响应：
可能的原因和排查步骤：
1. **残留测试数据**：
- 检查代码中是否有硬编码的测试向量
- 搜索代码库中是否有 `[0.1] * 512` 这样的模式
2. **索引混淆**：
- 确认使用的索引名称是否正确
- 检查环境变量是否指向了不同的索引
3. **缓存问题**：
- 清除向量缓存或重新初始化索引连接
4. **数据污染**：
- 检查数据库中是否存错了维度的向量
建议添加维度验证代码：CodeBuddy提供的验证代码：
def validate_vector_dimension(vector: List[float], expected_dim: int = 1536):
"""验证向量维度是否符合预期"""
if len(vector) != expected_dim:
raise ValueError(f"Vector dimension mismatch. Expected {expected_dim}, got {len(vector)}. "
f"First few values: {vector[:10]}")
return True
# 在向量入库前调用
validate_vector_dimension(embedding_vector)通过这个验证函数，我很快定位到了问题所在：有一处测试代码使用了旧的向量生成方法。修复后问题解决。
性能优化：数据库连接池配置
在压力测试中，系统出现了数据库连接耗尽的问题。我向CodeBuddy寻求优化建议：
我的提示词：
我的混合检索系统在高并发下出现PostgreSQL连接耗尽错误。当前使用SQLAlchemy的默认配置。请提供数据库连接池的最佳实践配置，包括：
1. 连接池大小设置
2. 超时和回收策略
3. 异步支持配置CodeBuddy提供的配置：
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import AsyncAdaptedQueuePool
# 异步数据库引擎配置
def create_db_engine(db_url: str):
return create_async_engine(
db_url,
poolclass=AsyncAdaptedQueuePool,
pool_size=20, # 连接池大小
max_overflow=10, # 最大溢出连接数
pool_timeout=30, # 获取连接超时时间(秒)
pool_recycle=1800, # 连接回收时间(秒)
pool_pre_ping=True, # 连接前ping检测
echo=False, # 生产环境设为False
future=True
)
# 会话工厂配置
async def get_db_session():
engine = create_db_engine(DATABASE_URL)
async_session = sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False
)
return async_session
# 使用示例
async def query_keywords(query: str):
async with async_session() as session:
try:
result = await session.execute(some_query)
return result.scalars().all()
finally:
await session.close()这个配置显著改善了系统的并发处理能力。
第四阶段：最终实现与效果
经过多次迭代和优化，最终的混合检索系统实现了以下功能：
核心特性
- 三重数据库协同：
- PostgreSQL：存储和管理文档元数据
- MongoDB：存储原始文档内容
- Pinecone：处理语义向量检索
- 智能结果融合：
- 动态加权评分算法
- 基于排名的结果归一化
- 可配置的权重参数
- 高效异步处理：
- 全异步数据库操作
- 并行查询执行
性能数据
- 查询延迟：平均200-400ms（包括向量生成时间）
- 并发支持：单节点支持100+并发查询
- 准确率提升：相比纯关键词搜索，召回率提升65%
关键代码片段
主API端点：
@app.post("/search/hybrid")
async def hybrid_search(request: SearchRequest):
"""混合搜索端点"""
try:
# 生成查询向量
embedding = await openai_client.create_embedding(request.query)
query_vector = embedding.data[0].embedding
# 执行混合搜索
results = await search_engine.hybrid_search(
query=request.query,
query_embedding=query_vector,
limit=request.limit
)
return {
"success": True,
"data": results,
"count": len(results)
}
except Exception as e:
logger.error(f"Search failed: {str(e)}")
return JSONResponse(
status_code=500,
content={"success": False, "error": str(e)}
)数据同步服务：
async def process_document_update(doc_id: str):
"""处理文档更新和向量化"""
try:
# 1. 从MongoDB获取内容
content_doc = await mongo_db.documents.find_one({"_id": doc_id})
if not content_doc:
return False
# 2. 生成嵌入向量
content = content_doc.get('content', '')
embedding = await openai_client.create_embedding(content)
vector = embedding.data[0].embedding
# 3. 更新向量数据库
vector_id = f"doc_{doc_id}"
await pinecone_index.upsert([(vector_id, vector, {"pg_id": doc_id})])
# 4. 更新PostgreSQL元数据
await pg_db.execute(
"UPDATE document_metadata SET pinecone_vector_id = :vector_id WHERE id = :doc_id",
{"vector_id": vector_id, "doc_id": doc_id}
)
return True
except Exception as e:
logger.error(f"Failed to process document update: {str(e)}")
# 重试逻辑
await retry_operation(doc_id)
return FalseAI协作经验总结
成功的协作模式
- 架构设计伙伴：CodeBuddy在技术选型和架构设计阶段提供了有价值的建议，帮助避免了常见的设计陷阱。
- 代码生成助手：对于样板代码和常见模式，CodeBuddy能够快速生成高质量的基础实现，节省了大量编码时间。
- 调试排查伙伴：遇到难以理解的错误时，向CodeBuddy描述问题往往能得到有针对性的排查建议和解决方案。
- 最佳实践来源：数据库配置、性能优化等方面的最佳实践是CodeBuddy的强项。
需要注意的方面
- 代码审查必要：AI生成的代码需要仔细审查，特别是业务逻辑和安全方面。
- 版本兼容性：需要检查生成代码与当前项目依赖版本的兼容性。
- 过度依赖风险：不能完全依赖AI，开发者的架构设计和业务理解仍然是核心。
效率提升数据
- 开发时间减少：相比传统开发，节省约40%的编码时间
- 代码质量提升：生成的样板代码质量一致，减少了低级错误
- 知识获取加速：快速获取新技术的最佳实践，学习曲线大幅降低
结论
通过CodeBuddy AI的协作，我成功构建了一个复杂的三数据库混合检索系统。这种协作模式不仅提高了开发效率，还在架构设计、问题排查和性能优化等方面提供了 valuable 的支持。
关键成功因素在于：
- 明确的需求描述和问题定义
- 对AI生成内容的批判性审查和调整
- 将AI作为增强能力而非替代品的态度
这种AI辅助的开发模式代表了软件工程的新范式，特别是在处理复杂系统集成和多技术栈项目时，AI工具能够显著降低认知负荷和开发成本。
本次基于 CodeBuddy AI 的协作开发实战表明，AI 助手已经能够成为软件开发过程中一位强大的“副驾驶”。它不仅加速了代码编写，更在系统架构、技术选型和故障排查等高层思维活动中提供了关键支持。通过构建一个融合了 PostgreSQL、MongoDB 和 Pinecone 的混合检索系统，我们成功地将结构化查询的精确性、文档存储的灵活性以及向量检索的语义理解能力结合为一体，有效解决了传统搜索的局限性。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
评论
登录后参与评论
推荐阅读
目录
腾讯云开发者
