#WorkBuddy# 哪些活儿不必烧积分：把本地 Ollama 模型接进 WorkBuddy 的混合用法（附体检脚本）

# 安装（Homebrew，macOS）
brew install ollama

# 拉一个适合干活的小模型（示例用 gemma4:e4b，约 9.6GB）
ollama pull gemma4:e4b

# 确认服务在监听默认端口
lsof -nP -iTCP:11434 -sTCP:LISTEN

curl -s http://127.0.0.1:11434/v1/models

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
local_model_probe.py —— 本地模型「能不能接这活儿」体检脚本
运行环境：macOS（Apple Silicon 实测），Python 3.9+，无第三方依赖
用法：
    python3 local_model_probe.py                 # 用内置样例跑 3 轮
    python3 local_model_probe.py --rounds 5
    python3 local_model_probe.py --model qwen3:8b
"""

import argparse, json, statistics, time, urllib.error, urllib.request

OLLAMA_BASE = "http://127.0.0.1:11434/v1"
DEFAULT_MODEL = "gemma4:e4b"

# 体检用样例任务：刻意选「结构化、低创造、高频」三类活儿
# 说明：样例内容全部为虚构演示数据，不含任何真实业务信息
TASK_BANK = {
    "会议纪要待办提取": {
        "system": ("你是会议纪要助手。只输出 JSON 数组，不要任何解释文字。"
                   "每项包含 todo（动作）、owner（负责人，未知填null）、due（截止时间原文，无则null）。"),
        "user": ("从下面这段会议记录里提取待办事项：\n"
                 "张三说样品检测报告下周三前必须给到采购部。"
                 "李四提了个问题，客户那边的验收标准还没最终确认，需要再跟对方核对一次。"
                 "王五表示仓库那边的新货架本周就能装完，装完他会拍照发群里。"
                 "最后大家同意下次例会时间不变。\n只输出 JSON 数组。"),
        "check": lambda out: out.strip().startswith("["),
    },
    "批量重命名规则生成": {
        "system": ("你是文件整理助手。用户给一批文件名和命名规则，"
                   "你只输出重命名后的文件名列表，一行一个，不要序号、不要解释。"),
        "user": ("规则：项目缩写_日期_序号.扩展名，日期用 YYYYMMDD，序号三位补零。\n"
                 "原文件：立项说明.docx、现场照片.jpg、结算单.xlsx、验收记录.pdf\n"
                 "项目缩写：DEMO，日期：2026年10月17日。\n只输出四个新文件名，一行一个。"),
        "check": lambda out: out.count("\n") >= 3 and "DEMO_20261017" in out,
    },
    "长文本摘要": {
        "system": "你是摘要助手。用不超过 80 字的中文概括要点，只输出摘要本身。",
        "user": ("请概括下面这段话：\n本周的主要工作集中在三块。第一块是流程梳理，把原来散落在"
                 "多个表格里的登记环节合并成了一张总表，字段从原来的二十多个精简到九个，填写时间明显下降。"
                 "第二块是历史数据补录，把过去半年的纸质记录录入系统，期间发现有三处填写口径不一致，已经统一。"
                 "第三块是对接测试，与上下游系统做了两次联调，第一次因为字段命名不一致失败，第二次通过。"
                 "下周计划继续推进剩余模块的验证。"),
        "check": lambda out: 5 < len(out.strip()) < 200,
    },
}

def strip_fence(text: str) -> str:
    """剥掉本地小模型最爱加的 ```json / ``` 围栏。

    实测里最常踩的坑：system 里明明写了「只输出 JSON」，
    本地模型还是习惯性包一层 Markdown 代码块，下游 json.loads 直接炸。
    所以校验和落库之前必须先剥壳。
    """
    t = text.strip()
    if t.startswith("```"):
        t = t.split("\n", 1)[1] if "\n" in t else ""
        if t.rstrip().endswith("```"):
            t = t.rstrip()[:-3]
    return t.strip()

def chat(base, api_key, model, system, user, temperature=0.2, timeout=300):
    """调用 OpenAI 兼容的 /chat/completions，返回 内容+耗时+用量。"""
    payload = {
        "model": model,
        "messages": [{"role": "system", "content": system},
                     {"role": "user", "content": user}],
        "temperature": temperature,
        "stream": False,
    }
    req = urllib.request.Request(
        base.rstrip("/") + "/chat/completions",
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json",
                 "Authorization": "Bearer " + api_key},
        method="POST",
    )
    t0 = time.time()
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        data = json.loads(resp.read().decode("utf-8"))
    elapsed = time.time() - t0
    content = data["choices"][0]["message"]["content"]
    return {"content": content.strip(),
            "content_clean": strip_fence(content),
            "elapsed": elapsed,
            "usage": data.get("usage", {})}

def probe(base, api_key, model, rounds):
    print("=" * 62)
    print(f"体检目标：{model}  @ {base}")
    print(f"每类任务跑 {rounds} 轮，取中位数")
    print("=" * 62)

    summary = []
    for name, task in TASK_BANK.items():
        times, oks, sample = [], 0, ""
        for i in range(rounds):
            try:
                r = chat(base, api_key, model, task["system"], task["user"])
            except urllib.error.URLError as e:
                print(f"  [!] {name} 第{i+1}轮请求失败：{e}")
                continue
            except (KeyError, json.JSONDecodeError) as e:
                print(f"  [!] {name} 第{i+1}轮返回解析失败：{e}")
                continue
            times.append(r["elapsed"])
            if task["check"](r["content_clean"]):
                oks += 1
            sample = r["content_clean"]

        if not times:
            print(f"\n{name}：全部失败，检查服务是否启动")
            continue

        med = statistics.median(times)
        print(f"\n{name}")
        print(f"  耗时中位数：{med:.2f} s   (min {min(times):.2f} / max {max(times):.2f})")
        print(f"  格式合规率：{oks}/{len(times)}")
        print(f"  样例输出：{sample[:160].replace(chr(10), ' / ')}")
        summary.append((name, med, oks, len(times)))

    if summary:
        print("\n" + "-" * 62)
        print("结论表（可下放 = 格式合规率 100% 且 耗时可接受）")
        for name, med, oks, total in summary:
            verdict = "可下放" if oks == total else "不建议下放"
            print(f"  {name:<14} {med:>6.2f}s  {oks}/{total}  {verdict}")
        print("-" * 62)

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--base", default=OLLAMA_BASE)
    ap.add_argument("--model", default=DEFAULT_MODEL)
    ap.add_argument("--api-key", default="ollama")
    ap.add_argument("--rounds", type=int, default=3)
    args = ap.parse_args()

    # 连通性自检，避免后面一堆超时误判
    try:
        with urllib.request.urlopen(args.base.rstrip("/") + "/models", timeout=5) as resp:
            models = [m["id"] for m in json.loads(resp.read())["data"]]
        print(f"服务在线，可用模型：{models}")
    except Exception as e:
        print(f"[致命] 连不上 {args.base}：{e}")
        print("排查：ollama serve 起了吗？端口是 11434 吗？")
        return

    if args.model and args.model not in models:
        print(f"[警告] 模型 {args.model} 不在列表里，可能没拉取：ollama pull {args.model}")

    probe(args.base, args.api_key, args.model, args.rounds)

if __name__ == "__main__":
    main()

# 后台常驻（macOS）
nohup ollama serve > /tmp/ollama.log 2>&1 &

# 或者用 brew services 托管
brew services start ollama