增強型RAG系統的查詢轉換

采用三種查詢轉換技術，以提高RAG系統中的檢索性能，而無需依賴于像LangChain這樣的專門庫。通過修改用戶查詢，我們可以顯著提高檢索信息的相關性和全面性。

關鍵轉換技術

1.查詢重寫：使查詢更加具體和詳細，以提高搜索精度。
2.退步提示：生成更廣泛的查詢以檢索有用的上下文信息。
3.子查詢分解：將復雜的查詢分解成更簡單的組件進行全面檢索。

具體代碼實現

查詢變換相關函數

查詢重寫

def rewrite_query(original_query: str, model: str = LLM_MODEL) -> str:"""查詢重寫：讓查詢更具體詳細，提高檢索精度。"""print(f"[查詢重寫] 原始問題: {original_query}")system_prompt = "你是查詢優化專家，請將用戶問題改寫得更具體詳細，包含有助于檢索的相關術語和細節。"user_prompt = f"""
請將下列問題改寫得更具體詳細，補充有助于檢索的相關術語和細節：\n\n原始問題：{original_query}\n\n改寫后："""try:response = Generation.call(model=model,messages=[{"role": "system", "content": system_prompt},{"role": "user", "content": user_prompt}],api_key=ALI_API_KEY,result_format='message')if response.status_code == 200:rewritten = response.output.choices[0].message.content.strip()print(f"[查詢重寫] 改寫結果: {rewritten}\n")return rewrittenelse:print(f"[查詢重寫] 失敗: {response.message}")return original_queryexcept Exception as e:print(f"[查詢重寫] 異常: {e}")return original_query

生成更寬泛的背景性問題

def generate_step_back_query(original_query: str, model: str = LLM_MODEL) -> str:"""Step-back提示：生成更寬泛的背景性問題。"""print(f"[Step-back] 原始問題: {original_query}")system_prompt = "你是檢索策略專家，請將具體問題泛化為更寬泛的背景性問題，用于獲取相關背景信息。"user_prompt = f"""
請將下列問題泛化為更寬泛的背景性問題：\n\n原始問題：{original_query}\n\n泛化后："""try:response = Generation.call(model=model,messages=[{"role": "system", "content": system_prompt},{"role": "user", "content": user_prompt}],api_key=ALI_API_KEY,result_format='message')if response.status_code == 200:step_back = response.output.choices[0].message.content.strip()print(f"[Step-back] 泛化結果: {step_back}\n")return step_backelse:print(f"[Step-back] 失敗: {response.message}")return original_queryexcept Exception as e:print(f"[Step-back] 異常: {e}")return original_query

子查詢分解

def decompose_query(original_query: str, num_subqueries: int = 4, model: str = LLM_MODEL) -> List[str]:"""子查詢分解：將復雜問題拆解為若干簡單子問題。"""print(f"[子查詢分解] 原始問題: {original_query}")system_prompt = "你是復雜問題拆解專家，請將復雜問題拆解為若干簡單子問題，每個子問題關注不同方面。"user_prompt = f"""
請將下列復雜問題拆解為{num_subqueries}個子問題，每行一個：\n\n原始問題：{original_query}\n\n子問題："""try:response = Generation.call(model=model,messages=[{"role": "system", "content": system_prompt},{"role": "user", "content": user_prompt}],api_key=ALI_API_KEY,result_format='message')if response.status_code == 200:content = response.output.choices[0].message.content.strip()lines = content.split("\n")sub_queries = []for line in lines:if line.strip() and any(line.strip().startswith(f"{i}.") for i in range(1, 10)):q = line.strip()q = q[q.find(".")+1:].strip()sub_queries.append(q)elif line.strip():sub_queries.append(line.strip())print(f"[子查詢分解] 子問題: {sub_queries}\n")return sub_querieselse:print(f"[子查詢分解] 失敗: {response.message}")return [original_query]except Exception as e:print(f"[子查詢分解] 異常: {e}")return [original_query]

PDF文本處理與分塊

PDF文件中提取全部文本

def extract_text_from_pdf(pdf_path: str) -> str:"""從PDF文件中提取全部文本。"""print(f"[PDF提取] 正在提取: {pdf_path}")with open(pdf_path, 'rb') as f:reader = PdfReader(f)text = ""for i, page in enumerate(reader.pages):page_text = page.extract_text()if page_text:text += page_textprint(f"  - 已提取第{i+1}頁")print(f"[PDF提取] 完成，總長度: {len(text)} 字符\n")return text

文本分割為帶重疊的塊

def chunk_text(text: str, n: int = 1000, overlap: int = 200) -> List[str]:"""將文本分割為帶重疊的塊。"""print(f"[分塊] 每塊{n}字符，重疊{overlap}字符")chunks = []for i in range(0, len(text), n - overlap):chunks.append(text[i:i + n])print(f"[分塊] 完成，共{len(chunks)}塊\n")return chunks

向量生成與存儲

阿里embedding模型批量生成文本向量

def create_embeddings(texts, model: str = EMBEDDING_MODEL) -> List[np.ndarray]:"""用阿里embedding模型批量生成文本向量。支持單條或多條文本。"""if isinstance(texts, str):texts = [texts]print(f"[嵌入生成] 正在生成{len(texts)}條文本的向量...")try:response = TextEmbedding.call(model=model,input=texts,api_key=ALI_API_KEY)if response.status_code == 200:embeddings = [np.array(item['embedding']) for item in response.output['embeddings']]print(f"[嵌入生成] 成功，返回{len(embeddings)}條向量\n")return embeddings if len(embeddings) > 1 else embeddings[0]else:print(f"[嵌入生成] 失敗: {response.message}")return [np.zeros(1536)] * len(texts)except Exception as e:print(f"[嵌入生成] 異常: {e}")return [np.zeros(1536)] * len(texts)

簡單的向量存儲與檢索類

class SimpleVectorStore:"""簡單的向量存儲與檢索類。"""def __init__(self):self.vectors = []self.texts = []self.metadata = []def add_item(self, text, embedding, metadata=None):self.vectors.append(np.array(embedding))self.texts.append(text)self.metadata.append(metadata or {})def similarity_search(self, query_embedding, k=5):if not self.vectors:return []query_vector = np.array(query_embedding)similarities = []for i, vector in enumerate(self.vectors):sim = np.dot(query_vector, vector) / (np.linalg.norm(query_vector) * np.linalg.norm(vector))similarities.append((i, sim))similarities.sort(key=lambda x: x[1], reverse=True)results = []for i in range(min(k, len(similarities))):idx, score = similarities[i]results.append({"text": self.texts[idx],"metadata": self.metadata[idx],"similarity": score})return results

主流程與RAG實現

處理PDF文檔，提取文本、分塊、生成向量并構建向量庫

def process_document(pdf_path: str, chunk_size: int = 1000, chunk_overlap: int = 200) -> SimpleVectorStore:"""處理PDF文檔，提取文本、分塊、生成向量并構建向量庫。"""print("[主流程] 開始處理文檔...")extracted_text = extract_text_from_pdf(pdf_path)text_chunks = chunk_text(extracted_text, chunk_size, chunk_overlap)print("[主流程] 初始化向量庫...")vector_store = SimpleVectorStore()print("[主流程] 為每個塊生成向量...")for i, chunk in enumerate(text_chunks):print(f"[塊{i+1}/{len(text_chunks)}] 正在處理文本塊，長度: {len(chunk)} 字符")chunk_emb = create_embeddings(chunk)vector_store.add_item(chunk, chunk_emb, {"type": "chunk", "index": i})print("[主流程] 文檔處理完畢，向量庫構建完成\n")return vector_store

用變換后的查詢進行檢索

def transformed_search(query: str, vector_store: SimpleVectorStore, transformation_type: str, top_k: int = 3) -> List[Dict]:"""用變換后的查詢進行檢索。"""print(f"[檢索] 查詢變換類型: {transformation_type}")print(f"[檢索] 原始問題: {query}")results = []if transformation_type == "rewrite":transformed_query = rewrite_query(query)query_embedding = create_embeddings(transformed_query)results = vector_store.similarity_search(query_embedding, k=top_k)elif transformation_type == "step_back":transformed_query = generate_step_back_query(query)query_embedding = create_embeddings(transformed_query)results = vector_store.similarity_search(query_embedding, k=top_k)elif transformation_type == "decompose":sub_queries = decompose_query(query)print("[檢索] 子問題:")for i, sub_q in enumerate(sub_queries, 1):print(f"  {i}. {sub_q}")sub_query_embeddings = create_embeddings(sub_queries)if isinstance(sub_query_embeddings, np.ndarray):sub_query_embeddings = [sub_query_embeddings]all_results = []for i, embedding in enumerate(sub_query_embeddings):sub_results = vector_store.similarity_search(embedding, k=2)all_results.extend(sub_results)seen_texts = {}for result in all_results:text = result["text"]if text not in seen_texts or result["similarity"] > seen_texts[text]["similarity"]:seen_texts[text] = resultresults = sorted(seen_texts.values(), key=lambda x: x["similarity"], reverse=True)[:top_k]else:query_embedding = create_embeddings(query)results = vector_store.similarity_search(query_embedding, k=top_k)print(f"[檢索] 檢索結果數: {len(results)}\n")return results

用大模型基于上下文生成回答

def generate_response(query: str, context: str, model: str = LLM_MODEL) -> str:"""用大模型基于上下文生成回答。"""print("[生成] 正在調用大模型生成回答...")system_prompt = "你是一個AI助手，只能基于給定上下文回答問題。如果上下文無法直接回答，請回復：'信息不足，無法回答。'"user_prompt = f"""
上下文：\n{context}\n\n問題：{query}\n\n請只基于上述上下文簡明準確作答。"""try:response = Generation.call(model=model,messages=[{"role": "system", "content": system_prompt},{"role": "user", "content": user_prompt}],api_key=ALI_API_KEY,result_format='message')if response.status_code == 200:answer = response.output.choices[0].message.content.strip()print(f"[生成] 回答生成成功: {answer}\n")return answerelse:print(f"[生成] 回答生成失敗: {response.message}")return ""except Exception as e:print(f"[生成] 回答生成異常: {e}")return ""

完整RAG流程

def rag_with_query_transformation(pdf_path: str, query: str, transformation_type: str = None):"""完整RAG流程，支持可選的查詢變換。"""vector_store = process_document(pdf_path)if transformation_type:results = transformed_search(query, vector_store, transformation_type)else:query_embedding = create_embeddings(query)results = vector_store.similarity_search(query_embedding, k=3)context = "\n\n".join([f"PASSAGE {i+1}:\n{result['text']}" for i, result in enumerate(results)])response = generate_response(query, context)return {"original_query": query,"transformation_type": transformation_type,"context": context,"response": response}

附錄

執行結果展示

===== RAG 查詢變換增強示例 =====[主流程] 開始處理文檔...
[PDF提取] 正在提取: data/2888年Java程序員找工作最新場景題.pdf- 已提取第1頁- 已提取第2頁- 已提取第3頁- 已提取第4頁- 已提取第5頁- 已提取第6頁- 已提取第7頁- 已提取第8頁- 已提取第9頁- 已提取第10頁
[PDF提取] 完成，總長度: 6984 字符[分塊] 每塊1000字符，重疊200字符
[分塊] 完成，共9塊[主流程] 初始化向量庫...
[主流程] 為每個塊生成向量...
[塊1/9] 正在處理文本塊，長度: 1000 字符
[嵌入生成] 正在生成1條文本的向量...
[嵌入生成] 成功，返回1條向量[塊2/9] 正在處理文本塊，長度: 1000 字符
[嵌入生成] 正在生成1條文本的向量...
[嵌入生成] 成功，返回1條向量[塊3/9] 正在處理文本塊，長度: 1000 字符
[嵌入生成] 正在生成1條文本的向量...
[嵌入生成] 成功，返回1條向量[塊4/9] 正在處理文本塊，長度: 1000 字符
[嵌入生成] 正在生成1條文本的向量...
[嵌入生成] 成功，返回1條向量[塊5/9] 正在處理文本塊，長度: 1000 字符
[嵌入生成] 正在生成1條文本的向量...
[嵌入生成] 成功，返回1條向量[塊6/9] 正在處理文本塊，長度: 1000 字符
[嵌入生成] 正在生成1條文本的向量...
[嵌入生成] 成功，返回1條向量[塊7/9] 正在處理文本塊，長度: 1000 字符
[嵌入生成] 正在生成1條文本的向量...
[嵌入生成] 成功，返回1條向量[塊8/9] 正在處理文本塊，長度: 1000 字符
[嵌入生成] 正在生成1條文本的向量...
[嵌入生成] 成功，返回1條向量[塊9/9] 正在處理文本塊，長度: 584 字符
[嵌入生成] 正在生成1條文本的向量...
[嵌入生成] 成功，返回1條向量[主流程] 文檔處理完畢，向量庫構建完成[檢索] 查詢變換類型: rewrite
[檢索] 原始問題: 人工智能對未來就業的影響有哪些？
[查詢重寫] 原始問題: 人工智能對未來就業的影響有哪些？
[查詢重寫] 改寫結果: 改寫后的問題：人工智能技術（包括但不限于機器學習、自然語言處理、計算機視覺等）在未來十年內對全球不同行業（如制造業、服務業、信息技術業等）的就業市場會產生哪些具體影響？這些影響包括但不限于工作崗位的變化趨勢（新增崗位與消失崗位）、所需技能的轉變以及對勞動力結構（例如年齡、教育背景）的影響。此外，還請考慮這種變化對于政策制定者、企業雇主及員工個人分別意味著什么挑戰和機遇。[嵌入生成] 正在生成1條文本的向量...
[嵌入生成] 成功，返回1條向量[檢索] 檢索結果數: 3[生成] 正在調用大模型生成回答...
[生成] 回答生成成功: 信息不足，無法回答。===== 最終結果 =====
原始問題: 人工智能對未來就業的影響有哪些？
變換類型: rewrite
檢索上下文: 
PASSAGE 1:
涉及的行業如金融、電子商務，及特定業務如客戶關系管理、支付
系統等。
? 軟實力與經驗：團隊管理規模、項目管理經驗、個人特質等，簡潔明了即可。
確保簡歷內容與目標職位的技能要求對齊，以便快速建立匹配印象。工作與教育背景需
精挑細選，強調關鍵項目、挑戰、責任及所獲成就，同時，這些內容應緊密支撐你的技
能陳述，避免離題。
簡歷長度以不超過兩頁A4紙為宜，采用PDF格式以保證格式一致性。
可借鑒LinkedIn或MicrosoftOffice模板美化外觀，力求內容精煉，重點突出。
記住，簡歷的目標是凸顯你的獨特之處，哪怕僅展示兩三個亮點，也足以引起注意。
最后，簡歷是打開機遇之門的第一步，尤其在競爭激烈的就業市場中，除了實質性的技
能和經驗，別出心裁的簡歷設計與正面積極的自我評價亦能增加脫穎而出的機會。
即便初始條件有限，展現出積極的態度、持續的學習意愿和解決問題的能力，同樣能傳
達出你是一個值得投資的潛力股。
1.2技術知識儲備
在準備簡歷時，切記實事求是，你所列出的每一項技能都可能成為面試對話的起點。"
精通"、"熟悉"、"了解"需準確區分，以免自相矛盾。對于提及的每項技術，務必把握其基礎及核心概念，因為面試過程中，面試官往往會逐
步深入探討，以此評估你的實際水平。因此，系統性復習相關書籍和資料是不可或缺的
步驟，以備不時之需。
例如：
? 如你標明熟練掌握Java，那么不僅限于基礎語法，還應涵蓋并發編程、NIO、JVM
等進階知識，同時對Spring、Netty等流行框架的基本原理有所認識。
? 提及Go語言，意味著你應至少閱讀過官方的《EffectiveGo》，理解其核心理念。
? 當列舉Redis時，對其數據結構、性能調優策略、高可用部署方式及分布式鎖機制
等，通過官方文檔的研讀應達到一定的理解深度。
? 如聲稱掌握面向對象設計，熟悉《設計模式》中的經典23種模式將是基本要求。
? 對于分布式架構的宣稱，則需對CAP原則、微服務架構、彈性設計以及Spring
Cloud、CloudNative等相關技術框架有深刻理解。
? 關于網絡編程的技能，理解TCP/IP協議的三次握手、四次揮手過程，Socket編程
基礎，以及select、poll、epoll等I/O多路復用技術，都是必不可少的知識點。
綜上所述，你簡歷上的每一項技術標注，都應當基于你對該技術核心知識點的PASSAGE 2:
本要求。
? 對于分布式架構的宣稱，則需對CAP原則、微服務架構、彈性設計以及Spring
Cloud、CloudNative等相關技術框架有深刻理解。
? 關于網絡編程的技能，理解TCP/IP協議的三次握手、四次揮手過程，Socket編程
基礎，以及select、poll、epoll等I/O多路復用技術，都是必不可少的知識點。
綜上所述，你簡歷上的每一項技術標注，都應當基于你對該技術核心知識點的掌握之上。
這好比備考期末考試，你需要全面回顧教材，確保掌握大多數關鍵知識點，即使不必面
面俱到，但對于80%以上的重點內容，你都應做到心中有數。這樣的準備不僅是為了
應對面試，更是對自己技術深度和廣度的負責態度體現。
1.3項目準備（非常重要）
在面試過程中，分享個人項目經歷或解決過的挑戰幾乎是每個面試官必問的環節，但令
人詫異的是，許多候選人并未對此做好充分準備。以下四個經典問題頻繁出現于面試之
中：
1.分享一個你最為自豪或最近完成的項目。
2.講述一次你攻克的最復雜或技術含量最高的難題。
3.描述一個你經歷過的最具挑戰性或最艱難的項目。
4.談談你曾犯下的最大技術失誤或引發的技術故障。
這些問題背后，面試官的意圖各異：
? 第一個問題旨在探查你的成就頂峰、興趣所在；? 第二、三題側重于你的問題解決能力和面對逆境時的心態韌性；
? 而第四題則關注你對待錯誤的態度，以及是否具備反思與成長的能力。
值得注意的是，面試官會通過連續追問細節來驗證信息的真實性，因為虛構的情節難以
在嚴密的追問下自圓其說。
為有效應對這類問題，以下建議或許能幫助你更好地準備：
? 構建故事框架：運用STAR法則（情境Situation、任務Task、行動Action、結果
Result）來組織你的敘述，確保內容條理清晰，避免冗長繁雜。
? 添加細節：豐富的技術細節是說服力的關鍵，它能讓故事顯得更加真實可信。
? 注入情感：真摯的情感表達能傳遞你的熱情、自豪與堅持，確保情感源自真實的體
驗。
? 融入反思：在敘述中穿插你的思考、教訓總結及后續的改進措施，展現你的成長和
成熟。
達到這樣的敘述水平并非易事，需要持續的練習與積累。日常工作中，培養即時總結的
習慣，對經歷進行記錄與反思，是避免臨陣磨槍的有效方法。
此外，提升語言組織能力與邏輯思維同樣重要。通過撰寫工作文檔和經營個人技術博客，
不僅可以鍛PASSAGE 3:
2024年Java程序員找工作最新面試攻略
這個文檔是幫助正在找工作以及準備找工作的同學，在面試之前去復習和突擊的一種方
式。
適合已經在技術領域有一定積累，然后不確定面試切入點，所以可以通過這個面試文檔
來預熱和鞏固。
想直接通過刷面試文檔找到工作的同學也要注意，面試文檔的內容是靜態的，但是面試
過程是動態的，
面試官對于某一個領域的考察，通常是通過連環問的方式去問，所以在面試之前，求職
者要對Java相
關技術有一個體系化的了解，從而更好地突出自己的綜合能力。
在科技日新月異的今天，軟件開發行業正經歷著前所未有的變革。Java，作為企業級應
用開發的中流砥柱，其生態系統也在不斷進化，從微服務架構的普及到云原生技術的興
起，再到AI與大數據的深度融合，Java程序員的角色和技能需求隨之迭代升級。面對
這樣的行業背景，如何在求職路上脫穎而出，成為每位開發者必須深思的問題。
隨著Java這個賽道的不斷內卷，這兩年，Java程序員的面試，從原來的常規八股文（有
標準答案）到現在，以項目、場景問題、技術深度思考為主，逐步轉變成沒有標準答案，
需要大家基于自己的理解和技術底蘊來回答。
那針對市場中新的需求，有沒有最新的面試攻略呢？其實也是有的，雖然說沒有標準
答案，但是我們可以針對如今市場的面試變化，來針對性的設計一些面試回答的思路，
讓大家有一個清晰和明確的方向。
這里有什么？
1.針對2024年面試行情的變化設計的面試場景題以及回答思路2.如何快速通過面試的詳細攻略
3.簡歷優化技巧
1.知己知彼才能百戰百勝，如何做好面試前的準備
工作
2024年的行情，和3~4年前不同，通過海量簡歷投遞和海量面試找工作的時代已
經過去了。
在如今面試機會較少，并且面試難度較大的情況下。
充分做好面試的準備才是快速通過面試最有效的方法！
切忌把真實面試當靶場，最后帶來的代價是非常巨大的！
面試無非就兩個部分，投簡歷、面試！
很多人把重心放在投簡歷上，忽略了準備面試的重要性，最后的結果是獲得了面試機會
但是在面試過程中被刷下來了。
1.1怎么寫簡歷
著手準備的第一步聚焦于簡歷的打造。
簡歷是他人初步了解你的窗口，其重要性不言而喻，因此精心構思簡歷至關重要。
理想的簡歷應當圍繞你的親身經歷構建，正如某些杰出人士僅憑一句“Unix的創造
者”便足以令人印象深刻。
盡管并非所有人都擁有如此耀眼
大模型回答: 
信息不足，無法回答。進程已結束，退出代碼為 0

完整示例代碼

# -*- coding: utf-8 -*-
"""
RAG 查詢變換增強腳本（基于阿里大模型Qwen）
========================================
本腳本實現了三種查詢變換技術：查詢重寫、Step-back提示、子查詢分解。
所有大模型調用均基于阿里云通義千問Qwen，密鑰從api_keys.py讀取。
每一步均有詳細控制臺輸出，便于觀察效果。
"""
import os
import numpy as np
import json
import sys
from typing import List, Dict
from PyPDF2 import PdfReader
from dashscope import Generation, TextEmbedding# ========== 密鑰配置 ==========
# try:
#     from test.api_keys import ALI_API_KEY
# except ImportError:
#     raise RuntimeError("未找到test/api_keys.py或未定義ALI_API_KEY，請配置API密鑰！")
ALI_API_KEY="sk-148deabc0bcf4fdeaa70a78eaa829c7e"
# =============================LLM_MODEL = "qwen-max"  # 通義千問主力模型
EMBEDDING_MODEL = "text-embedding-v2"  # 阿里云嵌入模型# ========== 查詢變換相關函數 ==========
def rewrite_query(original_query: str, model: str = LLM_MODEL) -> str:"""查詢重寫：讓查詢更具體詳細，提高檢索精度。"""print(f"[查詢重寫] 原始問題: {original_query}")system_prompt = "你是查詢優化專家，請將用戶問題改寫得更具體詳細，包含有助于檢索的相關術語和細節。"user_prompt = f"""
請將下列問題改寫得更具體詳細，補充有助于檢索的相關術語和細節：\n\n原始問題：{original_query}\n\n改寫后："""try:response = Generation.call(model=model,messages=[{"role": "system", "content": system_prompt},{"role": "user", "content": user_prompt}],api_key=ALI_API_KEY,result_format='message')if response.status_code == 200:rewritten = response.output.choices[0].message.content.strip()print(f"[查詢重寫] 改寫結果: {rewritten}\n")return rewrittenelse:print(f"[查詢重寫] 失敗: {response.message}")return original_queryexcept Exception as e:print(f"[查詢重寫] 異常: {e}")return original_querydef generate_step_back_query(original_query: str, model: str = LLM_MODEL) -> str:"""Step-back提示：生成更寬泛的背景性問題。"""print(f"[Step-back] 原始問題: {original_query}")system_prompt = "你是檢索策略專家，請將具體問題泛化為更寬泛的背景性問題，用于獲取相關背景信息。"user_prompt = f"""
請將下列問題泛化為更寬泛的背景性問題：\n\n原始問題：{original_query}\n\n泛化后："""try:response = Generation.call(model=model,messages=[{"role": "system", "content": system_prompt},{"role": "user", "content": user_prompt}],api_key=ALI_API_KEY,result_format='message')if response.status_code == 200:step_back = response.output.choices[0].message.content.strip()print(f"[Step-back] 泛化結果: {step_back}\n")return step_backelse:print(f"[Step-back] 失敗: {response.message}")return original_queryexcept Exception as e:print(f"[Step-back] 異常: {e}")return original_querydef decompose_query(original_query: str, num_subqueries: int = 4, model: str = LLM_MODEL) -> List[str]:"""子查詢分解：將復雜問題拆解為若干簡單子問題。"""print(f"[子查詢分解] 原始問題: {original_query}")system_prompt = "你是復雜問題拆解專家，請將復雜問題拆解為若干簡單子問題，每個子問題關注不同方面。"user_prompt = f"""
請將下列復雜問題拆解為{num_subqueries}個子問題，每行一個：\n\n原始問題：{original_query}\n\n子問題："""try:response = Generation.call(model=model,messages=[{"role": "system", "content": system_prompt},{"role": "user", "content": user_prompt}],api_key=ALI_API_KEY,result_format='message')if response.status_code == 200:content = response.output.choices[0].message.content.strip()lines = content.split("\n")sub_queries = []for line in lines:if line.strip() and any(line.strip().startswith(f"{i}.") for i in range(1, 10)):q = line.strip()q = q[q.find(".")+1:].strip()sub_queries.append(q)elif line.strip():sub_queries.append(line.strip())print(f"[子查詢分解] 子問題: {sub_queries}\n")return sub_querieselse:print(f"[子查詢分解] 失敗: {response.message}")return [original_query]except Exception as e:print(f"[子查詢分解] 異常: {e}")return [original_query]# ========== PDF文本處理與分塊 ==========
def extract_text_from_pdf(pdf_path: str) -> str:"""從PDF文件中提取全部文本。"""print(f"[PDF提取] 正在提取: {pdf_path}")with open(pdf_path, 'rb') as f:reader = PdfReader(f)text = ""for i, page in enumerate(reader.pages):page_text = page.extract_text()if page_text:text += page_textprint(f"  - 已提取第{i+1}頁")print(f"[PDF提取] 完成，總長度: {len(text)} 字符\n")return textdef chunk_text(text: str, n: int = 1000, overlap: int = 200) -> List[str]:"""將文本分割為帶重疊的塊。"""print(f"[分塊] 每塊{n}字符，重疊{overlap}字符")chunks = []for i in range(0, len(text), n - overlap):chunks.append(text[i:i + n])print(f"[分塊] 完成，共{len(chunks)}塊\n")return chunks# ========== 向量生成與存儲 ==========
def create_embeddings(texts, model: str = EMBEDDING_MODEL) -> List[np.ndarray]:"""用阿里embedding模型批量生成文本向量。支持單條或多條文本。"""if isinstance(texts, str):texts = [texts]print(f"[嵌入生成] 正在生成{len(texts)}條文本的向量...")try:response = TextEmbedding.call(model=model,input=texts,api_key=ALI_API_KEY)if response.status_code == 200:embeddings = [np.array(item['embedding']) for item in response.output['embeddings']]print(f"[嵌入生成] 成功，返回{len(embeddings)}條向量\n")return embeddings if len(embeddings) > 1 else embeddings[0]else:print(f"[嵌入生成] 失敗: {response.message}")return [np.zeros(1536)] * len(texts)except Exception as e:print(f"[嵌入生成] 異常: {e}")return [np.zeros(1536)] * len(texts)class SimpleVectorStore:"""簡單的向量存儲與檢索類。"""def __init__(self):self.vectors = []self.texts = []self.metadata = []def add_item(self, text, embedding, metadata=None):self.vectors.append(np.array(embedding))self.texts.append(text)self.metadata.append(metadata or {})def similarity_search(self, query_embedding, k=5):if not self.vectors:return []query_vector = np.array(query_embedding)similarities = []for i, vector in enumerate(self.vectors):sim = np.dot(query_vector, vector) / (np.linalg.norm(query_vector) * np.linalg.norm(vector))similarities.append((i, sim))similarities.sort(key=lambda x: x[1], reverse=True)results = []for i in range(min(k, len(similarities))):idx, score = similarities[i]results.append({"text": self.texts[idx],"metadata": self.metadata[idx],"similarity": score})return results# ========== 主流程與RAG實現 ==========
def process_document(pdf_path: str, chunk_size: int = 1000, chunk_overlap: int = 200) -> SimpleVectorStore:"""處理PDF文檔，提取文本、分塊、生成向量并構建向量庫。"""print("[主流程] 開始處理文檔...")extracted_text = extract_text_from_pdf(pdf_path)text_chunks = chunk_text(extracted_text, chunk_size, chunk_overlap)print("[主流程] 初始化向量庫...")vector_store = SimpleVectorStore()print("[主流程] 為每個塊生成向量...")for i, chunk in enumerate(text_chunks):print(f"[塊{i+1}/{len(text_chunks)}] 正在處理文本塊，長度: {len(chunk)} 字符")chunk_emb = create_embeddings(chunk)vector_store.add_item(chunk, chunk_emb, {"type": "chunk", "index": i})print("[主流程] 文檔處理完畢，向量庫構建完成\n")return vector_storedef transformed_search(query: str, vector_store: SimpleVectorStore, transformation_type: str, top_k: int = 3) -> List[Dict]:"""用變換后的查詢進行檢索。"""print(f"[檢索] 查詢變換類型: {transformation_type}")print(f"[檢索] 原始問題: {query}")results = []if transformation_type == "rewrite":transformed_query = rewrite_query(query)query_embedding = create_embeddings(transformed_query)results = vector_store.similarity_search(query_embedding, k=top_k)elif transformation_type == "step_back":transformed_query = generate_step_back_query(query)query_embedding = create_embeddings(transformed_query)results = vector_store.similarity_search(query_embedding, k=top_k)elif transformation_type == "decompose":sub_queries = decompose_query(query)print("[檢索] 子問題:")for i, sub_q in enumerate(sub_queries, 1):print(f"  {i}. {sub_q}")sub_query_embeddings = create_embeddings(sub_queries)if isinstance(sub_query_embeddings, np.ndarray):sub_query_embeddings = [sub_query_embeddings]all_results = []for i, embedding in enumerate(sub_query_embeddings):sub_results = vector_store.similarity_search(embedding, k=2)all_results.extend(sub_results)seen_texts = {}for result in all_results:text = result["text"]if text not in seen_texts or result["similarity"] > seen_texts[text]["similarity"]:seen_texts[text] = resultresults = sorted(seen_texts.values(), key=lambda x: x["similarity"], reverse=True)[:top_k]else:query_embedding = create_embeddings(query)results = vector_store.similarity_search(query_embedding, k=top_k)print(f"[檢索] 檢索結果數: {len(results)}\n")return resultsdef generate_response(query: str, context: str, model: str = LLM_MODEL) -> str:"""用大模型基于上下文生成回答。"""print("[生成] 正在調用大模型生成回答...")system_prompt = "你是一個AI助手，只能基于給定上下文回答問題。如果上下文無法直接回答，請回復：'信息不足，無法回答。'"user_prompt = f"""
上下文：\n{context}\n\n問題：{query}\n\n請只基于上述上下文簡明準確作答。"""try:response = Generation.call(model=model,messages=[{"role": "system", "content": system_prompt},{"role": "user", "content": user_prompt}],api_key=ALI_API_KEY,result_format='message')if response.status_code == 200:answer = response.output.choices[0].message.content.strip()print(f"[生成] 回答生成成功: {answer}\n")return answerelse:print(f"[生成] 回答生成失敗: {response.message}")return ""except Exception as e:print(f"[生成] 回答生成異常: {e}")return ""def rag_with_query_transformation(pdf_path: str, query: str, transformation_type: str = None):"""完整RAG流程，支持可選的查詢變換。"""vector_store = process_document(pdf_path)if transformation_type:results = transformed_search(query, vector_store, transformation_type)else:query_embedding = create_embeddings(query)results = vector_store.similarity_search(query_embedding, k=3)context = "\n\n".join([f"PASSAGE {i+1}:\n{result['text']}" for i, result in enumerate(results)])response = generate_response(query, context)return {"original_query": query,"transformation_type": transformation_type,"context": context,"response": response}# ========== main方法示例 ==========
if __name__ == "__main__":# 示例PDF路徑（請替換為實際文件）pdf_path = "data/2888年Java程序員找工作最新場景題.pdf"# 示例問題query = "人工智能對未來就業的影響有哪些？"# 選擇變換類型：None, "rewrite", "step_back", "decompose"transformation_type = "rewrite"  # 可改為step_back、decompose或Noneprint("\n===== RAG 查詢變換增強示例 =====\n")result = rag_with_query_transformation(pdf_path, query, transformation_type)print("\n===== 最終結果 =====")print(f"原始問題: {result['original_query']}")print(f"變換類型: {result['transformation_type']}")print(f"檢索上下文: \n{result['context']}")print(f"大模型回答: \n{result['response']}")