def generate_questions_for_chunk(chunk: str, llm) -> list[str]:
    prompt = f"""
    请根据下面的知识切片生成用户可能提出的问题。
    要求覆盖直接事实、比较问题、推理问题和场景化问题。

    知识切片：
    {chunk}
    """
    return llm.invoke(prompt).splitlines()


def build_dual_bm25_index(chunks: list[str]):
    original_index = BM25Okapi([tokenize(c) for c in chunks])

    generated_questions = []
    question_to_chunk = []
    for idx, chunk in enumerate(chunks):
        questions = generate_questions_for_chunk(chunk, llm)
        generated_questions.extend(questions)
        question_to_chunk.extend([idx] * len(questions))

    question_index = BM25Okapi([tokenize(q) for q in generated_questions])
    return original_index, question_index, question_to_chunk

{
  "knowledge_type": "事实 | 需求 | 问题 | 流程 | 注意",
  "content": "可沉淀的知识内容",
  "confidence": 0.95,
  "source": "conversation_id",
  "keywords": ["门票", "营业时间", "停车"],
  "category": "旅游服务",
  "conversation_summary": "本轮对话摘要",
  "user_intent": "用户真实意图"
}

def extract_knowledge_from_conversation(conversation):
    """从一段对话中抽取结构化知识点。"""
    return llm_extract(conversation)


def filter_reusable_knowledge(items):
    """过滤临时需求和一次性问题，只保留可复用知识。"""
    reusable_types = {"事实", "流程", "注意"}
    return [item for item in items if item["knowledge_type"] in reusable_types]


def merge_similar_knowledge(items):
    """按类型和主题聚类，使用 LLM 合并重复知识。"""
    groups = group_by_type_and_topic(items)
    return [llm_merge(group) for group in groups]

def check_knowledge_base_health(kb, test_queries, current_date):
    missing = check_missing_knowledge(kb, test_queries)
    outdated = check_outdated_knowledge(kb, current_date)
    conflicts = check_conflicting_knowledge(kb)

    return {
        "coverage_score": score_coverage(missing),
        "freshness_score": score_freshness(outdated),
        "consistency_score": score_consistency(conflicts),
        "missing_knowledge": missing,
        "outdated_knowledge": outdated,
        "conflicts": conflicts,
    }

def compare_versions(v1, v2):
    v1_map = {item["id"]: item["content"] for item in v1}
    v2_map = {item["id"]: item["content"] for item in v2}

    added = set(v2_map) - set(v1_map)
    removed = set(v1_map) - set(v2_map)
    modified = [
        key for key in set(v1_map) & set(v2_map)
        if v1_map[key] != v2_map[key]
    ]

    return {"added": added, "removed": removed, "modified": modified}


def evaluate_version(version_name, test_queries):
    correct = 0
    total_time = 0

    for item in test_queries:
        chunks, latency = retrieve_relevant_chunks(item["query"], version_name)
        if item["expected_answer"] in " ".join(chunks):
            correct += 1
        total_time += latency

    return {
        "accuracy": correct / len(test_queries),
        "avg_response_time": total_time / len(test_queries),
    }

from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.llms import Tongyi

embeddings = DashScopeEmbeddings(
    model="text-embedding-v1",
    dashscope_api_key=DASHSCOPE_API_KEY,
)

knowledge_base = FAISS.from_texts(chunks, embeddings)

llm = Tongyi(
    model_name="deepseek-v3",
    dashscope_api_key=DASHSCOPE_API_KEY,
)

docs = knowledge_base.similarity_search(query, k=10)

def generate_multi_queries(query: str, llm, num_queries: int = 3) -> list[str]:
    prompt = f"""
    你是一个 AI 助手，负责生成多个不同视角的搜索查询。
    给定一个用户问题，生成 {num_queries} 个不同但相关的查询，
    以帮助检索更全面的信息。

    原始问题：{query}

    请直接输出 {num_queries} 个查询，每行一个，不要编号。
    """
    return parse_lines(llm.invoke(prompt))


def multi_query_search(query, retriever, llm):
    queries = [query] + generate_multi_queries(query, llm)
    docs = []
    for q in queries:
        docs.extend(retriever.search(q, k=4))
    return deduplicate(docs)

客户经理被投诉了，投诉一次扣多少分
客户经理投诉扣分标准是什么
银行客户经理被投诉一次会扣除多少绩效分
金融机构客户经理投诉处罚机制

def hybrid_search(query: str, alpha: float = 0.5, k: int = 5):
    tokenized_query = tokenize_chinese(query)
    bm25_scores = bm25.get_scores(tokenized_query)
    max_bm25 = max(bm25_scores) if max(bm25_scores) > 0 else 1
    bm25_scores = [score / max_bm25 for score in bm25_scores]

    vector_results = vectorstore.similarity_search_with_score(
        query,
        k=len(chunks),
    )

    max_distance = max(distance for _, distance in vector_results)
    vector_scores = {}
    for doc, distance in vector_results:
        idx = doc.metadata["chunk_id"]
        vector_scores[idx] = 1 - (distance / max_distance)

    combined = []
    for idx in range(len(chunks)):
        score = alpha * vector_scores.get(idx, 0) + (1 - alpha) * bm25_scores[idx]
        combined.append((idx, score))

    return sorted(combined, key=lambda x: x[1], reverse=True)[:k]

from transformers import AutoModelForSequenceClassification, AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("BAAI/bge-reranker-base")
model = AutoModelForSequenceClassification.from_pretrained("BAAI/bge-reranker-base")
model.eval()

pairs = [
    [query, doc.page_content]
    for doc in candidate_docs
]

inputs = tokenizer(
    pairs,
    padding=True,
    truncation=True,
    max_length=512,
    return_tensors="pt",
)

scores = model(**inputs).logits.squeeze(-1).tolist()

def hybrid_multi_query_search_with_rerank(
    query,
    hybrid_retriever,
    reranker,
    llm,
    initial_k=10,
    final_k=4,
):
    queries = [query] + generate_multi_queries(query, llm)

    candidate_docs = []
    for q in queries:
        docs = hybrid_retriever.search(q, k=initial_k)
        candidate_docs.extend(docs)

    candidate_docs = deduplicate(candidate_docs)
    reranked_docs = reranker.rerank(query, candidate_docs, top_k=final_k)
    return reranked_docs

def query_to_doc(query: str, llm) -> str:
    prompt = f"""
    请把下面的短查询改写成一段用于检索的技术说明。
    要求保留原始意图，补充可能相关的术语和上下文。

    查询：{query}
    """
    return llm.invoke(prompt)

def doc_to_query(chunk: str, llm) -> list[str]:
    prompt = f"""
    请根据文档内容生成用户可能提出的问题。
    问题应覆盖事实、方法、比较、原因和场景化使用。

    文档内容：{chunk}
    """
    return parse_lines(llm.invoke(prompt))

small_index_items = [
    {
        "small_text": "Transformer 通过自注意力机制实现高效并行计算。",
        "parent_id": "paper_001_section_3",
    },
    {
        "small_text": "混合精度训练可以减少显存占用并提升训练速度。",
        "parent_id": "paper_002_section_4",
    },
]

parent_chunks = {
    "paper_001_section_3": "完整章节内容……",
    "paper_002_section_4": "完整章节内容……",
}


def small_to_big_retrieve(query):
    small_hits = small_vectorstore.search(query, k=5)
    parent_ids = deduplicate([hit.metadata["parent_id"] for hit in small_hits])
    return [parent_chunks[parent_id] for parent_id in parent_ids]

git clone https://github.com/microsoft/graphrag.git
cd graphrag
pip install -e .
graphrag init --root .

graphrag index --root .

graphrag query --root . --method global --query "和曹操相关的人物都有哪些?"
graphrag query --root . --method local --query "关羽战胜过哪些武将?"

def agentic_rag(question):
    plan = planner(question)

    evidence = []
    for sub_query in plan.sub_queries:
        docs = retriever.search(sub_query)
        ranked_docs = reranker.rerank(sub_query, docs)
        evidence.extend(ranked_docs)

    if not evidence_is_enough(question, evidence):
        return "当前知识库证据不足，无法可靠回答。"

    answer = generator(question, evidence)
    return verify_answer_with_evidence(answer, evidence)