异步编程等待的时候不干等可以去干别的事情异步 ≠ 多线程 ≠ 多进程异步解决IO等待任务在等待的时候把执行权让出去先处理其他任务A开始↓A等待API ─────────────┐↓B开始 │↓ │B等待API ───────┐ │↓ ↓C开始 │↓ │C等待API ───────┘↓A回来↓B回来↓C回来多线程创建多个线程执行任务增加线程会占用资源或内存Thread 1↓Task AThread 2↓Task BThread 3↓Task C多进程每个进程有自己的内存、Python解释器、资源餐厅1 → Process 1餐厅2 → Process 2餐厅3 → Process 3实例任务A请求API需要等待10秒任务B请求API需要等待10秒同步总共 ≈ 20sA ──────────10s──────────B ──────────10s──────────异步总共 ≈ 10sA ──等待───────────────B ──等待────────────了多线程总共 ≈ 10sThread 1A ──────────等待10s────────Thread 2B ──────────等待10s────────Pydantic类型注解用于校验输入数据是否符合预期类型。from pydantic import BaseModel class Product(BaseModel): sku: str price: float quantity: int product Product( skuABC123, price19.9, quantity10 )类型注解定义的是参数的类型约束两者的区别类型注解如def predict(price: float)用于告知 IDE 参数应为 float 类型。Pydantic 则在运行时真正校验输入数据是否符合类型要求。LangChain LlamaIndex——底层机制解释Runnable任何这种输入→输出的组件一个可以执行的节点每一个组件都是runnableRunnable├── Prompt├── Retriever├── LLM输入prompt,输出回答├── Parser└── Rerankerchain:将多个核心步骤串起来chain prompt | llm | parserRunnable1↓Runnable2↓Runnable3LlamaIndex专注于数据增强和文档索引将私有数据文本、表格、PDF 等组织成 LLM 可直接使用的上下文适合快速搭建文档问答系统或 RAG 系统尤其是文档结构复杂时。常规RAGLlamaIndexPDF 读取fitz / PyMuPDFDocument文本清洗is_noise_page、clean_text文本清洗is_noise_page、clean_text切块固定长度切分 /CharacterTextSplitterSentenceSplitter/TokenTextSplitterEmbeddingSentenceTransformer(all-mpnet-base-v2)IngestionPipeline向量库FAISSFaissVectorStore召回index.search(...)retriever / query engine重排调bge-reranker-v2-m3Reranker生成Claude 或本地 HuggingFace 模型QueryEngineRAG使用LLamaIndex实现#文档加载import os import fitz # PyMuPDF from loguru import logger from tqdm import tqdm PDF_PATH /opt/cyc/rag/simple-local-rag/human-nutrition-text.pdf LOG_DIR log os.makedirs(LOG_DIR, exist_okTrue) logger.remove() logger.add( f{LOG_DIR}/pdf_load_llamaindex.log, rotation200 MB, retention1, levelINFO, format{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}, encodingutf-8, )#数据清洗import re NOISE_KEYWORDS [ unsplash license, creative commons, attribution, photo by, image by, reused, license, copyright, all rights reserved, ] def is_noise_page(text): lower_text text.lower() hit_count sum(keyword in lower_text for keyword in NOISE_KEYWORDS) if hit_count 3: return True url_count len(re.findall(rhttps?://|www\., lower_text)) if url_count 5: return True return False def clean_text(text): text text.replace(\xa0, ) text re.sub(r[ \t], , text) text re.sub(r(?!\n)\n(?!\n), , text) text re.sub(r\n{3,}, \n\n, text) return text.strip() def load_pdf(pdf_path): pages [] with fitz.open(pdf_path) as doc: logger.info(f开始加载PDF: {pdf_path}, 总页数: {doc.page_count}) for page_index, page in tqdm(enumerate(doc), totaldoc.page_count): text clean_text(page.get_text()) if not text: logger.info(f第{page_index}页为空跳过) continue if is_noise_page(text): logger.info(f第{page_index}页为噪声页跳过) continue pages.append({ page: page_index, text: text, }) logger.info(fPDF加载完成有效页数: {len(pages)}) return pages pages load_pdf(PDF_PATH) print(f有效页数: {len(pages)})#文档读取from llama_index.core import Document documents [ Document(textpage[text], metadata{page: page[page]}) for page in pages ]#文本切分from llama_index.core.node_parser import SentenceSplitter splitter SentenceSplitter(chunk_size300, chunk_overlap20) nodes splitter.get_nodes_from_documents(documents) print(fDocument 数量: {len(documents)}) print(fNode 数量: {len(nodes)})#向量化和存储import faiss from llama_index.core import Settings, StorageContext, VectorStoreIndex from llama_index.embeddings.huggingface import HuggingFaceEmbedding from llama_index.vector_stores.faiss import FaissVectorStore embed_model HuggingFaceEmbedding( model_namesentence-transformers/all-mpnet-base-v2, cache_foldermodel, ) Settings.embed_model embed_model embed_dim len(embed_model.get_text_embedding(蛋白质是什么)) faiss_index faiss.IndexFlatIP(embed_dim) vector_store FaissVectorStore(faiss_indexfaiss_index) storage_context StorageContext.from_defaults(vector_storevector_store) index VectorStoreIndex( nodes, storage_contextstorage_context, embed_modelembed_model, ) INDEX_PATH rag/simple-local-rag/docs_llamaindex.index faiss.write_index(faiss_index, INDEX_PATH) print(f已保存索引到: {INDEX_PATH})#检索召回query 蛋白质的主要功能是什么 retriever index.as_retriever(similarity_top_k3) retrieved_nodes retriever.retrieve(query) def preview_nodes(nodes): for idx, node in enumerate(nodes, start1): score getattr(node, score, None) score_text f{score:.4f} if score is not None else N/A page node.metadata.get(page) text node.text.replace(\n, ) print(f[{idx}] score{score_text} | page{page}) print(text[:400]) print(- * 80) preview_nodes(retrieved_nodes)#精排可选#生成from anthropic import Anthropic def build_context_from_nodes(nodes): blocks [] for idx, node in enumerate(nodes, start1): page node.metadata.get(page, unknown) score getattr(node, score, None) score_text f{score:.4f} if score is not None else N/A blocks.append( f[Chunk {idx}] page{page}, score{score_text}\n{node.text} ) return \n\n.join(blocks) def answer_with_rag(query, retriever, top_k3, use_rerankerFalse): nodes retriever.retrieve(query)[:top_k] if use_reranker: nodes rerank_nodes_with_bge(query, nodes) context build_context_from_nodes(nodes) client Anthropic() prompt f你是一个基于资料回答问题的助手。 请只根据“参考资料”回答不要编造。 如果参考资料不足以回答就明确说“参考资料不足”。 用户问题 {query} 参考资料 {context} 请输出 1. 简洁答案 2. 如果能定位来源顺带标注 page resp client.messages.create( modelclaude-sonnet-5, max_tokens500, messages[{role: user, content: prompt}], ) return { answer: resp.content[0].text, context: context, nodes: nodes, } rag_result answer_with_rag( 蛋白质的主要功能是什么, retriever, use_rerankerFalse, ) print(rag_result[answer]) Python基础_体系
异步编程等待的时候不干等可以去干别的事情异步 ≠ 多线程 ≠ 多进程异步解决IO等待任务在等待的时候把执行权让出去先处理其他任务A开始↓A等待API ─────────────┐↓B开始 │↓ │B等待API ───────┐ │↓ ↓C开始 │↓ │C等待API ───────┘↓A回来↓B回来↓C回来多线程创建多个线程执行任务增加线程会占用资源或内存Thread 1↓Task AThread 2↓Task BThread 3↓Task C多进程每个进程有自己的内存、Python解释器、资源餐厅1 → Process 1餐厅2 → Process 2餐厅3 → Process 3实例任务A请求API需要等待10秒任务B请求API需要等待10秒同步总共 ≈ 20sA ──────────10s──────────B ──────────10s──────────异步总共 ≈ 10sA ──等待───────────────B ──等待────────────了多线程总共 ≈ 10sThread 1A ──────────等待10s────────Thread 2B ──────────等待10s────────Pydantic类型注解用于校验输入数据是否符合预期类型。from pydantic import BaseModel class Product(BaseModel): sku: str price: float quantity: int product Product( skuABC123, price19.9, quantity10 )类型注解定义的是参数的类型约束两者的区别类型注解如def predict(price: float)用于告知 IDE 参数应为 float 类型。Pydantic 则在运行时真正校验输入数据是否符合类型要求。LangChain LlamaIndex——底层机制解释Runnable任何这种输入→输出的组件一个可以执行的节点每一个组件都是runnableRunnable├── Prompt├── Retriever├── LLM输入prompt,输出回答├── Parser└── Rerankerchain:将多个核心步骤串起来chain prompt | llm | parserRunnable1↓Runnable2↓Runnable3LlamaIndex专注于数据增强和文档索引将私有数据文本、表格、PDF 等组织成 LLM 可直接使用的上下文适合快速搭建文档问答系统或 RAG 系统尤其是文档结构复杂时。常规RAGLlamaIndexPDF 读取fitz / PyMuPDFDocument文本清洗is_noise_page、clean_text文本清洗is_noise_page、clean_text切块固定长度切分 /CharacterTextSplitterSentenceSplitter/TokenTextSplitterEmbeddingSentenceTransformer(all-mpnet-base-v2)IngestionPipeline向量库FAISSFaissVectorStore召回index.search(...)retriever / query engine重排调bge-reranker-v2-m3Reranker生成Claude 或本地 HuggingFace 模型QueryEngineRAG使用LLamaIndex实现#文档加载import os import fitz # PyMuPDF from loguru import logger from tqdm import tqdm PDF_PATH /opt/cyc/rag/simple-local-rag/human-nutrition-text.pdf LOG_DIR log os.makedirs(LOG_DIR, exist_okTrue) logger.remove() logger.add( f{LOG_DIR}/pdf_load_llamaindex.log, rotation200 MB, retention1, levelINFO, format{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}, encodingutf-8, )#数据清洗import re NOISE_KEYWORDS [ unsplash license, creative commons, attribution, photo by, image by, reused, license, copyright, all rights reserved, ] def is_noise_page(text): lower_text text.lower() hit_count sum(keyword in lower_text for keyword in NOISE_KEYWORDS) if hit_count 3: return True url_count len(re.findall(rhttps?://|www\., lower_text)) if url_count 5: return True return False def clean_text(text): text text.replace(\xa0, ) text re.sub(r[ \t], , text) text re.sub(r(?!\n)\n(?!\n), , text) text re.sub(r\n{3,}, \n\n, text) return text.strip() def load_pdf(pdf_path): pages [] with fitz.open(pdf_path) as doc: logger.info(f开始加载PDF: {pdf_path}, 总页数: {doc.page_count}) for page_index, page in tqdm(enumerate(doc), totaldoc.page_count): text clean_text(page.get_text()) if not text: logger.info(f第{page_index}页为空跳过) continue if is_noise_page(text): logger.info(f第{page_index}页为噪声页跳过) continue pages.append({ page: page_index, text: text, }) logger.info(fPDF加载完成有效页数: {len(pages)}) return pages pages load_pdf(PDF_PATH) print(f有效页数: {len(pages)})#文档读取from llama_index.core import Document documents [ Document(textpage[text], metadata{page: page[page]}) for page in pages ]#文本切分from llama_index.core.node_parser import SentenceSplitter splitter SentenceSplitter(chunk_size300, chunk_overlap20) nodes splitter.get_nodes_from_documents(documents) print(fDocument 数量: {len(documents)}) print(fNode 数量: {len(nodes)})#向量化和存储import faiss from llama_index.core import Settings, StorageContext, VectorStoreIndex from llama_index.embeddings.huggingface import HuggingFaceEmbedding from llama_index.vector_stores.faiss import FaissVectorStore embed_model HuggingFaceEmbedding( model_namesentence-transformers/all-mpnet-base-v2, cache_foldermodel, ) Settings.embed_model embed_model embed_dim len(embed_model.get_text_embedding(蛋白质是什么)) faiss_index faiss.IndexFlatIP(embed_dim) vector_store FaissVectorStore(faiss_indexfaiss_index) storage_context StorageContext.from_defaults(vector_storevector_store) index VectorStoreIndex( nodes, storage_contextstorage_context, embed_modelembed_model, ) INDEX_PATH rag/simple-local-rag/docs_llamaindex.index faiss.write_index(faiss_index, INDEX_PATH) print(f已保存索引到: {INDEX_PATH})#检索召回query 蛋白质的主要功能是什么 retriever index.as_retriever(similarity_top_k3) retrieved_nodes retriever.retrieve(query) def preview_nodes(nodes): for idx, node in enumerate(nodes, start1): score getattr(node, score, None) score_text f{score:.4f} if score is not None else N/A page node.metadata.get(page) text node.text.replace(\n, ) print(f[{idx}] score{score_text} | page{page}) print(text[:400]) print(- * 80) preview_nodes(retrieved_nodes)#精排可选#生成from anthropic import Anthropic def build_context_from_nodes(nodes): blocks [] for idx, node in enumerate(nodes, start1): page node.metadata.get(page, unknown) score getattr(node, score, None) score_text f{score:.4f} if score is not None else N/A blocks.append( f[Chunk {idx}] page{page}, score{score_text}\n{node.text} ) return \n\n.join(blocks) def answer_with_rag(query, retriever, top_k3, use_rerankerFalse): nodes retriever.retrieve(query)[:top_k] if use_reranker: nodes rerank_nodes_with_bge(query, nodes) context build_context_from_nodes(nodes) client Anthropic() prompt f你是一个基于资料回答问题的助手。 请只根据“参考资料”回答不要编造。 如果参考资料不足以回答就明确说“参考资料不足”。 用户问题 {query} 参考资料 {context} 请输出 1. 简洁答案 2. 如果能定位来源顺带标注 page resp client.messages.create( modelclaude-sonnet-5, max_tokens500, messages[{role: user, content: prompt}], ) return { answer: resp.content[0].text, context: context, nodes: nodes, } rag_result answer_with_rag( 蛋白质的主要功能是什么, retriever, use_rerankerFalse, ) print(rag_result[answer])