"""Measure what the assistant's tool search returns for real employee questions. Why this exists as a script rather than a test: retrieval quality is not a pass/fail property, it is a ranking, and the ranking depends on a live embedding backend plus every tool description in the registry. A unit test can pin "this description mentions 负责人"; only this can answer "would the model have been offered the right tool at all". It ranks the REAL read-only corpus (``action_search._corpus``) against a battery of questions and prints the top candidates — i.e. exactly the shortlist ``search_helix_tools`` would hand the model. Read-only; the only writes are to stdout. # against the local database (needs ai_spending.llm.api_key configured) cd server && .venv/bin/python scripts/eval_assistant_retrieval.py # against production, read-only — see reference_prod_access for the URL DATABASE_URL=mysql+asyncmy://…tidbcloud… \ .venv/bin/python scripts/eval_assistant_retrieval.py # add your own questions .venv/bin/python scripts/eval_assistant_retrieval.py 「我的工位在哪」 … How to read the output: a question whose correct tool is absent from the shortlist is a defect in a DESCRIPTION, not in the model — the model can only choose from what it is shown. Two failure shapes to look for: * the right tool missing entirely — its description does not name the words employees use. Add them (in the languages they use). * an administrative sibling outranking the self-service tool (the 工号 ledger above ``whoami``) — say what the sibling IS, positively. Do NOT write "this is not the answer to 「X」": the corpus is embedded verbatim, so quoting the question you are disclaiming makes the wrong tool MORE similar to it. Measured 2026-08-19: adding that disclaimer left the ranking unchanged at best. 2026-08-19 baseline (production embeddings, 275 read-only tools): 42 of the 50 questions below put a correct tool in the top 3, and none of the 50 comes back empty. What the other eight are is the useful part — it is a to-do list, not a clean bill of health: * 「电脑蓝屏了怎么办」「打印机怎么连」「怎么连访客 Wi-Fi」「离职流程是怎样的」 — how-to questions have no data tool BY DESIGN; they belong to ``search_knowledge_base`` (246 published articles, all embedded), which the channel prompt now names explicitly. Measured separately, the KB answers all four well except the bluescreen one, which no article covers — that one is a gap for IT to fill with an article, not with code. * 「我要提个 IT 工单」 — a write, answered by the propose path, not this corpus. * 「我的工号是多少」 — the HR allocation ledger outranks ``whoami`` (which sits at #4–5). Both legitimately contain 工号, and the prompt covers it by telling the model to call ``whoami`` for anything about the caller themselves. * 「我这个月的考勤怎么样」 — Helix only tracks INTERN attendance. A product gap, not a retrieval one. * 「我的账号都有哪些」 and 「哪些账号是离职员工的」 — no single tool answers either. "My accounts" is spread across SSO grants / API keys / email groups / the IT footprint, and leaver accounts across the cloud residue scans and seat reconciliation. These want an aggregate read, i.e. a change to the PRODUCT, and a description edit would only paper over it. """ from __future__ import annotations import asyncio import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import app.main # noqa: F401, E402 — registers every MCP tool from app.assistant import action_search # noqa: E402 from app.core.database import async_session_factory # noqa: E402 from app.domains.support.copilot import embed_texts # noqa: E402 #: Questions employees actually ask an IT assistant, in the words they use. #: Extend it — a question that reaches production and lands badly belongs here. QUESTIONS: tuple[str, ...] = ( # 设备 / 资产 "我有几台电脑", "我的电脑保修到什么时候", "我的笔记本序列号是多少", "电脑坏了怎么报修", "我要借一台备用机", "我的显示器是哪一台", # 软件 "我有哪些软件坐席", "Figma 怎么申请", "Netsuite这个应用的负责人是谁", "这个软件一年多少钱", "我的软件花了多少钱", "公司有哪些设计软件", # 账号 / 身份 "我的 GitHub id 是什么", "我在多少个邮件组里", "我的工号是多少", "我的账号都有哪些", "我的邮箱别名有哪些", # HR "我还有多少年假", "我的经理是谁", "我什么时候入职的", "我在哪个部门", "我这个月的考勤怎么样", "我的直属下属有谁", "公司有哪些法定假期", # 入离职 "新员工入职指南", "离职流程是怎样的", "我的入职待办还有哪些", # 审批 / 请求 "我提交的申请到哪一步了", "有什么需要我审批的", "我的申请为什么被拒了", # 网络 / VPN / 云 "VPN 怎么申请", "怎么连访客 Wi-Fi", "我有哪些云权限", "我的 AWS 账号是哪个", "API key 怎么申请", "我的 API key 用了多少钱", # IT 求助 (these belong to the knowledge base, not to a data tool) "电脑蓝屏了怎么办", "打印机怎么连", "我要提个 IT 工单", # IT / admin 视角 "张三的电脑有哪些", "谁在用 Figma 的坐席", "这个月软件花了多少钱", "哪些证书快到期了", "哪些人还没做安全自查", "上个月云成本是多少", "有哪些没人认领的资产", "这个部门有多少人", "谁是这个部门的负责人", "哪些账号是离职员工的", "本月入职了哪些人", ) #: How many candidates the bridge shows the model. Mirrors #: ``mcp_bridge._SEARCH_LIMIT`` — showing more here would flatter the ranking. SHORTLIST = 6 async def main(questions: tuple[str, ...]) -> int: async with async_session_factory() as db: corpus = action_search._corpus(action_search.CORPUS_READONLY) names = list(corpus) print(f"corpus: {len(names)} read-only tools", flush=True) embedded = await embed_texts( db, [corpus[n] for n in names], timeout=300.0, ) if embedded is None: print( "embeddings unavailable — the assistant is running on lexical " "matching, which is what this script exists to detect. Check " "ai_spending.llm.api_key and copilot.enabled.", ) return 1 vectors, model = embedded index = dict(zip(names, vectors, strict=False)) print(f"model: {model}", flush=True) probes = await embed_texts(db, list(questions), timeout=300.0) if probes is None: print("could not embed the questions") return 1 empty = 0 for question, probe in zip(questions, probes[0], strict=False): scored = sorted( ( (action_search._cosine(probe, vector), name) for name, vector in index.items() ), key=lambda pair: (-pair[0], pair[1]), )[:SHORTLIST] kept = [ (score, name) for score, name in scored if score >= action_search.MIN_SCORE ] print(f"\n{question}") for score, name in kept: print(f" {score:.3f} {name}") if not kept: empty += 1 print( " *** nothing above MIN_SCORE — falls back to lexical " "matching over an English corpus ***", ) print(f"\n{empty}/{len(questions)} questions matched no tool at all") return 0 if __name__ == "__main__": asked = tuple(sys.argv[1:]) or QUESTIONS raise SystemExit(asyncio.run(main(asked)))