# QMDSearch API 示例与 Python 客户端封装 > 基址占位符 `{BASE}` 替换为实际服务地址，例如 `http://localhost:8000` 或 `http://:8000`。 ## 一、curl 示例 ### 登录 ```bash curl -s -X POST {BASE}/api/v1/auth/login \ -H "Content-Type: application/json" \ -d '{"username":"admin","password":""}' ``` ### 文本入库（202 + task_id） ```bash curl -s -X POST {BASE}/api/v1/documents \ -H "Authorization: Bearer " \ -H "Content-Type: application/json" \ -d '{"text":"QMDSearch 是面向 AI Agent 的分层信息检索服务……","title":"QMDSearch 介绍","source":"notes"}' ``` ### 文件上传（202 + task_id） ```bash curl -s -X POST {BASE}/api/v1/documents/upload \ -H "Authorization: Bearer " \ -F "file=@document.pdf" \ -F "source=manual" ``` ### 轮询任务状态 ```bash curl -s {BASE}/api/v1/documents/tasks/ ``` ### 批量文件上传（202，逐文件建任务） ```bash curl -s -X POST {BASE}/api/v1/documents/upload-batch \ -H "Authorization: Bearer " \ -F "files=@a.pdf" -F "files=@b.md" ``` ### 列出入库任务 / 重试 / 删除任务 ```bash curl -s -H "Authorization: Bearer " "{BASE}/api/v1/documents/tasks?limit=20" curl -s -X POST -H "Authorization: Bearer " {BASE}/api/v1/documents/tasks//retry curl -s -X DELETE -H "Authorization: Bearer " {BASE}/api/v1/documents/tasks/ ``` ### 重新摘要入库（读原文件 → 删旧 → 重跑） ```bash curl -s -X POST -H "Authorization: Bearer " {BASE}/api/v1/documents//reingest ``` ### 检索（免登录） ```bash curl -s -X POST {BASE}/api/v1/search \ -H "Content-Type: application/json" \ -d '{"query":"如何配置 Redis 缓存","top_k":5,"summarize":false}' ``` ## 二、Python 客户端封装 可直接在 Agent 工具代码里复用： ```python import time import requests class QMDSearchClient: """QMDSearch 最小客户端：登录、入库（文本/文件）、轮询、检索。""" def __init__(self, base_url: str, username: str, password: str, poll_interval: float = 1.5, poll_timeout: float = 120.0): self.base = base_url.rstrip("/") self.username = username self.password = password self.poll_interval = poll_interval self.poll_timeout = poll_timeout self.token: str | None = None # ---- 鉴权 ---- def login(self) -> str: r = requests.post( f"{self.base}/api/v1/auth/login", json={"username": self.username, "password": self.password}, timeout=30, ) r.raise_for_status() body = r.json() if body.get("code") != 0: raise RuntimeError(f"登录失败: {body}") self.token = body["data"]["token"] return self.token def _headers(self) -> dict: if not self.token: self.login() return {"Authorization": f"Bearer {self.token}"} def _ok(self, resp: requests.Response): resp.raise_for_status() body = resp.json() if body.get("code") != 0: raise RuntimeError(f"API 错误 code={body.get('code')} msg={body.get('message')}") return body["data"] # ---- 入库 ---- def ingest_text(self, text: str, title: str = "", source: str = "", metadata: dict | None = None) -> str: data = {"text": text} if title: data["title"] = title if source: data["source"] = source if metadata: data["metadata"] = {str(k): str(v) for k, v in metadata.items()} resp = requests.post(f"{self.base}/api/v1/documents", headers=self._headers(), json=data, timeout=30) return self._ok(resp)["task_id"] def upload_file(self, path: str, title: str = "", source: str = "", metadata: dict | None = None) -> str: files = {"file": open(path, "rb")} data = {} if title: data["title"] = title if source: data["source"] = source if metadata: data["metadata"] = str({str(k): str(v) for k, v in metadata.items()}) resp = requests.post(f"{self.base}/api/v1/documents/upload", headers=self._headers(), files=files, data=data, timeout=60) return self._ok(resp)["task_id"] def upload_batch(self, paths: list[str]) -> dict: """批量上传多个文件：返回 {tasks:[{filename,task_id}], failed:[{filename,error}]}""" files = [("files", open(p, "rb")) for p in paths] resp = requests.post(f"{self.base}/api/v1/documents/upload-batch", headers=self._headers(), files=files, timeout=60) return self._ok(resp) # ---- 轮询 ---- def wait_task(self, task_id: str) -> dict: deadline = time.time() + self.poll_timeout while time.time() < deadline: resp = requests.get(f"{self.base}/api/v1/documents/tasks/{task_id}", timeout=30) data = self._ok(resp) if data["status"] == "done": return data["result"] if data["status"] == "failed": raise RuntimeError(f"入库失败: {data.get('error')}") time.sleep(self.poll_interval) raise TimeoutError(f"任务 {task_id} 轮询超时") def ingest_text_wait(self, *args, **kwargs) -> dict: return self.wait_task(self.ingest_text(*args, **kwargs)) def upload_file_wait(self, *args, **kwargs) -> dict: return self.wait_task(self.upload_file(*args, **kwargs)) # ---- 检索 ---- def search(self, query: str, top_k: int | None = None, summarize: bool = False) -> dict: payload = {"query": query, "summarize": summarize} if top_k is not None: payload["top_k"] = top_k resp = requests.post(f"{self.base}/api/v1/search", headers=self._headers(), json=payload, timeout=60) return self._ok(resp) # 用法 if __name__ == "__main__": client = QMDSearchClient("http://localhost:8000", "admin", "") client.login() # 文本入库并等待完成 result = client.ingest_text_wait("这是一篇关于分层检索的笔记……", title="笔记") print("document_id =", result["document_id"]) # 检索 hits = client.search("分层检索是什么", top_k=5)["hits"] for h in hits: print(f"[{h['score']:.3f}] {h['title']}: {h['text'][:80]}") ``` ## 三、响应结构速记 - 文本 / 文件 / 批量入库 `202` → `data = {task_id, status:"pending"[, saved_path]}`；批量返回 `{tasks, failed}` - 任务 `done` → `data.result = {document_id, summary:{l1_summary,l2_outline,l3_content_outline,level}, category, chunks_count, tags, category_confidence, deduplicated}` - 检索 → `data = {query, hits:[{text,doc_id,title,section_path,score,doc_summary}], routed_categories, fallback, extracted_info, summary?}`