最近一段时间，各大模型公司、APP都掀起了一阵养虾潮，不过，平心而论，有什么APP的刚需性，还有使用频率，会超过微信呢？
正好，最近腾讯官方的QClaw 出来了，可以直接接到微信的聊天框里，所以我第一时间，把它和我日常工作——Milvus运维结合了起来。
以下是实操步骤。
01
QClaw怎么用
最近各大APP主推的各种Claw都主打不用终端，不用配环境，不用自备 API Key，点击安装包，扫个码即可使用。
QClaw 也不例外，先安装并用微信账号登录
装完以后，整个界面长这样，首页有几个功能属于是开箱即用的。
这里需要输入邀请码才可以继续
Qclaw 模型支持自定义接入模型或者选择默认大模型都可以的，本文我选择默认大模型。
配置关联微信就可以用手机端微信遥控龙虾了
扫码完成以后，微信通讯录里直接多了一个 QClaw 的联系人。打开手机微信就会收到来自龙虾的问候：
先给它一个身份让它记住：
就这样，电脑和手机微信，连上了。
02
如何让Qclaw，搞定Milvus运维
我日常维护 Milvus，有两件事天天都要干。
一个是嵌入脚本——调模型、把文档向量化、写入 Milvus。
#!/usr/bin/env python3import argparseimport osfrom pathlib import Pathfrom pymilvus import MilvusClient, DataTypefrom openai import OpenAIimport pdfplumber# ── 配置区（按需修改）──────────────────────────────────MILVUS_URI = "http://localhost:19530"MILVUS_TOKEN = "root:Milvus"OPENAI_MODEL = "text-embedding-3-small" # 换成 text-embedding-3-large 则改 VECTOR_DIM=3072VECTOR_DIM = 1536CHUNK_SIZE = 400CHUNK_OVERLAP = 50BATCH_SIZE = 64 # OpenAI 单次最多 2048 条，64 条一批兼顾速度和稳定性# ────────────────────────────────────────────────────────# 从环境变量读取，bash 里 export OPENAI_API_KEY=sk-xxx 即可# openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))openai_client = OpenAI(api_key="sk-xxxxxxx")def get_embeddings(texts: list[str]) -> list[list[float]]: """批量调用 OpenAI Embedding API，返回向量列表""" all_vectors = [] for i in range(0, len(texts), BATCH_SIZE): batch = texts[i : i + BATCH_SIZE] resp = openai_client.embeddings.create( model=OPENAI_MODEL, input=batch, ) all_vectors.extend([item.embedding for item in resp.data]) return all_vectorsdef read_file(path: Path) -> str: suffix = path.suffix.lower() if suffix == ".pdf": with pdfplumber.open(path) as pdf: return "\n".join(page.extract_text() or "" for page in pdf.pages) return path.read_text(encoding="utf-8", errors="ignore")def chunk_text(text: str, size: int, overlap: int) -> list[str]: chunks, start = [], 0 while start < len(text): chunk = text[start : start + size] if len(chunk.strip()) > 20: chunks.append(chunk) start += size - overlap return chunksdef ensure_collection(client: MilvusClient, name: str) -> None: if client.has_collection(name): print(f"[INFO] Collection '{name}' 已存在，追加写入") return schema = MilvusClient.create_schema( auto_id=True, enable_dynamic_field=False, ) schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True) schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=VECTOR_DIM) schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=2000) schema.add_field(field_name="source", datatype=DataType.VARCHAR, max_length=512) index_params = client.prepare_index_params() index_params.add_index( field_name="vector", index_type="AUTOINDEX", metric_type="COSINE", ) client.create_collection( collection_name=name, schema=schema, index_params=index_params, ) print(f"[INFO] Collection '{name}' 创建并加载成功")def main() -> None: parser = argparse.ArgumentParser(description="文档向量化入库 Milvus（OpenAI Embedding）") parser.add_argument("--folder", required=True, help="文档目录（PDF / MD / TXT）") parser.add_argument("--collection", required=True, help="目标 Milvus collection 名称") args = parser.parse_args() # API Key 已在全局配置区设置，无需检查环境变量 folder = Path(args.folder).expanduser().resolve() if not folder.exists(): print(f"[ERROR] 目录不存在：{folder}") return files = ( list(folder.glob("**/*.pdf")) + list(folder.glob("**/*.md")) + list(folder.glob("**/*.txt")) ) if not files: print("[WARN] 未找到任何文档，退出") return print(f"[INFO] 发现 {len(files)} 个文件，开始处理...") milvus = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN) ensure_collection(milvus, args.collection) total_chunks = 0 for f in files: try: text = read_file(f) chunks = chunk_text(text, CHUNK_SIZE, CHUNK_OVERLAP) if not chunks: print(f" [SKIP] {f.name}（无有效文本）") continue vectors = get_embeddings(chunks) data = [ {"vector": v, "text": c, "source": str(f)} for v, c in zip(vectors, chunks) ] res = milvus.insert(collection_name=args.collection, data=data) total_chunks += res["insert_count"] print(f" ✅ {f.name} → {res['insert_count']} 块") except Exception as e: print(f" [ERROR] {f.name} 处理失败：{e}") print(f"\n[DONE] 共写入 {total_chunks} 条向量 → collection: {args.collection}")if __name__ == "__main__": main()一个是巡检脚本——检查 Collection 状态、索引健康度、有没有异常。
#!/usr/bin/env python3import osimport timefrom datetime import datetimefrom pymilvus import MilvusClient, DataTypefrom openai import OpenAI# ── 配置区（按需修改）──────────────────────────────────MILVUS_URI = "http://localhost:19530"MILVUS_TOKEN = "root:Milvus"OPENAI_MODEL = "text-embedding-3-small"VECTOR_DIM = 1536REPORT_PATH = os.path.expanduser("~/Documents/milvus_daily_report.txt")# 探针句子：用这句话做语义检索，测试各 collection 的真实查询延迟# 可以改成你自己业务中有代表性的一句话PROBE_TEXT = "What is vector database and how does it work?"# 延迟预警阈值（毫秒），超过此值标记为偏慢SLOW_QUERY_MS = 100# ────────────────────────────────────────────────────────# openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))openai_client = OpenAI(api_key="sk-xxxx")def get_probe_vector() -> list[float]: """用 OpenAI 把探针句子转成向量""" resp = openai_client.embeddings.create( model=OPENAI_MODEL, input=[PROBE_TEXT], ) return resp.data[0].embeddingdef get_vector_field(client: MilvusClient, collection: str) -> tuple[str, int]: """ 返回 (向量字段名, 向量维度) 优先取第一个 FLOAT_VECTOR 字段 """ try: desc = client.describe_collection(collection) for field in desc.get("fields", []): if field.get("type") == "FLOAT_VECTOR": dim = field.get("params", {}).get("dim", VECTOR_DIM) return field["name"], int(dim) except Exception: pass return "vector", VECTOR_DIM # fallbackdef check_collection( client: MilvusClient, name: str, probe_vector: list[float],) -> dict: try: # 向量数 stats = client.get_collection_stats(name) count = int(stats.get("row_count", 0)) # 获取向量字段信息 vec_field, dim = get_vector_field(client, name) # 维度不匹配时降级为随机向量，避免查询报错 if dim != len(probe_vector): import random query_vec = [random.random() for _ in range(dim)] probe_note = f"（维度不符 {dim}D，已用随机向量）" else: query_vec = probe_vector probe_note = f"（OpenAI 语义探针，{dim}D）" # 真实 search，测量延迟 start = time.time() client.search( collection_name=name, data=[query_vec], anns_field=vec_field, limit=3, search_params={"metric_type": "COSINE"}, output_fields=[], ) latency_ms = round((time.time() - start) * 1000, 1) # 判断状态 if latency_ms > SLOW_QUERY_MS: status = "🐢 查询偏慢" else: status = "✅ 正常" return { "name": name, "count": count, "latency_ms": latency_ms, "status": status, "probe_note": probe_note, } except Exception as e: return { "name": name, "count": -1, "latency_ms": -1, "status": f"❌ 错误", "probe_note": str(e), }def main() -> None: # API Key 已在全局配置区设置，无需检查环境变量 client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN) collections = client.list_collections() if not collections: msg = "[INFO] 当前 Milvus 中没有任何 collection" print(msg) return # 只生成一次探针向量，所有 collection 复用 print("[INFO] 正在生成 OpenAI 语义探针向量...") probe_vector = get_probe_vector() print(f"[INFO] 探针向量维度：{len(probe_vector)}D，句子：\"{PROBE_TEXT}\"") date_str = datetime.now().strftime("%Y/%m/%d %H:%M") lines = [f"📊 Milvus 日报 - {date_str}", "=" * 48] all_ok = True for col in sorted(collections): info = check_collection(client, col, probe_vector) count_str = f"{info['count']:,}" if info["count"] >= 0 else "N/A" latency_str = f"{info['latency_ms']}ms" if info["latency_ms"] >= 0 else "N/A" lines.append( f"{info['status']} {info['name']}\n" f" 向量数：{count_str} | 查询延迟：{latency_str}\n" f" {info['probe_note']}" ) if "❌" in info["status"] or "🐢" in info["status"]: all_ok = False lines.append("=" * 48) lines.append( "✅ 所有 collection 状态健康" if all_ok else "⚠️ 存在异常 collection，请检查" ) lines.append(f"探针语句：\"{PROBE_TEXT}\"") report = "\n".join(lines) print("\n" + report) # 写入本地文件，方便 Qclaw 读取并推送 os.makedirs(os.path.dirname(REPORT_PATH), exist_ok=True) with open(REPORT_PATH, "w", encoding="utf-8") as f: f.write(report) print(f"\n[INFO] 报告已保存 → {REPORT_PATH}")if __name__ == "__main__": main()这两个脚本本身没问题，问题是每次跑都要开电脑、进终端、手动触发。
不难，就是麻烦。
跑之前先初始化一下，让它记住脚本路径和文档目录。
嵌入文本这边：
掏出手机，微信发一条：
把 milvus 文件夹里的文档入库，完成后告诉我写了多少条。
巡检脚本这边：
掏出手机就能触发
“帮我跑一下 Milvus 巡检，结果发我”
结果直接回到微信里。
然后我顺手把巡检挂成了定时任务，每天早上 9 点自动跑，结果推微信。
粤ICP备14082021号
免费POC，
零成本试错
