v0.6-wip: build_report \u7edf\u4e00\u5165\u53e3 + build_glossary \u672f\u8bed\u6838\u67e5

\u7ee7\u7eed\u89e3\u51b3\u7528\u6237\u53cd\u9988\u7684 PDF \u95ee\u9898\u3002

scripts/build_report.py\uff08\u65b0\u589e\uff09\uff1a
- \u5355\u4e00\u5165\u53e3\u540c\u65f6\u51fa PDF + DOCX
- \u6587\u4ef6\u540d\u81ea\u52a8\u4ece manifest.report_title \u751f\u6210\uff08\u89e3\u51b3 "final.pdf" \u6CDB\u540d\u95EE\u9898\uff09
- pandoc --from=markdown-tex_math_dollars \u4fee\u590d DOCX \u751f\u6210\u65f6\u7684 $ \u8bef\u89e3
- \u81ea\u52a8\u5bfb\u627e phase2/sources.jsonl \u4f5c\u4e3a\u53c2\u8003\u6587\u732e\u5f15\u6587\u6e90

scripts/lib/search_client.py\uff08\u65b0\u589e\uff09\uff1a
- Exa \u4e3b\u529b + Tavily fallback \u7684\u7edf\u4e00\u63a5\u53e3
- \u5173\u952e\u4fee\u590d\uff1atrust_env=False \u7ed5\u5f00\u7cfb\u7edf socks5 \u4ee3\u7406
  \uff08Clash on macOS \u5c0a httpx TLS \u63e1\u624b\u5728 CONNECT \u540e EOF\uff09

scripts/build_glossary.py\uff08\u65b0\u589e\uff09\uff1a
- \u7528\u7684\u4e92\u65b9\u5f0f\u89e3\u51b3\u4e86\u7528\u6237\u53cd\u9988 #6\uff1a\u672f\u8bed\u7ffb\u8bd1\u4e0d\u4e13\u4e1a / \u4e8b\u5b9e\u9519\u8bef
- ThreadPoolExecutor \u5e76\u53d1\uff08\u9ed8\u8ba4 6 worker\uff09\uff0c\u6bcf\u4e2a\u672f\u8bed\u72ec\u7acb\uff1a
  Search \u2192 Top-3 snippet \u2192 Haiku \u5224\u5b9a \u2192 \u8fd4\u56de {zh, en_full, confidence, issue}
- \u5b9e\u6d4b\u6210\u529f\u8bc6\u522b "Maywavee" \u4e3a "Mabwell" \u7684\u62fc\u5199\u9519\u8bef\u5e76\u6807\u51fa issue
- \u65ad\u70b9\u7eed\u4f20\uff08\u5df2\u6807 verified_at \u7684\u9ed8\u8ba4\u8df3\u8fc7\uff09
- Haiku \u6210\u672c\u6781\u4f4e\uff083 \u4e2a\u672f\u8bed\u8c03\u7528 \u2248 0.01 \u7f8e\u5206\uff09
- \u652f\u6301 --extra terms.txt \u8865\u5145\u7ffb\u8bd1\u9636\u6bb5\u672a\u6536\u5165\u7684\u672f\u8bed

scripts/prompts/glossary_system.txt\uff08\u65b0\u589e\uff09\uff1a
- Haiku \u6838\u67e5\u672f\u8bed\u7684 prompt\uff0c\u660e\u786e\u5224\u5b9a\u7ef4\u5ea6\u548c JSON \u8f93\u51fa\u683c\u5f0f

Co-authored-by: User <human>
This commit is contained in:
kai
2026-04-22 13:12:01 +08:00
co-authored by User <human>
parent a86010e9a7
commit d3fde1cbb8
6 changed files with 1761 additions and 310 deletions
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,3 @@
{"tag": "glossary:phosphoramidite monomer", "model": "anthropic/claude-haiku-4.5", "attempt": 0, "elapsed": 2.81, "usage": {"completion_tokens": 48, "prompt_tokens": 2187, "total_tokens": 2235, "prompt_tokens_details": {"cached_tokens": 0, "ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 0, "web_search": 0, "cacheCreationInputTokens": 0}}, "out_chars": 101, "status": 200}
{"tag": "glossary:RNA interference", "model": "anthropic/claude-haiku-4.5", "attempt": 0, "elapsed": 2.82, "usage": {"completion_tokens": 37, "prompt_tokens": 2585, "total_tokens": 2622, "prompt_tokens_details": {"cached_tokens": 0, "ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 0, "web_search": 0, "cacheCreationInputTokens": 0}}, "out_chars": 93, "status": 200}
{"tag": "glossary:Maywavee", "model": "anthropic/claude-haiku-4.5", "attempt": 0, "elapsed": 3.71, "usage": {"completion_tokens": 136, "prompt_tokens": 3185, "total_tokens": 3321, "prompt_tokens_details": {"cached_tokens": 0, "ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 0, "web_search": 0, "cacheCreationInputTokens": 0}}, "out_chars": 186, "status": 200}
+332
View File
@@ -0,0 +1,332 @@
#!/usr/bin/env python3
"""双语术语表事实核查脚本。
输入:
- <project>/phase4/glossary.json(来自 translate.py 累积的初版术语表)
- 可选:--extra terms.txt(每行一个英文术语,补充进来一起核查)
流程(每个术语独立可并行):
1. 用 SearchClientExa > Tavily)搜一次(query = "<term> <domain hint>"
2. 把 top 3-5 snippet 喂给 Haiku,让模型返回 {zh, en_full, confidence, issue}
3. 合并回 glossary,字段扩展:
{
"Mabwell": {
"zh": "迈威生物",
"en_full": "Mabwell (Shanghai) Bioscience Co., Ltd.",
"confidence": "high",
"issue": "...",
"verified_at": "2026-04-22",
"sources": ["https://mabwell.com/", ...]
}
}
用法:
uv run python scripts/build_glossary.py <project_slug>
uv run python scripts/build_glossary.py <project_slug> --workers 6
uv run python scripts/build_glossary.py <project_slug> --only "Mabwell,Maywavee"
uv run python scripts/build_glossary.py <project_slug> --force
断点续传:已核查过的条目(有 verified_at 字段)默认跳过;--force 全部重跑。
"""
from __future__ import annotations
import argparse
import concurrent.futures
import datetime as dt
import json
import sys
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from scripts.lib.search_client import SearchClient, SearchError
from scripts.lib.zenmux_client import ZenMuxClient, ZenMuxError, load_secrets
DEFAULT_MODEL = "anthropic/claude-haiku-4.5"
PROMPT_FILE = Path(__file__).parent / "prompts" / "glossary_system.txt"
def resolve_project(arg: str) -> Path:
p = Path(arg)
if p.is_dir():
return p
cand = Path.cwd() / "projects" / arg
if cand.is_dir():
return cand
raise SystemExit(f"project not found: {arg}")
def load_glossary(path: Path) -> dict:
if not path.exists():
return {}
return json.loads(path.read_text(encoding="utf-8"))
def save_glossary(path: Path, glossary: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(glossary, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
def migrate_legacy_entry(value) -> dict:
"""旧版 glossary 里 value 是字符串;升级为对象格式。"""
if isinstance(value, str):
return {"zh": value}
if isinstance(value, dict):
return value
return {"zh": str(value)}
def build_query(term: str, domain: str) -> str:
if domain:
return f"{term} {domain} 中文名"
return term
def parse_json_line(text: str) -> dict:
"""模型返回的单行 JSON。容忍前后额外字符。"""
s = text.strip()
l = s.find("{")
r = s.rfind("}")
if l == -1 or r == -1:
raise ValueError(f"no JSON object: {text[:200]}")
obj = json.loads(s[l : r + 1])
if not isinstance(obj, dict):
raise ValueError("top-level JSON not object")
return obj
def build_user_prompt(term: str, domain: str, current_zh: str, hits: list) -> str:
hits_text = "\n\n".join(
f"[{i+1}] {h.title}\n URL: {h.url}\n {h.snippet[:500]}"
for i, h in enumerate(hits[:5])
)
if not hits_text:
hits_text = "(无搜索结果)"
return (
f"term: {term}\n"
f"domain: {domain or '(未指定)'}\n"
f"current_zh: {current_zh or '(空)'}\n\n"
f"search_hits:\n{hits_text}\n"
)
def verify_term(
term: str,
current_zh: str,
domain: str,
search_client: SearchClient,
llm_client: ZenMuxClient,
*,
model: str,
system_prompt: str,
) -> dict:
try:
hits = search_client.search(
build_query(term, domain),
num_results=4,
)
except SearchError as e:
return {
"zh": current_zh or "",
"en_full": term,
"confidence": "low",
"issue": f"搜索失败:{e}",
"sources": [],
}
user = build_user_prompt(term, domain, current_zh, hits)
try:
raw = llm_client.chat_complete(
model=model,
system=system_prompt,
user=user,
temperature=0.1,
max_tokens=800,
tag=f"glossary:{term[:30]}",
)
obj = parse_json_line(raw)
except (ZenMuxError, ValueError) as e:
return {
"zh": current_zh or "",
"en_full": term,
"confidence": "low",
"issue": f"LLM 核查失败:{e}",
"sources": [h.url for h in hits[:3]],
}
# 规范化输出
return {
"zh": str(obj.get("zh", current_zh or "")),
"en_full": str(obj.get("en_full", term)),
"confidence": str(obj.get("confidence", "low")),
"issue": str(obj.get("issue", "")),
"sources": [h.url for h in hits[:3]],
}
def main() -> int:
parser = argparse.ArgumentParser(description="双语术语表事实核查(Haiku + Exa)")
parser.add_argument("project", help="项目 slug 或完整路径")
parser.add_argument(
"--input",
default="phase4/glossary.json",
help="初版术语表路径(相对项目根)",
)
parser.add_argument(
"--output",
default="phase4/glossary.json",
help="输出路径(默认原地覆盖)",
)
parser.add_argument("--model", default=DEFAULT_MODEL, help="LLM 模型")
parser.add_argument(
"--workers", type=int, default=6, help="并发度(默认 6,Exa 限速 5 QPS)"
)
parser.add_argument(
"--force", action="store_true",
help="忽略已核查状态,全部重跑",
)
parser.add_argument(
"--only",
default=None,
help="只核查指定术语(逗号分隔,大小写敏感)",
)
parser.add_argument(
"--extra",
default=None,
help="每行一个英文术语的文本文件,补充进术语表一起核查",
)
parser.add_argument(
"--domain",
default=None,
help="术语领域提示(默认自动读 manifest.topic",
)
args = parser.parse_args()
load_secrets()
project_root = resolve_project(args.project)
input_path = project_root / args.input
output_path = project_root / args.output
manifest_path = project_root / "manifest.json"
domain = args.domain
if not domain and manifest_path.exists():
m = json.loads(manifest_path.read_text(encoding="utf-8"))
domain = m.get("topic") or m.get("report_title") or ""
glossary = load_glossary(input_path)
# migration
for k, v in list(glossary.items()):
glossary[k] = migrate_legacy_entry(v)
if args.extra:
extra_path = Path(args.extra)
if not extra_path.exists():
raise SystemExit(f"--extra 文件不存在:{extra_path}")
for line in extra_path.read_text(encoding="utf-8").splitlines():
term = line.strip()
if term and term not in glossary:
glossary[term] = {"zh": ""}
only_terms: set[str] | None = None
if args.only:
only_terms = {t.strip() for t in args.only.split(",") if t.strip()}
system_prompt = PROMPT_FILE.read_text(encoding="utf-8")
# 筛选需要核查的
todo: list[str] = []
for term, entry in glossary.items():
if only_terms is not None and term not in only_terms:
continue
if not args.force and entry.get("verified_at"):
continue
todo.append(term)
print(f"Project: {project_root.name}")
print(f"Domain hint: {domain or '(none)'}")
print(f"Glossary size: {len(glossary)} | to verify: {len(todo)} | workers: {args.workers}")
print(f"Model: {args.model}")
if not todo:
print(" 没有需要核查的条目(使用 --force 强制重跑)")
save_glossary(output_path, glossary)
return 0
today = dt.date.today().isoformat()
logs_dir = project_root / "phase4" / "logs"
log_file = logs_dir / "glossary.jsonl"
start = time.time()
done_count = 0
failed: list[str] = []
# 并发执行:SearchClient/ZenMuxClient 都是 thread-safehttpx.Client 支持)
with SearchClient() as search_client, ZenMuxClient(log_file=log_file) as llm_client:
def worker(term: str) -> tuple[str, dict]:
current_zh = glossary.get(term, {}).get("zh", "")
result = verify_term(
term,
current_zh,
domain or "",
search_client,
llm_client,
model=args.model,
system_prompt=system_prompt,
)
return term, result
with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as pool:
futures = {pool.submit(worker, term): term for term in todo}
for fut in concurrent.futures.as_completed(futures):
term = futures[fut]
try:
_term, result = fut.result()
except Exception as e:
print(f" [FAIL] {term}: {e}")
failed.append(term)
continue
# 合并到 glossary
old = glossary.get(term, {})
old.update(result)
old["verified_at"] = today
glossary[term] = old
done_count += 1
conf = result.get("confidence", "?")
issue = result.get("issue", "")
zh = result.get("zh") or "(保留英文)"
marker = {"high": "", "medium": "~", "low": "?"}.get(conf, " ")
issue_str = f"{issue[:80]}" if issue else ""
print(f" [{marker}] {term:<35}{zh}{issue_str}")
# 阶段性存盘,避免中途挂掉丢数据
if done_count % 10 == 0:
save_glossary(output_path, glossary)
save_glossary(output_path, glossary)
elapsed = time.time() - start
print(f"\n完成:{done_count}/{len(todo)}(失败 {len(failed)},耗时 {elapsed:.1f}s")
print(f"术语表:{output_path.relative_to(project_root)}")
print(llm_client.usage.summary())
# 高警示项汇总
issues = [
(k, v) for k, v in glossary.items()
if v.get("issue") and v.get("confidence") != "high"
]
if issues:
print(f"\n⚠ 低置信度或带问题的条目({len(issues)} 条):")
for k, v in issues[:15]:
print(f" - {k:<30} [{v.get('confidence','?')}] {v.get('issue','')[:100]}")
if len(issues) > 15:
print(f" … 还有 {len(issues) - 15} 条,见 {output_path.name}")
return 0
if __name__ == "__main__":
sys.exit(main())
+238
View File
@@ -0,0 +1,238 @@
#!/usr/bin/env python3
"""Phase 4 成稿阶段:统一入口。
从 final_zh_polished.md(或指定的 Markdown+ manifest.json 生成:
- <title>.pdf ReportLab 出中文 PDF
- <title>.docx Pandoc 出 DOCX
- <title>-en.pdf 如果存在 final_en.md 也一并出英文版(可选)
文件名来自 manifest.report_title(去掉非法字符),不再用 "final.pdf" 这种通用名。
用法:
uv run python scripts/build_report.py <project_slug>
# 只生成 PDF
uv run python scripts/build_report.py <project_slug> --no-docx
# 从自定义 md 生成:
uv run python scripts/build_report.py <project_slug> --input phase4/final_zh.md
环境依赖:
- reportlab, pypandoc, 思源字体(bash .opencode/templates/fonts/download-fonts.sh
- pandoc 可执行文件在 PATH
"""
from __future__ import annotations
import argparse
import json
import re
import shutil
import subprocess
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from scripts.lib.zenmux_client import load_secrets # noqa: F401 (为一致性)
REPO_ROOT = Path(__file__).resolve().parent.parent
PDF_TEMPLATE = REPO_ROOT / ".opencode" / "templates" / "report-template.py"
DEFAULT_FONTS_DIR = REPO_ROOT / ".opencode" / "templates" / "fonts"
def resolve_project(arg: str) -> Path:
p = Path(arg)
if p.is_dir():
return p.resolve()
cand = REPO_ROOT / "projects" / arg
if cand.is_dir():
return cand.resolve()
raise SystemExit(f"project not found: {arg}")
# 文件系统对文件名的常见限制:Windows 更严格,按最小公倍数来
_FS_ILLEGAL_RE = re.compile(r'[\\/:*?"<>|\r\n\t]+')
_WHITESPACE_RE = re.compile(r"\s+")
def sanitize_filename(name: str, max_len: int = 120) -> str:
"""把报告标题变成可跨 OS 使用的文件名。"""
if not name:
return "report"
# 去掉非法字符
cleaned = _FS_ILLEGAL_RE.sub(" ", name)
# 合并空白
cleaned = _WHITESPACE_RE.sub(" ", cleaned).strip()
# 首尾 . 空格 . Windows 要求)
cleaned = cleaned.strip(". ").strip()
if len(cleaned) > max_len:
cleaned = cleaned[:max_len].rstrip()
return cleaned or "report"
def load_manifest(path: Path) -> dict:
return json.loads(path.read_text(encoding="utf-8"))
def determine_input_md(project_root: Path, preferred: str | None) -> Path:
"""决定用哪个 Markdown 出稿。
优先级:--input 指定 > final_zh_polished.md > final_zh.md > final_en.md
"""
if preferred:
p = project_root / preferred
if not p.exists():
raise SystemExit(f"指定的 --input 不存在:{p}")
return p
for candidate in (
"phase4/final_zh_polished.md",
"phase4/final_zh.md",
"phase4/final_en.md",
):
p = project_root / candidate
if p.exists():
return p
raise SystemExit(
f"找不到任何 Markdown 源。项目根:{project_root}\n"
"先跑 translate.py / polish.py 或使用 --input 指定路径。"
)
def build_pdf(
md_path: Path,
manifest_path: Path,
output_pdf: Path,
fonts_dir: Path,
sources_path: Path | None,
) -> None:
"""调用 report-template.py 生成 PDF。"""
cmd = [
sys.executable,
str(PDF_TEMPLATE),
"--input", str(md_path),
"--manifest", str(manifest_path),
"--output", str(output_pdf),
"--fonts-dir", str(fonts_dir),
]
if sources_path and sources_path.exists():
cmd += ["--sources", str(sources_path)]
print(f"\n→ 生成 PDF{output_pdf.name}")
result = subprocess.run(cmd, check=False)
if result.returncode != 0:
raise SystemExit(f"PDF 生成失败,返回码 {result.returncode}")
def build_docx(md_path: Path, output_docx: Path, title: str) -> None:
"""用 pandoc 生成 DOCX。"""
if not shutil.which("pandoc"):
print(f" ⚠ pandoc 不在 PATH,跳过 DOCX 生成", file=sys.stderr)
return
print(f"\n→ 生成 DOCX{output_docx.name}")
# 关闭 tex_math_dollars/tex_math_single_backslash 防止文中 "$100" "$10^6" 被当数学公式
cmd = [
"pandoc",
str(md_path),
"-o", str(output_docx),
"--from=markdown-tex_math_dollars-tex_math_single_backslash-raw_tex",
"--to=docx",
"--standalone",
"-M", f"title={title}",
"--wrap=preserve",
]
# reference-doc 如果存在就用
ref_doc = REPO_ROOT / ".opencode" / "templates" / "reference.docx"
if ref_doc.exists():
cmd += ["--reference-doc", str(ref_doc)]
result = subprocess.run(cmd, check=False)
if result.returncode != 0:
print(f" ⚠ DOCX 生成失败(返回码 {result.returncode}),但不阻断流程", file=sys.stderr)
def main() -> int:
parser = argparse.ArgumentParser(description="Phase 4 成稿(PDF + DOCX")
parser.add_argument("project", help="项目 slug 或完整路径")
parser.add_argument(
"--input",
default=None,
help="Markdown 源(默认自动寻找 phase4/final_zh_polished.md",
)
parser.add_argument(
"--output-dir",
default="phase4",
help="输出目录(相对项目根,默认 phase4)",
)
parser.add_argument(
"--fonts-dir",
default=str(DEFAULT_FONTS_DIR),
help="字体目录",
)
parser.add_argument(
"--sources",
default=None,
help="sources.jsonl 路径(默认 phase2/sources.jsonl",
)
parser.add_argument("--no-docx", action="store_true", help="跳过 DOCX 生成")
parser.add_argument("--no-pdf", action="store_true", help="跳过 PDF 生成")
parser.add_argument(
"--basename",
default=None,
help="文件名 stem(不带扩展名),默认从 manifest.report_title 生成",
)
args = parser.parse_args()
project_root = resolve_project(args.project)
manifest_path = project_root / "manifest.json"
if not manifest_path.exists():
raise SystemExit(f"manifest.json not found: {manifest_path}")
manifest = load_manifest(manifest_path)
md_path = determine_input_md(project_root, args.input)
title = manifest.get("report_title") or manifest.get("topic") or "Deep Research Report"
basename = args.basename or sanitize_filename(title)
output_dir = project_root / args.output_dir
output_dir.mkdir(parents=True, exist_ok=True)
pdf_path = output_dir / f"{basename}.pdf"
docx_path = output_dir / f"{basename}.docx"
fonts_dir = Path(args.fonts_dir)
if not fonts_dir.is_absolute():
fonts_dir = REPO_ROOT / fonts_dir
sources_path: Path | None = None
if args.sources:
sources_path = Path(args.sources)
else:
default_src = project_root / "phase2" / "sources.jsonl"
if default_src.exists():
sources_path = default_src
print("========== Phase 4 成稿 ==========")
print(f"项目: {project_root.name}")
print(f"Markdown {md_path.relative_to(project_root)}")
print(f"标题: {title}")
print(f"输出名: {basename}")
print(f"字体目录: {fonts_dir}")
print(f"Sources {sources_path if sources_path else '(缺失)'}")
if not args.no_pdf:
build_pdf(md_path, manifest_path, pdf_path, fonts_dir, sources_path)
if not args.no_docx:
build_docx(md_path, docx_path, title)
print("\n========== 完成 ==========")
if pdf_path.exists():
print(f" PDF: {pdf_path.relative_to(project_root)} ({pdf_path.stat().st_size // 1024} KB)")
if docx_path.exists():
print(f" DOCX: {docx_path.relative_to(project_root)} ({docx_path.stat().st_size // 1024} KB)")
return 0
if __name__ == "__main__":
sys.exit(main())
+177
View File
@@ -0,0 +1,177 @@
"""通用搜索客户端(Exa 优先,Tavily fallback)。
build_glossary.py 这类术语核查场景服务
关键设计
- `trust_env=False` 绕开系统 socks 代理Clash on macOS socks5 httpx TLS EOF
- Exa 优先LinkedIn / 官网 / 百度百科返回质量最高
- 遇到配额问题自动降级到 Tavily 或返回 empty
- 不做深度 crawl只要摘要
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Any
import httpx
@dataclass
class SearchHit:
title: str
url: str
snippet: str
class SearchError(RuntimeError):
pass
class ExaClient:
def __init__(self, api_key: str | None = None, timeout: float = 30.0) -> None:
self.api_key = api_key or os.environ.get("EXA_API_KEY")
if not self.api_key:
raise SearchError("EXA_API_KEY not set")
# trust_env=False 关键:不吃系统代理,避免 TLS EOF
self._client = httpx.Client(trust_env=False, timeout=timeout)
def close(self) -> None:
self._client.close()
def __enter__(self) -> "ExaClient":
return self
def __exit__(self, *_args: Any) -> None:
self.close()
def search(self, query: str, *, num_results: int = 5) -> list[SearchHit]:
body = {
"query": query,
"numResults": num_results,
"type": "auto",
"contents": {"text": {"maxCharacters": 800}},
}
r = self._client.post(
"https://api.exa.ai/search",
json=body,
headers={"x-api-key": self.api_key, "Content-Type": "application/json"},
)
if r.status_code != 200:
raise SearchError(f"Exa HTTP {r.status_code}: {r.text[:200]}")
data = r.json()
out: list[SearchHit] = []
for item in data.get("results", [])[:num_results]:
out.append(
SearchHit(
title=(item.get("title") or "")[:200],
url=item.get("url") or "",
snippet=(item.get("text") or item.get("snippet") or "")[:600],
)
)
return out
class TavilyClient:
def __init__(self, api_key: str | None = None, timeout: float = 30.0) -> None:
self.api_key = api_key or os.environ.get("TAVILY_API_KEY")
if not self.api_key:
raise SearchError("TAVILY_API_KEY not set")
self._client = httpx.Client(trust_env=False, timeout=timeout)
def close(self) -> None:
self._client.close()
def __enter__(self) -> "TavilyClient":
return self
def __exit__(self, *_args: Any) -> None:
self.close()
def search(self, query: str, *, num_results: int = 5) -> list[SearchHit]:
body = {
"api_key": self.api_key,
"query": query,
"search_depth": "basic",
"max_results": num_results,
"include_answer": False,
"include_raw_content": False,
}
r = self._client.post("https://api.tavily.com/search", json=body)
if r.status_code != 200:
raise SearchError(f"Tavily HTTP {r.status_code}: {r.text[:200]}")
data = r.json()
out: list[SearchHit] = []
for item in data.get("results", [])[:num_results]:
out.append(
SearchHit(
title=(item.get("title") or "")[:200],
url=item.get("url") or "",
snippet=(item.get("content") or "")[:600],
)
)
return out
class SearchClient:
"""统一搜索门面:先用 Exa,失败/配额问题降级 Tavily。"""
def __init__(self) -> None:
self._exa: ExaClient | None = None
self._tavily: TavilyClient | None = None
try:
self._exa = ExaClient()
except SearchError:
pass
try:
self._tavily = TavilyClient()
except SearchError:
pass
if not (self._exa or self._tavily):
raise SearchError(
"neither EXA_API_KEY nor TAVILY_API_KEY available"
)
def close(self) -> None:
if self._exa:
self._exa.close()
if self._tavily:
self._tavily.close()
def __enter__(self) -> "SearchClient":
return self
def __exit__(self, *_args: Any) -> None:
self.close()
def search(self, query: str, *, num_results: int = 5) -> list[SearchHit]:
# 优先 Exa
if self._exa:
try:
return self._exa.search(query, num_results=num_results)
except SearchError as e:
msg = str(e).lower()
if "exceed" in msg or "quota" in msg or "429" in msg or "402" in msg:
# 降级
pass
else:
# 其它错误继续往下试
pass
if self._tavily:
try:
return self._tavily.search(query, num_results=num_results)
except SearchError:
pass
return []
if __name__ == "__main__":
from scripts.lib.zenmux_client import load_secrets
load_secrets()
with SearchClient() as c:
hits = c.search("Mabwell 迈威生物 biopharmaceutical", num_results=3)
for i, h in enumerate(hits, 1):
print(f"[{i}] {h.title[:80]}")
print(f" {h.url}")
print(f" {h.snippet[:160]}")
+54
View File
@@ -0,0 +1,54 @@
你是一名生物医药行业的资深双语术语编辑。现在要根据网络搜索片段,核查并确定一个英文术语/专有名词的"中文规范译名"、"英文全称"与"可信度"。
## 你将收到
- `term`:待核查的英文术语(可能是公司名、缩写、化学品、平台名、药物通用名等)
- `domain`:研究领域(例如 "生物医药双靶点 RNAi 药物"),作为消歧背景
- `current_zh`:系统已有的中文译名(可能来自上一步翻译,也可能为空)
- `search_hits`35 条 web 搜索结果(title + url + snippet
## 你要判断
1. **中文规范译名**`zh`):
- 如果搜索结果里有权威的中文表达(公司官网、维基、百度百科、权威医药媒体),以它为准
- 如果 `current_zh` 已经正确,沿用它,避免无谓变更
- 如果无中文通用译名(小众学术术语、新兴化合物),保留英文原文,`zh` 字段置空字符串
- 对公司/机构名:必须使用工商注册的正式中文名(例如 "Mabwell → 迈威生物"、"Sirnaomics → 圣诺生物"
- 对药物:优先 INN 通用名(例如 "inclisiran → 英克司兰"
- 对缩写/技术术语:使用业内通行译名(例如 "GalNAc → N-乙酰半乳糖胺""RNAi → RNA 干扰"
2. **英文全称**`en_full`):
- 如果术语是缩写(≤ 8 字符全大写或混合大小写),给出英文全称(例如 "ASGPR → Asialoglycoprotein Receptor"
- 如果术语本身就是全称,填相同的字符串或其最规范的写法
- 如果术语是公司名,填其英文法人全称(例如 "Mabwell → Mabwell (Shanghai) Bioscience Co., Ltd."
3. **可信度**`confidence`):
- `high`:至少两条独立、高质量信源(官网/监管机构/权威媒体)一致支持
- `medium`:一条高质量信源支持,或多条一般信源一致支持
- `low`:只能推断或无法确认,建议人工复核
4. **问题/警示**`issue`):
- 如果 `current_zh` 明显错误(例如把 "Mabwell" 译成 "Maywavee"),在 `issue` 中指出错在哪里
- 如果术语有多种译法争议,在 `issue` 中简述
- 如果搜索结果完全不相关(错别字、生僻词),在 `issue` 中说"搜索无有效结果"
- 无异常则留空字符串
## 输出格式(严格)
输出**一行 JSON**,不要用代码围栏,不要加任何解释。
```
{"zh": "迈威生物", "en_full": "Mabwell (Shanghai) Bioscience Co., Ltd.", "confidence": "high", "issue": ""}
```
如果 `current_zh` 错了:
```
{"zh": "迈威生物", "en_full": "Mabwell (Shanghai) Bioscience Co., Ltd.", "confidence": "high", "issue": "current_zh 'Maywavee' 为拼写错误,正确为 Mabwell → 迈威生物"}
```
如果无需中文译名(保留英文):
```
{"zh": "", "en_full": "Phosphoramidite", "confidence": "high", "issue": ""}
```