QClaw养虾教程|我把OpenClaw装进了微信，无痛搞定 Milvus 运维巡检

推荐语

将OpenClaw接入微信，轻松实现Milvus运维自动化，告别繁琐操作！

核心内容：
1. QClaw的安装与微信关联步骤
2. 如何配置QClaw进行Milvus日常运维
3. 嵌入脚本实现文档向量化与写入Milvus的实操方法

杨芳贤

53AI创始人/腾讯云(TVP)最具价值专家

最近一段时间，各大模型公司、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 点自动跑，结果推微信。

第二天早上，电脑还没开，巡检结果已经在手机上了。

接下来，我计划把剩下一些一直吃灰的脚本，接下来一个个挂进去。大家有什么好的想法，欢迎评论区分享。

阅读推荐
黄仁勋GTC演讲上，Milvus为什么能站稳非结构化数据处理C位
Agent都是伪需求！如何判断是否需要Multi-Agent，以及如何搭？" data-itemshowtype="0" linktype="text" data-linktype="2">80%的 Multi-Agent都是伪需求！如何判断是否需要Multi-Agent，以及如何搭？
教程：OpenClaw + Milvus + LLM，如何避免GEO批量生文变AI投毒 | 一线实战
养虾实战教程：我用OpenClaw做了个能盯盘，也能深度复盘的投资agent