1
class TextSplitter:
2
    def split(self, text: str, chunk_size: int = 500) -> list[str]:
3
        # 1. 文本标准化
4
        text = self.normalize(text)
5

6
        # 2. 按段落/句子切分
7
        chunks = self.split_by_paragraph(text)
8

9
        # 3. 合并过小片段
10
        chunks = self.merge_small_chunks(chunks, chunk_size)
11

12
        return chunks

1
# 固定大小切分
2
def fixed_split(text, chunk_size=500, overlap=50):
3
    chunks = []
4
    start = 0
5
    while start < len(text):
6
        end = start + chunk_size
7
        chunks.append(text[start:end])
8
        start = end - overlap
9
    return chunks
10

11
# 语义切分（保留句子完整性）
12
def semantic_split(text):
13
    sentences = split_by_sentence(text)
14
    chunks = []
15
    current = []
16
    for sent in sentences:
17
        current.append(sent)
18
        if len(' '.join(current)) > chunk_size:
19
            chunks.append(' '.join(current[:-1]))
20
            current = [sent]
21
    return chunks

1
# Query 优化
2
class QueryRewriter:
3
    def rewrite(self, query: str) -> str:
4
        # 1. 隐式表述展开
5
        query = self.expand_abbreviations(query)
6

7
        # 2. 同义词扩展
8
        query = self.expand_synonyms(query)
9

10
        # 3. 假设类型注入
11
        if "如何" in query:
12
            query += " 请提供具体步骤"
13

14
        return query

1
# 混合检索 = 关键词 + 向量
2
class HybridRetriever:
3
    def retrieve(self, query: str, top_k: int = 10):
4
        # 向量检索
5
        vector_results = self.vector_index.search(
6
            self.embed(query), top_k * 2
7
        )
8

9
        # 关键词检索
10
        bm25_results = self.bm25_index.search(query, top_k * 2)
11

12
        # RRF 融合
13
        fused = self.rrf_fusion(
14
            vector_results,
15
            bm25_results,
16
            top_k
17
        )
18

19
        return fused
20

21
# RRF (Reciprocal Rank Fusion)
22
def rrf(results_list, k=60):
23
    scores = {}
24
    for results in results_list:
25
        for i, doc in enumerate(results):
26
            doc_id = doc.id
27
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + i + 1)
28
    return sorted(scores.items(), key=lambda x: -x[1])

1
class ContextBuilder:
2
    def __init__(self, max_tokens: int = 4000):
3
        self.max_tokens = max_tokens
4

5
    def build(self, query: str, retrieved_docs: list) -> str:
6
        context_parts = [f"问题: {query}\n"]
7
        used_tokens = self.count_tokens(context_parts[0])
8

9
        for doc in retrieved_docs:
10
            doc_text = f"参考文档:\n{doc.content}"
11
            doc_tokens = self.count_tokens(doc_text)
12

13
            if used_tokens + doc_tokens > self.max_tokens:
14
                break
15

16
            context_parts.append(doc_text)
17
            used_tokens += doc_tokens
18

19
        return '\n\n'.join(context_parts)

1
class Reranker:
2
    def rerank(self, query: str, docs: list, top_k: int = 5):
3
        # 使用交叉编码器重排序
4
        pairs = [(query, doc.content) for doc in docs]
5
        scores = self.cross_encoder.predict(pairs)
6

7
        ranked = sorted(
8
            zip(docs, scores),
9
            key=lambda x: -x[1]
10
        )[:top_k]
11

12
        return [doc for doc, _ in ranked]

1
# RAG 评估指标
2
metrics = {
3
    # 检索质量
4
    "retrieval_recall": "相关文档召回比例",
5
    "retrieval_precision": "召回文档中相关比例",
6

7
    # 生成质量
8
    "answer_relevance": "答案与问题的相关性",
9
    "faithfulness": "答案对 Context 的忠实度",
10
    "answer_correctness": "答案正确性"
11
}

1
# RAG 系统架构
2
rag_pipeline:
3
  document_store: milvus
4
  vector_store: openai_embedder
5
  llm: gpt-4-turbo
6

7
  retrieval:
8
    top_k: 10
9
    rerank: true
10
    hybrid_search: true
11

12
  generation:
13
    max_tokens: 2000
14
    temperature: 0.1