从数据流动
理解每一行代码

# 知识库：5 段文本（真实场景来自 PDF/Wiki/手册）
DOCUMENTS = [
    "RAG 在生成前检索外部文档，让回答有据可查，有效减少幻觉。",
    "向量数据库通过余弦相似度找到语义最接近的文档块。",
    "LangChain 封装了文档加载、分块、检索、生成的完整流程。",
    "Transformer 的核心是注意力机制，让模型动态关注输入。",
    "RAG 的瓶颈在检索而非生成：垃圾进，垃圾出。",
]

def embed(text: str) -> np.ndarray:
    """
    调用 OpenAI Embedding API。
    为什么用 API 而不是本地模型？
    → 快速验证概念。v4 会讨论模型选型和本地部署。
    """
    response = client.embeddings.create(
        input=text,
        model="text-embedding-3-small"  # 1536 维，中英文均可
    )
    return np.array(response.data[0].embedding)  # shape: (1536,)

# 为所有文档建立索引
doc_embeddings = np.array([embed(doc) for doc in DOCUMENTS])
# doc_embeddings.shape → (5, 1536)   [文档数 × 向量维度]

def cosine_sim(a: np.ndarray, b: np.ndarray) -> float:
    return float(
        np.dot(a, b)                               # 点积
        / (np.linalg.norm(a) * np.linalg.norm(b))  # 除以两向量的模
    )

def retrieve(query: str, doc_embeddings: np.ndarray, top_k=2):
    q_emb = embed(query)                            # 问题也转成向量
    scores = [cosine_sim(q_emb, de) for de in doc_embeddings]
    ranked = sorted(
        enumerate(scores),
        key=lambda x: x[1], reverse=True
    )[:top_k]                                      # 取最高的 K 个
    return [{"text": DOCUMENTS[i], "score": s} for i, s in ranked]

def build_prompt(query: str, contexts: list) -> str:
    ctx = "\n".join(
        f"  [{i+1}] {c['text']}"
        for i, c in enumerate(contexts)
    )
    return f"""你是一个严谨的知识助手。请仅根据以下参考资料回答问题。
若参考资料中没有相关信息，请明确说"资料中未提及"，不要猜测。

参考资料：
{ctx}

问题：{query}
回答："""

# 构建后的 Prompt 长这样：
# ┌──────────────────────────────────────────────┐
# │ 你是一个严谨的知识助手...（系统指令）           │
# │ 参考资料：                                    │
# │   [1] RAG 的瓶颈在检索而非生成...             │  ← 检索结果注入
# │   [2] RAG 在生成前检索外部文档...             │  ← 检索结果注入
# │ 问题：RAG 系统里最关键的环节是什么？            │  ← 用户原问题
# └──────────────────────────────────────────────┘

# temperature=0：让输出稳定，便于对比，不是随机创作
def chat(prompt: str) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    return resp.choices[0].message.content.strip()

# 对比实验
ans_raw = chat(QUERY)              # ❌ 无 RAG，直接问
ans_rag = chat(prompt_rag)         # ✅ 有 RAG，带上下文

def chunk_fixed_no_overlap(text: str, chunk_size: int = 150) -> list[str]:
    """
    按字符数切割，走到哪切到哪。
    
    问题：可能在句子中间切断。
    例："推荐分块大小在 200 到 500 个字符之间，并设置重叠"
    → 块 A：...推荐分块大小在 200 到 500
    → 块 B：个字符之间，并设置重叠...
    两个块都是残缺信息。
    """
    text = text.strip()
    chunks, start = [], 0
    while start < len(text):
        chunk = text[start : start + chunk_size].strip()
        if chunk: chunks.append(chunk)
        start += chunk_size    # 每次直接跳 chunk_size，无回退
    return chunks

def chunk_fixed_with_overlap(text: str, chunk_size=200, overlap=40):
    text = text.strip()
    chunks, start = [], 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end].strip())
        next_start = end - overlap    # ← 核心：回退 overlap 个字符再开始下一块
        if next_start <= start: next_start = start + 1  # 防死循环
        start = next_start
        if start >= len(text): break
    return [c for c in chunks if len(c) > 20]

# 直觉理解 overlap=40 的效果：
#
#  位置:  0        200      360      520  ...
#          ├────────┤         ├────────┤
#  块 A:  [0      :200]
#  块 B:          [160     :360]   ← 从 200-40=160 开始
#  块 C:                  [320   :520]   ← 从 360-40=320 开始
#
#  160~200 这 40 个字符，同时出现在块 A 和块 B 里。
#  任何跨越 200 这个边界的句子，都会完整保存在某个块中。

def chunk_by_sentence(text: str, max_chars: int = 300) -> list[str]:
    # 按中文标点分句，保留标点（lookahead 不消耗字符）
    sentences = re.split(r'(?<=[。！？；\n])', text)
    sentences = [s.strip() for s in sentences if s.strip() and len(s) > 5]

    chunks, current = [], ""
    for sent in sentences:
        if len(current) + len(sent) <= max_chars:
            current += sent           # 累积句子，不超限就继续加
        else:
            if current: chunks.append(current.strip())
            current = sent             # 超限，保存旧块，开新块
    if current.strip():
        chunks.append(current.strip()) # 最后一块别忘了
    return [c for c in chunks if len(c) > 20]