把文档助手的“关键词检索”换成语义检索

交付有什么不确定性？

上线时间存在延期风险。

问题里出现了哪些词 ↓这些词在文档片段里出现了几次 ↓命中越多，相关性越高

项目有没有延期可能？

当前排期存在交付风险。

风险在哪里？

上线时间存在延期风险。项目交付可能会推迟。

今天中午吃什么？

文档片段 ↓转成 embedding 向量 ↓存入向量库 ↓用户问题也转成向量 ↓在向量库里找最相似的片段 ↓把片段交给模型回答

切文档 → 找片段 → 基于片段回答 → 给出来源

08_vector_rag.py

pip install faiss-cpu numpy

pypdfpdfplumberpython-docxmarkitdownlangchainlangchain-openaifaiss-cpunumpy

from dataclasses import dataclass@dataclassclass DocumentChunk: content: str source: str index: int

from pathlib import Pathfrom docx import Documentfrom pypdf import PdfReaderSUPPORTED_SUFFIXES = {”.pdf”, ”.docx”, ”.txt”, ”.md”, ”.log”}def read_pdf(file_path: str) -> str: reader = PdfReader(file_path) pages = [] for i, page in enumerate(reader.pages): text = page.extract_text() or ”” text = text.strip() if text: pages.append(f”[第 {i + 1} 页]\n{text}”) return ”\n\n”.join(pages)def read_docx(file_path: str) -> str: doc = Document(file_path) parts = [] for p in doc.paragraphs: text = p.text.strip() if text: parts.append(text) for table_index, table in enumerate(doc.tables): parts.append(f”\n[表格 {table_index + 1}]”) for row in table.rows: cells = [cell.text.strip().replace(”\n”, ” ”) for cell in row.cells] if any(cells): parts.append(” | ”.join(cells)) return ”\n”.join(parts)def read_txt(file_path: str) -> str: return Path(file_path).read_text(encoding=”utf-8”)def read_document(file_path: str) -> str: path = Path(file_path) suffix = path.suffix.lower() if suffix == ”.pdf”: return read_pdf(str(path)) if suffix == ”.docx”: return read_docx(str(path)) if suffix in {”.txt”, ”.md”, ”.log”}: return read_txt(str(path)) return ””

def load_documents(folder: str) -> dict[str, str]: folder_path = Path(folder) docs = {} for file_path in folder_path.rglob(”*”): if not file_path.is_file(): continue if file_path.suffix.lower() not in SUPPORTED_SUFFIXES: continue text = read_document(str(file_path)) if text.strip(): docs[str(file_path)] = text return docs

def chunk_text(text: str, chunk_size: int = 800, overlap: int = 100) -> list[str]: text = text.strip() if not text: return [] chunks = [] start = 0 while start < len(text): end = start + chunk_size chunk = text[start:end].strip() if chunk: chunks.append(chunk) if end >= len(text): break start = end - overlap return chunksdef build_chunks(folder: str) -> list[DocumentChunk]: docs = load_documents(folder) all_chunks = [] for source, text in docs.items(): chunks = chunk_text(text) for i, chunk in enumerate(chunks, start=1): all_chunks.append( DocumentChunk( content=chunk, source=source, index=i, ) ) return all_chunks

import osfrom langchain_openai import OpenAIEmbeddingsdef build_embeddings() -> OpenAIEmbeddings: api_key = os.getenv(”OPENAI_API_KEY”) if not api_key: raise RuntimeError(”请先设置环境变量 OPENAI_API_KEY”) return OpenAIEmbeddings( model=”text-embedding-3-small”, api_key=api_key, )

import faissimport numpy as npdef build_vector_index(chunks: list[DocumentChunk], embeddings: OpenAIEmbeddings): texts = [chunk.content for chunk in chunks] vectors = embeddings.embed_documents(texts) matrix = np.array(vectors).astype(”float32”) dimension = matrix.shape[1] index = faiss.IndexFlatL2(dimension) index.add(matrix) return index

def retrieve_by_vector( question: str, chunks: list[DocumentChunk], index, embeddings: OpenAIEmbeddings, top_k: int = 4,) -> list[DocumentChunk]: query_vector = embeddings.embed_query(question) query_matrix = np.array([query_vector]).astype(”float32”) distances, indices = index.search(query_matrix, top_k) results = [] for idx in indices[0]: if idx == -1: continue results.append(chunks[idx]) return results

问题关键词 → 文档关键词匹配

问题向量 → 文档片段向量相似度

from langchain_openai import ChatOpenAIdef build_llm() -> ChatOpenAI: api_key = os.getenv(”DEEPSEEK_API_KEY”) if not api_key: raise RuntimeError(”请先设置环境变量 DEEPSEEK_API_KEY”) return ChatOpenAI( model=”deepseek-chat”, base_url=”https://api.deepseek.com/v1”, api_key=api_key, temperature=0, )

def answer_question(llm: ChatOpenAI, question: str, chunks: list[DocumentChunk]) -> str: if not chunks: return ”没有检索到相关内容，无法基于文档回答。” context_parts = [] for i, chunk in enumerate(chunks, start=1): context_parts.append( f”[资料 {i}]\n来源：{chunk.source}\n片段编号：{chunk.index}\n内容：\n{chunk.content}” ) context = ”\n\n”.join(context_parts) prompt = f”””你是一个文档问答助手。请只根据下面提供的资料回答问题。要求：- 不要编造资料中没有的信息- 如果资料不足，直接说“资料中未明确提到”- 回答要简洁- 最后列出你参考了哪些资料来源用户问题：{question}可参考资料：{context}请按下面格式输出：# 回答直接回答用户问题。# 依据列出引用的资料来源和片段编号。””” result = llm.invoke(prompt) return result.content

def vector_rag_answer(folder: str, question: str, top_k: int = 4) -> str: print(”正在读取并切分文档...”) chunks = build_chunks(folder) if not chunks: return ”没有读取到可用文档。请检查 test_docs 目录。” print(f”共生成 {len(chunks)} 个文档片段。”) print(”正在创建 embedding...”) embeddings = build_embeddings() print(”正在构建向量索引...”) index = build_vector_index(chunks, embeddings) print(”正在向量检索相关片段...”) related_chunks = retrieve_by_vector(question, chunks, index, embeddings, top_k=top_k) print(”检索到的片段：”) for chunk in related_chunks: print(f”- {chunk.source} / 片段 {chunk.index}”) print(”正在生成回答...”) llm = build_llm() return answer_question(llm, question, related_chunks)if __name__ == ”__main__”: folder = ”test_docs” question = ”这个项目目前有哪些不确定性？” answer = vector_rag_answer(folder, question) print(”\n” + ”=” * 80) print(answer)

python 08_vector_rag.py

AI 能不能先查资料再回答

AI 能不能更像人一样，按意思去找资料

https://gitee.com/ai-gravity-field/guide-to-ai-agent-exploration