大模型别再只写 RAG，太趋同了!

def classify_intent_rule(query: str) -> str:
    query = query.lower()  # 统一小写处理（如适用英文，此处对中文影响可忽略）
    # 定义关键词列表
    intent_keywords = {
        "报销流程查询": ["报销", "费用", "报销流程"],
        "保险产品销售技巧": ["卖保险", "销售", "技巧", "销售技巧"]
    }
    for intent, keywords in intent_keywords.items():
        if any(keyword in query for keyword in keywords):
            return intent
    return




    
 "未知意图"
测试
queries = ["怎么报销保险费用？", "新人怎么提升保险产品销售技巧？", "保险理赔需要哪些材料？"]
for q in queries:
    print(q, "->", classify_intent_rule(q))

from transformers import AutoTokenizer, AutoModelForSequenceClassification
#加载已训练好的中文BERT意图分类模型
tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese")
model = AutoModelForSequenceClassification.from_pretrained("./fine_tuned_intent_model")  # 本地路径
id2label = {0: "报销流程查询", 1: "保险产品销售技巧"}  # 意图ID到标签的映射
对用户查询进行分类
query = "怎么报销保险费用？"
inputs = tokenizer(query, return_tensors="pt")
outputs = model(**inputs)
pred_id = int(outputs.logits.argmax(dim=1))
pred_intent = id2label[pred_id]
print(f"查询: {query} -> 意图类别: {pred_intent}")

import openai
openai.api_key = "YOUR_API_KEY"# 设置API密钥
def classify_intent_llm(query: str) -> str:
    system_prompt = "你是一个智能助手，帮助分类用户意图。可能的意图类别包括：\n1. 报销流程查询\n2. 保险产品销售技巧\n只回复序号1或2。"
    user_prompt = f"用户问：{query}\n上述用户的意图属于哪个类别？"
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo"




    
,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ]
    )
    answer = response["choices"][0]["message"]["content"].strip()
    return"报销流程查询"if answer.startswith("1") else"保险产品销售技巧"if answer.startswith("2") else"未知"
测试 LLM 意图分类
print(classify_intent_llm("怎么报销保险费用？"))      # 预计输出 "报销流程查询"
print(classify_intent_llm("有哪些有效的保险产品销售技巧？"))  # 预计输出 "保险产品销售技巧"

import re
def rewrite_query(query: str) -> str:
    # 1. 去除疑问口语用词
    fillers = ["请问", "一下", "呢", "啊", "吧"]
    for f in fillers:
        query = query.replace(f, "")
    query = query.strip()
    # 2. 将“怎么XXX”改写为“XXX是什么？”
    if query.startswith(("怎么", "怎样")):
        main_query = query[2:]  # 去掉开头的"怎么"/"怎样"




    

        # 如果以问号结尾则去掉问号，稍后统一加
        main_query = main_query.rstrip("？?")
        # 示例中特定的短语调整：将“报销保险费用”改为“保险费用报销”
        if"报销保险费用"in main_query:
            main_query = main_query.replace("报销保险费用", "保险费用报销")
        # 若句尾缺少“流程”，根据上下文添加
        if main_query.endswith("报销") or main_query.endswith("费用"):
            main_query += "流程"
        # 添加结尾的问句形式
        query = main_query + "是什么"
    # 3. 确保以问号结尾
    query = query.rstrip("？?") + "？"
    return query
测试重写函数
original_query = "怎么报销保险费用？"
rewritten_query = rewrite_query(original_query)
print("原始查询：", original_query)
print("重写后的查询：", rewritten_query)

原始查询： 怎么报销保险费用？
重写后的查询： 保险费用报销流程是什么？

构建一个简单的同义词词典
synonym_dict = {
    "保险理赔": ["保险索赔", "理赔", "索赔流程", "理赔流程"],
    "理赔": ["索赔", "赔付"],
    # 其他词的同义词...
}
def expand_query(query: str) -> [str]:
    expansions = set()
    # 如果查询短语本身在词典中
    if query in synonym_dict:
        expansions.update(synonym_dict[query])
    # 将查询拆分为词（简单按字符，这里假设输入短语本身是一个词或固定短语）




    

    for term, syns in synonym_dict.items():
        if term in query:
            expansions.update(syns)
    # 加入原始查询本身
    expansions.add(query)
    return list(expansions)
测试查询扩展
query = "保险理赔"
expanded_queries = expand_query(query)
print("原始查询：", query)
print("扩展结果：", expanded_queries)

原始查询： 保险理赔  
扩展结果： ['保险理赔', '保险索赔', '理赔流程', '索赔流程', '索赔', '理赔']

import openai
from sentence_transformers import SentenceTransformer
openai.api_key = "YOUR_API_KEY"
def hyde_retrieval(query: str, embed_model, doc_embeddings, doc_ids):
    # 1. 使用LLM生成假设文档
    prompt = f"请针对以下问题给出详细的回答：{query}"
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    hypo_doc = response["choices"][0]["message"]["content"]

    # 2. 将生成的文档进行向量化





    
    query_vector = embed_model.encode(hypo_doc)

    # 3. 在向量空间中检索相似文档
    # （这里假设已有知识库文档向量 doc_embeddings 和对应的 doc_ids 列表）
    # 计算与所有文档向量的余弦相似度，并选取最高的若干
    import numpy as np
    sims = np.dot(doc_embeddings, query_vector)
    top_idx = sims.argsort()[-5:][::-1]  # 取前5个相似度最高的文档索引
    results = [(doc_ids[i], sims[i]) for i in top_idx]
    return results
初始化句向量模型（例如中文多语言模型）
embed_model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
假设我们有预先计算好的文档向量和IDs
doc_embeddings = ...  # shape: (N_docs, dim)
doc_ids = ...         # 文档ID或内容索引
对查询进行 HyDE 检索
query = "保险销售技巧"
retrieved_docs = hyde_retrieval(query, embed_model, doc_embeddings, doc_ids)
print("HyDE 检索结果文档ID及分数：", retrieved_docs)

import fitz  # PyMuPDF for PDF
from pptx import Presentation  # python-pptx for PPT
from PIL import Image
import pytesseract
import re
def ocr_image(image):
    """OCR识别图像，返回文本字符串。"""
    # 使用简体中文+英文识别，配置可根据需要调整
    config = "--psm 6"# 假设6: 把图像看作一个统一块
    text = pytesseract.image_to_string(image, config=config, lang='chi_sim+eng')
    return text
def parse_document(file_path):
    """解析文档（PDF/PPT/文本），返回文本块列表，每块包含内容和类型等标记。"""
    blocks = []
    if file_path.lower().endswith('.pdf'):
        doc = fitz.open(file_path)
        for page in doc:
            # 尝试直接提取文本
            text = page.get_text("text")
            if text.strip():





    
                # 简单按换行拆分为块，可进一步按段落细分
                lines = text.splitlines()
            else:
                # 若无文本（扫描页），则OCR整个页面
                pix = page.get_pixmap(dpi=150)
                img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
                ocr_text = ocr_image(img)
                lines = ocr_text.splitlines()
            # 将提取的行转换为块初步列表，标记来源页
            for line in lines:
                if line.strip() == "":
                    continue
                blocks.append({
                    "text": line.strip(),
                    "type": "text",  # 初始默认为普通文本，后续再调整类型
                    "page": page.number + 1
                })
    elif file_path.lower().endswith('.pptx'):
        prs = Presentation(file_path)
        for idx, slide in enumerate(prs.slides, start=1):
            title = ""
            if slide.shapes.title:
                title = slide.shapes.title.text.strip()
                if title:
                    # 幻灯片标题作为单独块
                    blocks.append({
                        "text": title,
                        "type": "heading",
                        "level": 1,  # 幻灯片题目视为一级标题
                        "slide": idx
                    })
            # 提取其他文本框
            for shape in slide.shapes:
                ifnot shape.has_text_frame or shape.text.strip() == "":
                    continue
                text = shape.text.strip()
                # 如果文本与标题相同就跳过（已添加）
                if title and text == title:
                    continue
                # 按换行将文本框内容拆成行块（对应子弹列表逐行）
                for line in text.splitlines():
                    if line.strip() == "":
                        continue
                    blocks.append({
                        "text": line.strip(),
                        "type": "text",
                        "slide": idx
                    })
    elif file_path.lower().endswith('.txt'):
        with open(file_path, 'r', encoding='utf-8') as f:
            for line in f:
                if line.strip():
                    blocks.append({
                        




    
"text": line.strip(),
                        "type": "text"
                    })
    else:
        # 其他格式（如视频）：假设已转换为字幕文本
        # 这里直接读取同名的字幕文本文件
        subs_path = file_path + ".txt"
        try:
            with open(subs_path, 'r', encoding='utf-8') as f:
                for line in f:
                    if line.strip():
                        blocks.append({
                            "text": line.strip(),
                            "type": "text"
                        })
        except FileNotFoundError:
            print("Unsupported file format or missing transcript:", file_path)
    return blocks
def split_and_tag_blocks(blocks):
    """根据内容将文本块切分、合并，并标注层级和类别标签。"""
    chunk_list = []
    hierarchy = []  # 用于跟踪当前层级标题栈
    current_chunk = {"text": "", "meta": {}}  # 临时聚合当前chunk内容
    for blk in blocks:
        text = blk["text"]
        blk_type = blk.get("type", "text")
        # 检测标题：根据内容格式判断
        is_heading = blk_type == "heading"
        level = blk.get("level", None)
        ifnot is_heading:
            # 简单规则：长度较短且以冒号结束，或符合编号模式的，视为标题
            if len(text) 20and (text.endswith(":") or text.endswith("：") or re.match(r'^[\d一二三]+\D', text)):
                is_heading = True
                # 确定层级level（基于数字章节或者默认1级）
                m = re.match(r'^(\d+(\.\d+)*)', text)
                if m:
                    level = m.group(1).count('.') + 1
                else:
                    level = 1
                blk_type = "heading"
        if is_heading:
            # 遇到新标题块，先结束上一chunk
            if current_chunk["text"]:
                chunk_list.append(current_chunk)
                current_chunk = {"text": "", "meta": {}}





    
            # 更新层级栈
            if level isNone:
                level = 1
            # 调整hierarchy列表长度
            if level - 1 < len(hierarchy):
                hierarchy = hierarchy[:level-1]
            # 确保hierarchy长度足够
            while len(hierarchy) < level-1:
                hierarchy.append("")
            # 设置当前级别标题
            if len(hierarchy) < level:
                hierarchy.append(text)
            else:
                hierarchy[level-1] = text
            # 将标题本身作为一个Chunk的元数据存入层级，但内容不上屏检索
            # （可以选择是否将标题内容并入chunk文本，这里不直接作为内容）
            # 开始一个新的chunk上下文，继承层级meta
            current_chunk["meta"]["section"] = " > ".join(hierarchy)
            current_chunk["meta"]["type"] = "text"# 默认文本类型
            continue# 不把标题行本身当做内容
        # 非标题块：
        # 检测表格行：根据是否包含制表符或对齐空格判断
        if re.search(r'\t', text) or re.search(r'\s{2,}', text):
            blk_type = "table"
        # 检测代码块：根据常见代码特征，如以4空格开头，或包含 {}; 等（简单判断）
        if re.match(r'^\s{4}', text) or re.search(r'[{};]', text):
            # 注意：真实环境可结合格式或```标记判断
            blk_type = "code"
        # 如果当前chunk已存在且本块类型与当前chunk类型不同，而且当前chunk不空，则开启新chunk
        if current_chunk["text"] and current_chunk["meta"].get("type") != blk_type:
            # 结束之前的chunk
            chunk_list.append(current_chunk)
            current_chunk = {"text": "", "meta": {}}
        # 设置chunk元数据（继承当前层级路径和类型）
        if"section"notin current_chunk["meta"]:
            current_chunk["meta"]["section"] = " > ".join(hierarchy) if hierarchy else""
        current_chunk["meta"]["type"] = blk_type
        # 累加文本（表格和代码保持原格式，普通文本在块内加空格拼接）
        if blk_type == "text":




    

            if current_chunk["text"]:
                # 若当前已有内容，先检查语义连贯（比如前句末尾无句号，则直接接续）
                if current_chunk["text"].strip().endswith(tuple("。？！!?.")):
                    current_chunk["text"] += "\n" + text
                else:
                    current_chunk["text"] += " " + text  # 前一句未完，空格连接
            else:
                current_chunk["text"] = text
        else:
            # 表格或代码块，保持行作为换行
            current_chunk["text"] += (text + "\n")
    # 循环结束后，将最后的chunk加入列表
    if current_chunk["text"]:
        chunk_list.append(current_chunk)
    return chunk_list
示例：调用解析和切分函数（文件路径需换成实际存在的文件）
file_path = "保险公司报销制度.pdf"
blocks = parse_document(file_path)
chunks = split_and_tag_blocks(blocks)
for ch in chunks:
print(f"[{ch['meta'].get('section')}] ({ch['meta'].get('type')}) {ch['text'][:50]}...")

import math
import numpy as np
示例文档语料
docs = {





    
    1: "公司报销制度包括差旅费报销流程和标准。",
    2: "保险产品的销售技巧包括如何挖掘客户需求和突出产品优势。"
}
构建倒排索引用于BM25（简单实现：记录每个词在哪些文档出现及频次）
inverted_index = {}
doc_lengths = {}
for doc_id, text in docs.items():
    words = list(text)  # 这里简单将每个字作为词，实际应用需分词
    doc_lengths[doc_id] = len(words)
    forw in words:
        ifw not in inverted_index:
            inverted_index[w] = {}
        inverted_index[w][doc_id] = inverted_index[w].get(doc_id, 0) + 1
计算IDF值
N = len(docs)
idf = {}
for term, doc_dict in inverted_index.items():
    df = len(doc_dict)
    idf[term] = math.log((N - df + 0.5) / (df + 0.5) + 1)  # BM25 IDF公式
def bm25_search(query, k1=1.5, b=0.75, top_k=5):
    """简单BM25检索实现，返回doc_id及BM25分数"""
    # 分词
    query_terms = list(query)
    scores = {}
    avgdl = sum(doc_lengths.values()) / N
    for term in query_terms:
        if term not in inverted_index:
            continue
        for doc_id, tf in inverted_index[term].items():
            # BM25公式计算评分
            score = idf.get(term, 0) * (tf * (k1 + 1)) / (tf + k1 * (1 - b + b * (doc_lengths[doc_id] / avgdl)))
            scores[doc_id] = scores.get(doc_id, 0) + score
    # 返回按分数排序的Top K文档
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
模拟向量检索：计算简单的TF-IDF向量并用余弦相似度
这里为了模拟，同样使用字频作为向量表示
def vector_search(query, top_k=5):
    query_terms = list(query)
    # 计算查询向量（TF-IDF）
    query_vec = []
    vocab = list(inverted_index.keys())
    for term in vocab:
        tf




    
 = query_terms.count(term)
        query_vec.append(tf * idf.get(term, 0))
    query_vec = np.array(query_vec)
    # 计算每个文档的向量并求余弦相似度
    scores = {}
    for doc_id, text in docs.items():
        doc_terms = list(text)
        doc_vec = []
        for term in vocab:
            tf = doc_terms.count(term)
            doc_vec.append(tf * idf.get(term, 0))
        doc_vec = np.array(doc_vec)
        # 计算余弦相似度
        if np.linalg.norm(doc_vec) == 0or np.linalg.norm(query_vec) == 0:
            cos_sim = 0.0
        else:
            cos_sim = np.dot(query_vec, doc_vec) / (np.linalg.norm(query_vec) * np.linalg.norm(doc_vec))
        scores[doc_id] = cos_sim
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
定义查询
queries = {
    "Q1": "报销制度",       # 短查询，期待BM25擅长
    "Q2": "如何推销保险产品"  # 开放式查询，语义同义词“推销”≈“销售”，期待向量检索擅长
}
for qid, query in queries.items():
    bm25_results = bm25_search(query)
    vec_results = vector_search(query)
    print(f"\n查询: {query}")
    print("BM25结果:", bm25_results)
    print("向量检索结果:", vec_results)
    # 合并结果（简单示例：取并集，不同来源结果赋不同初始分数权重）
    combined_scores = {}
    for doc_id, score in bm25_results:
        combined_scores[doc_id] = combined_scores.get(doc_id, 0) + 0.6 * (score / (bm25_results[0][1] if bm25_results else1))
    for doc_id, score in vec_results:
        combined_scores[doc_id] = combined_scores.get(doc_id, 0) + 0.4 * score  # 假设向量相似度已是0-1
    final_ranked = sorted(combined_scores.items(), key=lambda x: x[1], reverse=True)
    print("合并后Top结果:"




    
, final_ranked)

from




    
 sentence_transformers import SentenceTransformer, InputExample, losses, models
from torch.utils.data import DataLoader
假设我们使用一个预训练的中文嵌入模型，例如BAAI发布的BGE模型
model = SentenceTransformer('BAAI/bge-base-zh')  # 加载预训练BGE中文基座模型
准备训练数据: List of InputExample(question, positive_passage)
train_examples = [
    InputExample(texts=["什么是保单的现金价值？", "保单的现金价值是指保单在退保或某些情况下可领取的金额。"]),
    InputExample(texts=["保险合同的冷静期有多长？", "一般人寿保险合同都有10天的冷静期，在此期间可以无条件退保。"]),
    # ... 更多问答对 ...
]
如果有负例，可以在texts加入第三项; 只有正例时MultipleNegativesRankingLoss会自动负采样
train_dataloader = DataLoader(train_examples, batch_size=16, shuffle=True)
定义训练损失为多负样本排名损失（适用于只有正向对的情况）
train_loss = losses.MultipleNegativesRankingLoss(model)
(可选) 定义评估方法: 使用SentenceTransformer提供的InformationRetrievalEvaluator
from sentence_transformers import evaluation
evaluator = evaluation.InformationRetrievalEvaluator(query_embeddings, corpus_embeddings, relevant_docs)
配置训练参数并训练
num_epochs = 1
warmup_steps = 100
model.fit(
    train_objectives=[(train_dataloader, train_loss)],
    epochs=num_epochs,
    warmup_steps=warmup_steps,
    show_progress_bar=True
)
微调完成后保存模型
model.save("finetuned-bge-insurance")

from sentence_transformers import CrossEncoder
query = "最近公司的车险理赔流程是什么？"
candidates = [
    "公司2020年的车险理赔流程包括报案、查勘定损、提交资料、理赔审核和赔付。",
    "最新的车险理赔流程（2023年更新）为：在线报案->现场查勘->材料上传->理算审核->赔款支付。",
    "车险理赔是指车辆发生保险事故后，被保险人向保险公司申请赔偿的过程。"
]
初始化Cross-Encoder重排模型（此示例用英文MiniLM模型，占位）
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')  # 实际应使用中文微调模型
将查询和每个候选组成对，进行相关性预测
pair_inputs = [(query, doc) for doc in candidates]
scores = reranker.predict(pair_inputs)
将候选及得分配对，按分数排序
ranked_results = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
for doc, score in ranked_results:
    print(f"Score: {score:.4f} | Passage: {doc[:30]}...")

@假设有两个查询的测试集
ground_truth = {
    "Q1": [2, 5],    # 查询Q1的相关文档ID为2和5
    "Q2": [1]        # 查询Q2的相关文档ID为1
}
retrieved = {
    "Q1": [5, 2, 8, 3, 10],  # 系统返回的文档顺序，5号和2号是相关，其它无关
    "Q2": [3, 7, 1, 9, 4]    # 返回顺序中，第3个结果ID=1是相关
}
def calculate_mrr(gt, res):
    """计算平均倒数排名（MRR）"""
    total_reciprocal_rank = 0.0
    num_queries = len(gt)
    for q, relevant_docs in gt.items():
        rank = 0
        for idx, doc_id in




    
 enumerate(res.get(q, []), start=1):
            if doc_id in relevant_docs:
                rank = idx
                break
        if rank != 0:
            total_reciprocal_rank += 1.0 / rank
    return total_reciprocal_rank / num_queries
def calculate_precision_at_k(gt, res, K=3):
    """计算每个查询的Precision@K，并返回平均Precision@K"""
    precisions = []
    for q, relevant_docs in gt.items():
        retrieved_k = res.get(q, [])[:K]
        ifnot retrieved_k:
            continue
        # 计算前K结果中相关文档的比例
        rel_count = sum(1for doc_id in retrieved_k if doc_id in relevant_docs)
        precisions.append(rel_count / K)
        print(f"{q}的Precision@{K}: {rel_count}/{K} = {rel_count/K:.2f}")
    # 返回平均Precision@K
    return sum(precisions) / len(precisions) if precisions else0.0
mrr_value = calculate_mrr(ground_truth, retrieved)
precision3 = calculate_precision_at_k(ground_truth, retrieved, K=3)
print(f"MRR = {mrr_value:.3f}")
print(f"平均 Precision@3 = {precision3:.3f}")
计算NDCG@K
def calculate_ndcg_at_k(gt, res, K=3):
    """计算平均 NDCG@K（相关性二值相关0/1情况）"""
    total_ndcg = 0.0
    num_q = len(gt)
    for q, relevant_docs in gt.items():
        # 计算DCG@K
        dcg = 0.0
        for idx, doc_id in enumerate(res.get(q, [])[:K], start=1):
            rel = 1.0if doc_id in relevant_docs else0.0# 二级相关性：相关=1，否则=0
            if idx == 1:
                dcg += rel
            else:
                dcg += rel / math.log2(idx)  # 折损：log2(rank)
        # 计算IDCG@K（理想情况下的DCG）
        # 将相关文档都假设排在最前面




    

        ideal_rels = sorted([1.0]*len(relevant_docs) + [0.0]*K, reverse=True)[:K]
        idcg = 0.0
        for idx, rel in enumerate(ideal_rels, start=1):
            if idx == 1:
                idcg += rel
            else:
                idcg += rel / math.log2(idx)
        ndcg = dcg / idcg if idcg > 0else0.0
        total_ndcg += ndcg
        print(f"{q}的NDCG@{K}: {ndcg:.3f}")
    return total_ndcg / num_q
avg_ndcg3 = calculate_ndcg_at_k(ground_truth, retrieved, K=3)
print(f"平均 NDCG@3 = {avg_ndcg3:.3f}")

Q1的Precision@3: 2/3 = 0.67  
Q2的Precision@3: 1/3 = 0.33  
MRR = 0.583  
Q1的NDCG@3: 0.789  
Q2的NDCG@3: 0.500  
平均 NDCG@3 = 0.644

初始化向量数据库，已预先加载PDF文本片段、PPT文本片段、视频字幕片段等多模态知识
vector_store = VectorStore(data=load_multimodal_chunks())
conversation_history = []  # 用于保存多轮对话的问答历史
def is_follow_up(question):
    """简单判定问题是否为跟进问答（根据代词/省略等）"""
    follow_keywords = ["这个", "那种"




    
, "这样", "怎么", "如何", "吗?"]  # 简化判断逻辑
    return any(kw in question for kw in follow_keywords)
def answer_question(user_question):
    # 如果是跟进问题且有历史，则将上一次用户提问或主题融入当前问题
    if conversation_history and is_follow_up(user_question):
        last_topic = conversation_history[-1]["topic"]  # 上一轮对话主题
        # 将用户问题重写，加入主题关键字
        rewritten_q = f"{last_topic}，{user_question}"
    else:
        rewritten_q = user_question

    # 检索相关知识片段（返回内容和来源标识）
    docs = vector_store.search(rewritten_q, top_k=3)
    context = "".join([f"[{i}] {doc.content}\n"for i, doc in enumerate(docs, start=1)])

    # 构造提示，要求GPT-4根据检索片段回答，并标注引用编号
    prompt = (
        "请根据以下资料回答用户问题，并在对应内容后标注来源编号：\n"
        f"{context}\n"
        f"用户问题：{user_question}\n"
        "回答："
    )
    # 调用GPT-4模型生成答案
    answer = openai.ChatCompletion.create(model="gpt-4", messages=[{"role": "user", "content": prompt}])
    answer_text = answer["choices"][0]["message"]["content"]

    # 记录本轮主题（可选：从user_question或检索结果中抽取主题关键词）
    topic = extract_topic(user_question, docs)
    conversation_history.append({"question": user_question, "answer": answer_text, "topic": topic})

    return answer_text
模拟对话：
q1 = "ABC寿险的保障范围是什么？"
print(answer_question(q1))  # 系统基于知识库回答，答案包含来源标注，例如[1]
q2 = "这个怎么申请？"# 跟进问答，指代上文提到的“ABC寿险”
print(answer_question(q2))  # 系统利用对话历史，将ABC寿险作为上下文，检索申请流程并回答，包含来源标注