🤖 OpenClaw 本地知识问答助手

# PDF 处理 pip install pymupdf pdfplumber paddleocr # Word 处理 pip install python-docx # Excel 处理 pip install openpyxl pandas # PPT 处理 pip install python-pptx # OCR 引擎 (扫描版 PDF) pip install paddlepaddle paddleocr

knowledge-assistant/ ├── docs/ # 知识库文档目录 │ ├── technical/ # 技术文档 │ ├── business/ # 业务资料 │ └── faq/ # 常见问题 ├── vectors/ # 向量数据库存储 ├── scripts/ │ ├── index_knowledge.py # 文档索引脚本 │ ├── qa_engine.py # 问答引擎核心 │ └── update_index.sh # 增量更新脚本 ├── web/ │ ├── app.py # FastAPI 服务 │ ├── index.html # Web 界面 │ └── static/ # 静态资源 ├── openclaw/ │ ├── qa_agent.py # OpenClaw 子 agent │ └── config.yaml # 配置文件 ├── requirements.txt # Python 依赖 └── README.md # 使用说明

import chromadb from sentence_transformers import SentenceTransformer from unstructured.partition.auto import partition # 初始化 model = SentenceTransformer('m3e-base') client = chromadb.PersistentClient(path='./vectors') collection = client.get_or_create_collection('knowledge') # 文档解析 def parse_document(filepath): elements = partition(filepath) text = "\n\n".join([str(el) for el in elements]) return text # 文本分块 def chunk_text(text, size=500, overlap=50): chunks = [] start = 0 while start < len(text): end = start + size chunk = text[start:end] chunks.append(chunk) start += size - overlap return chunks # 建立索引 for filepath in scan_docs(): text = parse_document(filepath) chunks = chunk_text(text) for i, chunk in enumerate(chunks): embedding = model.encode(chunk) collection.add( documents=[chunk], embeddings=[embedding], ids=[f"{filepath}_{i}"], metadatas=[{"source": filepath}] )

def query_knowledge(question, top_k=5): # 生成问题向量 query_embedding = model.encode(question) # 向量检索 vector_results = collection.query( query_embeddings=[query_embedding], n_results=top_k * 2 ) # 关键词检索 (BM25) keyword_results = bm25_search(question, top_k=top_k) # 混合重排序 (RRF) reranked = reciprocal_rank_fusion( vector_results, keyword_results ) # 提取答案 answer = extract_answer(reranked[:top_k], question) return answer

# 1. 创建问答子 agent sessions_spawn( runtime="subagent", task="回答用户关于知识库的问题", mode="session" ) # 2. 监听飞书消息 (通过 OpenClaw 插件) # 当收到@提及或关键词时触发 # 3. 调用本地问答 API import requests response = requests.post( "http://localhost:8000/query", json={"question": user_question} ) answer = response.json()["answer"] # 4. 回传飞书 message( action="send", channel="feishu", to=user_id, message=answer )