From 40be75fc35e2f66f4b72267198d5d4227384c8a3 Mon Sep 17 00:00:00 2001 From: kelriclink Date: Fri, 3 Jul 2026 00:50:49 +0800 Subject: [PATCH] feat: add tool-driven RAG agents --- README.md | 69 +- docs/PROJECT_SPEC.md | 36 +- src/videoroll/ai/client.py | 11 +- .../apps/subtitle_service/embeddings.py | 1 + src/videoroll/apps/subtitle_service/main.py | 2 + src/videoroll/apps/subtitle_service/rag.py | 894 +++++++++++++++--- .../apps/subtitle_service/schemas.py | 3 + .../translate_settings_store.py | 7 + src/videoroll/db/auto_migrate.py | 17 + src/web/src/components/ui.tsx | 10 +- src/web/src/pages/DashboardPage.tsx | 75 +- src/web/src/pages/KnowledgeBasePage.tsx | 2 +- src/web/src/pages/SettingsTranslatePage.tsx | 9 + tests/test_translation_rag.py | 214 ++++- 14 files changed, 1175 insertions(+), 175 deletions(-) diff --git a/README.md b/README.md index aec7803..f784389 100644 --- a/README.md +++ b/README.md @@ -2,16 +2,19 @@ 模块化视频处理流水线：内容接入 -> 字幕/翻译 -> 压制/封装 -> 可选投稿到哔哩哔哩。 +当前重点是**带 RAG Agent 的视频字幕翻译**：在翻译每个字幕 block 之前，系统会先判断哪些词真的需要知识库或联网查证，再把可信术语和背景知识注入翻译 prompt。目标不是把所有词都入库，而是解决 YouTube、动漫、游戏实况、技术讲解这类视频里“模型缺上下文”的问题。 + 详细规格见 [docs/PROJECT_SPEC.md](docs/PROJECT_SPEC.md)。 ## 显著功能 -- **RAG 翻译 Agent**：字幕翻译前先扫描当前 block 和前文 summary，判断哪些术语真的需要查知识库或联网检索，减少把基础词、局部变量写入长期知识库。 +- **主 Agent + 子 Agent 翻译 RAG**：字幕翻译前由主 Agent 扫描当前 block 和前文 summary，决定哪些术语需要研究；每个术语由独立子 Agent 维护 observation/evidence 上下文并选择工具。 - **PostgreSQL + pgvector 知识库**：术语、译法、来源、置信度和使用次数统一入库；支持向量检索、手动新增/删除条目、重建 embedding。 -- **Wikipedia Tool + SearXNG Tool**：Agent 优先查固定 English Wikipedia API，必要时再调用自建 SearXNG，并可读取网页正文作为证据。 -- **并发术语研究 Agent**：可在 `Settings -> Translate` 设置同时研究的术语数量；每个 Agent 都有独立工具调用 trace。 +- **真正的工具调用循环**：子 Agent 可按需调用 `rag_lookup`、`wiki_search`、`search_web`、`fetch_url`、`finish`，而不是固定按 wiki -> search 的流水线机械执行。 +- **Wikipedia Tool + SearXNG Tool**：固定 English Wikipedia API 用于通用百科知识；自建 SearXNG 用于补充网络搜索；搜索摘要不足时可继续读取网页正文。 +- **并发术语研究 Agent**：可在 `Settings -> Translate` 设置并发数和超时；主 Agent 等待子 Agent 返回结构化结果后再合并进当前 block 翻译。 - **保守入库策略**：summarize -> verify -> write_rag；只有 verifier 判断来源支撑、上下文一致且置信度足够时才正式入库，否则标记为 pending、context_only 或 skipped。 -- **独立 embedding 配置**：翻译 API key/base URL 与 embedding API key/base URL 分开；embedding 支持 OpenAI 兼容服务和本地模型。 +- **独立模型配置**：翻译大模型、embedding 模型的 API key/base URL 分开；翻译大模型支持单独设置失败重试次数；embedding 支持 OpenAI 兼容服务和本地模型。 - **可观测 Dashboard**：显示任务状态、CPU/内存/Intel GPU 资源监控，以及 RAG Agent 当前运行状态和历史对话流。 - **完整 Web 配置**：ASR、翻译、RAG、YouTube、Bilibili、自动模式、知识库均可在 Web UI 中配置，减少部署环境变量依赖。 @@ -26,7 +29,7 @@ - Bilibili 投稿支持在 Web UI 中保存 Cookie、测试登录、维护默认投稿模板。 - Web UI 支持浅色 / 暗色模式切换，Dashboard 增加 CPU、内存和 Intel GPU 资源监控。 - 字幕翻译支持真正的 RAG 术语知识库：可使用 PostgreSQL + pgvector 检索术语，支持本地 embedding 模型、手动下载模型、重建向量，以及 Agent 自动发现/搜索/验证/入库术语。 -- RAG Agent 支持固定 Wikipedia Tool、SearXNG 搜索、网页正文读取、独立 verifier、保守入库策略，并可在 Dashboard 查看每个 Agent 的工具调用 trace。 +- RAG Agent 已改为主 Agent / 子 Agent 结构：Dashboard 可查看主 Agent、其下属子 Agent、每一步工具调用、耗时、模型、错误类型和 JSON 输入输出。 ## 架构概览 @@ -134,7 +137,27 @@ S3_USE_SSL=false docker compose -f docker-compose.yml --env-file .env up --build -d ``` -### 5. 首次登录 +### 5. 使用已导出的镜像部署 + +如果你从 `dist/docker-images/videoroll-prod-app-web.tar` 部署： + +```bash +docker load -i dist/docker-images/videoroll-prod-app-web.tar +docker compose -f docker-compose.yml --env-file .env up -d +``` + +该 tar 只包含业务镜像： + +- `videoroll:prod` +- `videoroll-web:prod` + +Redis、MinIO、PostgreSQL 仍按 compose 或你的现有部署提供。导出后可用 sha256 文件校验： + +```bash +sha256sum -c dist/docker-images/videoroll-prod-app-web.tar.sha256 +``` + +### 6. 首次登录 打开 Web UI： @@ -328,17 +351,17 @@ RAG 用于解决视频字幕翻译中的上下文和术语问题，例如游戏 ```text 字幕片段 + 前文翻译 summary - -> RAG gate 判断是否需要检索 - -> 术语发现与噪声过滤 + -> 主 Agent 创建 rag_master run + -> RAG gate 判断哪些术语需要检索 -> 查询 PostgreSQL/pgvector 知识库 - -> 未命中时并发启动术语研究 Agent - -> 优先调用 Wikipedia Tool - -> 必要时调用 SearXNG 搜索 - -> 必要时读取网页正文 - -> LLM 总结候选术语 + -> 未命中时并发启动子 Agent + -> 子 Agent 自己决定调用 rag_lookup / wiki_search / search_web / fetch_url / finish + -> 子 Agent 收集证据并总结候选术语 -> verifier 检查来源和上下文 -> 写入知识库或标记为 context_only / skipped - -> 翻译时注入 rag_context + -> 子 Agent 返回 AgentResearchResult + -> 主 Agent 汇总 term_cards / knowledge_cards / hits + -> 翻译时注入 RagContext ``` 关键行为： @@ -347,8 +370,20 @@ RAG 用于解决视频字幕翻译中的上下文和术语问题，例如游戏 - 常见基础术语（如 `truth table`）可以作为本次翻译提示，但默认不自动污染长期知识库。 - 自动正式入库要求 verifier 通过、上下文一致、有有效外部来源，并达到较高置信度；否则进入待审核或跳过。 - Wikipedia Tool 使用固定 English Wikipedia API，不需要额外配置 wiki 地址；它适合通用百科类专有名词，不替代垂直领域资料库。 -- 可在 `Settings -> Translate` 调整 Agent 并发数和超时时间。并发越高，术语补全越快，但外部 API、LLM 和搜索服务压力也越大。 -- Dashboard 中可以点击 RAG Agent 查看每步工具调用、耗时、模型、错误类型和 JSON 输入输出。 +- SearXNG Tool 使用 base URL 配置，系统自动拼接 `/search?q=...&format=json`，并过滤搜索引擎 About/Preferences 等内部页面。 +- 可在 `Settings -> Translate` 调整 Agent 并发数、Agent 超时、大模型失败重试次数。并发越高，术语补全越快，但外部 API、LLM 和搜索服务压力也越大。 +- Dashboard 中可以点击主 Agent 查看下属子 Agent；点击子 Agent 可查看它自己的工具调用对话流。 + +子 Agent 返回主 Agent 的数据是结构化结果，而不是把整段 trace 直接塞回 prompt： + +```text +AgentResearchResult + - hit: 成功写入或命中的长期知识库条目 + - context_card: 不适合长期入库但当前 block 可用的临时术语提示 + - normalized_term: 去重用的规范化术语 +``` + +主 Agent 会把这些结果合并成 `RagContext`，后续翻译 prompt 只注入精简后的术语卡片和知识卡片，避免长 trace 拖慢翻译。 PostgreSQL 需要启用 pgvector： @@ -368,7 +403,7 @@ https://search.example.com Embedding 支持两种方式： -- OpenAI 兼容 embedding：填写 provider 支持的 embedding 模型名。 +- OpenAI 兼容 embedding：填写 provider 支持的 embedding 模型名、base URL 和独立 API key。 - 本地 embedding：在 `Settings -> Translate` 中选择本地 provider，可手动下载模型到 `data/models`，并选择 CPU / OpenVINO GPU 等设备。 更换 embedding 模型后，已有知识库向量不会静默重建；需要在 `Settings -> Translate` 中点击重建按钮，避免模型切换时旧向量和新向量混用。 diff --git a/docs/PROJECT_SPEC.md b/docs/PROJECT_SPEC.md index 804498c..1875881 100644 --- a/docs/PROJECT_SPEC.md +++ b/docs/PROJECT_SPEC.md @@ -8,34 +8,45 @@ ## 1. 当前显著能力 -当前实现已经从“普通字幕翻译流水线”升级为带可观测 RAG Agent 的视频翻译系统，重点能力如下： +当前实现已经从“普通字幕翻译流水线”升级为带可观测 RAG Agent 的视频翻译系统。重点不是把所有词都查一遍，而是在翻译前判断“哪些词确实需要外部知识”，再把可信知识注入翻译 prompt。 - **字幕翻译 RAG gate**：每个翻译 block 开始前，LLM 先结合当前字幕和前文 summary 判断哪些术语真的需要检索，避免把基础词、局部变量、普通短语都塞进知识库。 - **PostgreSQL + pgvector 知识库**：保存术语、译法、领域、别名、解释、来源、置信度、状态和 embedding，翻译时按向量相似度召回并注入上下文。 -- **单 Agent + 多 Tool 架构**：术语研究 Agent 统一调度 RAG、Wikipedia、SearXNG、网页读取、总结、验证和写库，不一开始拆成多个松散 Agent。 +- **主 Agent + 子 Agent 架构**：每个字幕 block 创建一个 `rag_master` 主 Agent；主 Agent 负责术语 gate、调度并发子 Agent、汇总结果；每个术语研究子 Agent 维护自己的 observation/evidence 上下文。 +- **工具调用循环**：子 Agent 可按需调用 `rag_lookup`、`wiki_search`、`search_web`、`fetch_url`、`finish`。工具选择由 LLM 根据当前 observation 决定，程序只做超时、重试、fallback 和写库安全控制。 - **Wikipedia Tool**：固定使用 English Wikipedia API，适合百科型专有名词和通用背景知识；不用在配置里维护各种不通用的 wiki 地址。 - **SearXNG Tool**：可接入自建 SearXNG，配置填写 base URL，系统自动请求 `/search?q=...&format=json` 并过滤搜索引擎自身页面等无效结果。 - **网页正文读取 Tool**：当搜索摘要不足时，Agent 可以打开候选链接抽取正文，再交给 LLM 总结证据。 - **Verifier 入库保护**：候选术语会经过 verifier 判断来源是否支撑、是否和字幕上下文一致、置信度是否足够；不满足条件时不会污染长期知识库。 -- **并发术语研究**：翻译设置里可调整 RAG Agent 并发数和超时时间，让多个需要查证的术语同时研究，最后合并回当前翻译上下文。 +- **并发术语研究**：翻译设置里可调整 RAG Agent 并发数和超时时间，让多个需要查证的术语同时研究；子 Agent 结束后返回结构化 `AgentResearchResult`，由主 Agent 合并回当前翻译上下文。 - **独立 embedding 配置**：翻译模型和 embedding 模型的 API key/base URL 分开；embedding 支持 OpenAI 兼容接口和本地模型目录。 -- **Web 管理与可观测性**：知识库页面支持查看、新增、删除条目；Dashboard 可查看资源监控、正在运行的 Agent、历史 trace 和每步 JSON 输入输出。 +- **独立大模型重试配置**：翻译大模型请求的失败重试次数可在翻译设置中单独配置，不和 embedding 请求混用。 +- **Web 管理与可观测性**：知识库页面支持查看、新增、删除条目；Dashboard 可查看资源监控、主 Agent/子 Agent 运行树、历史 trace 和每步 JSON 输入输出。 核心翻译流程： ```text 字幕 block + 前文 summary + -> 创建 rag_master 主 Agent -> RAG gate 判断需要检索的术语 - -> 查询 pgvector 知识库 - -> 未命中时并发启动术语研究 Agent - -> Wikipedia / SearXNG / fetch tools 收集证据 + -> 先查询 pgvector 知识库 + -> 未命中时并发启动术语研究子 Agent + -> 子 Agent 按 observation 决定 rag/wiki/search/fetch/finish -> LLM summarize 生成候选译法 -> verifier 校验来源和上下文 -> 写入知识库或标记为 pending/context_only/skipped - -> 将命中知识和本次研究结果注入翻译 prompt + -> 子 Agent 返回 hit/context_card + -> 主 Agent 汇总 RagContext + -> 将精简后的术语卡片和知识卡片注入翻译 prompt -> 输出当前 block 翻译并更新 summary ``` +子 Agent 的 trace 用于调试和审计，不直接塞回翻译 prompt。最终进入翻译 prompt 的是主 Agent 汇总后的结构化上下文： + +- `term_cards`：术语、译法、领域、说明、来源、置信度。 +- `knowledge_cards`：文档型背景知识。 +- `hits`：原始知识库命中，用于记录匹配和使用次数。 + --- ## 2. 总体拆分：三大模块 + 一个编排层（Orchestrator） @@ -169,7 +180,8 @@ - 输入：`segments.json` 或 `subtitle_src.srt` - 输出：`subtitle_zh.srt` - 可插拔 provider：LLM / 翻译引擎。 -- 翻译前可启用 RAG gate：结合当前 block 和前文 summary 发现需要查询的术语，先查 PostgreSQL/pgvector，未命中时由 Agent 调用 Wikipedia、SearXNG 和网页读取工具补充证据。 +- 翻译前可启用 RAG gate：结合当前 block 和前文 summary 发现需要查询的术语；主 Agent 先查 PostgreSQL/pgvector，未命中时并发启动术语研究子 Agent。 +- 子 Agent 通过工具调用循环自行决定下一步：本地 RAG、Wikipedia、SearXNG、网页正文读取或结束研究。 - RAG 结果不会无条件写入长期知识库；必须经过 verifier 校验来源、上下文一致性和置信度。 - 支持手动知识库条目、自动学习条目、context_only 临时提示和 pending 待确认条目。 @@ -347,7 +359,7 @@ ### 8.1 Web 管控台核心功能 - 任务列表：状态筛选、搜索、批量重试/取消 - 任务详情：查看每 stage 的日志与产物；字幕文件在线预览/编辑；投稿 meta 编辑；审核通过/驳回 -- Dashboard：任务状态、队列概览、CPU/内存/Intel GPU 资源监控、RAG Agent 运行状态和历史 trace +- Dashboard：任务状态、队列概览、CPU/内存/Intel GPU 资源监控、RAG 主 Agent/子 Agent 运行树和历史 trace - 知识库管理：查看、新增、删除 RAG 条目；触发 embedding 重建 - 模板中心：标题/简介/标签/分区/动态文案模板，支持变量（如 `{source_title}`、`{publish_date}`） - 账号与凭据：B 站凭据录入与轮换提醒；权限控制 @@ -383,8 +395,8 @@ **M5：RAG Agent 翻译增强** - PostgreSQL + pgvector 知识库 -- RAG gate + 术语噪声过滤 -- Wikipedia / SearXNG / fetch tools +- 主 Agent gate + 术语噪声过滤 +- 子 Agent 工具循环：RAG / Wikipedia / SearXNG / fetch URL - verifier 保守入库 - Dashboard trace 与知识库管理页面 diff --git a/src/videoroll/ai/client.py b/src/videoroll/ai/client.py index b3aa6f8..cae1768 100644 --- a/src/videoroll/ai/client.py +++ b/src/videoroll/ai/client.py @@ -21,6 +21,8 @@ class OpenAIChatConfig: model: str temperature: float = 0.2 timeout_seconds: float = 60.0 + max_retries: int = 3 + embedding_dimensions: int | None = None def openai_chat_config_from_settings(settings: Mapping[str, Any]) -> OpenAIChatConfig: @@ -30,6 +32,7 @@ def openai_chat_config_from_settings(settings: Mapping[str, Any]) -> OpenAIChatC model=str(settings.get("openai_model") or "").strip(), temperature=float(settings.get("openai_temperature") or 0.2), timeout_seconds=float(settings.get("openai_timeout_seconds") or 60.0), + max_retries=max(1, min(10, int(settings.get("openai_max_retries") or 3))), ) @@ -180,7 +183,7 @@ def request_openai_json_object( user_prompt: str, format_retry_notice: str = "注意：上一次输出不符合 JSON/结构要求，请严格按 JSON 输出。", format_retries: int = 2, - network_retries: int = 3, + network_retries: int | None = None, client: httpx.Client | None = None, ) -> dict[str, Any]: if client is not None: @@ -191,7 +194,7 @@ def request_openai_json_object( user_prompt=user_prompt, format_retry_notice=format_retry_notice, format_retries=format_retries, - network_retries=network_retries, + network_retries=max(1, int(network_retries if network_retries is not None else config.max_retries)), ) with create_openai_http_client(config.timeout_seconds) as owned_client: @@ -202,7 +205,7 @@ def request_openai_json_object( user_prompt=user_prompt, format_retry_notice=format_retry_notice, format_retries=format_retries, - network_retries=network_retries, + network_retries=max(1, int(network_retries if network_retries is not None else config.max_retries)), ) @@ -223,6 +226,8 @@ def request_openai_embedding( url = build_openai_embeddings_url(config.base_url) headers = {"Authorization": f"Bearer {config.api_key}"} req = {"model": config.model, "input": source} + if config.embedding_dimensions is not None and config.embedding_dimensions > 0: + req["dimensions"] = int(config.embedding_dimensions) attempts_network = max(1, int(network_retries)) last_err: Exception | None = None diff --git a/src/videoroll/apps/subtitle_service/embeddings.py b/src/videoroll/apps/subtitle_service/embeddings.py index 32bb3c8..9b8c4f1 100644 --- a/src/videoroll/apps/subtitle_service/embeddings.py +++ b/src/videoroll/apps/subtitle_service/embeddings.py @@ -270,5 +270,6 @@ def embedding_settings_from_translate_settings(settings: dict[str, Any]) -> Embe model=str(settings.get("rag_embedding_model") or "text-embedding-3-small").strip() or "text-embedding-3-small", temperature=0.0, timeout_seconds=embedding_timeout, + embedding_dimensions=max(1, min(4096, int(settings.get("rag_embedding_dimensions") or 1536))), ), ) diff --git a/src/videoroll/apps/subtitle_service/main.py b/src/videoroll/apps/subtitle_service/main.py index e88cf9a..fd6663c 100644 --- a/src/videoroll/apps/subtitle_service/main.py +++ b/src/videoroll/apps/subtitle_service/main.py @@ -336,6 +336,7 @@ def get_translate_settings_view( openai_model=cfg["openai_model"], openai_temperature=cfg["openai_temperature"], openai_timeout_seconds=cfg["openai_timeout_seconds"], + openai_max_retries=cfg["openai_max_retries"], rag_enabled=cfg["rag_enabled"], rag_top_k=cfg["rag_top_k"], rag_min_score=cfg["rag_min_score"], @@ -377,6 +378,7 @@ def put_translate_settings_view( openai_model=cfg["openai_model"], openai_temperature=cfg["openai_temperature"], openai_timeout_seconds=cfg["openai_timeout_seconds"], + openai_max_retries=cfg["openai_max_retries"], rag_enabled=cfg["rag_enabled"], rag_top_k=cfg["rag_top_k"], rag_min_score=cfg["rag_min_score"], diff --git a/src/videoroll/apps/subtitle_service/rag.py b/src/videoroll/apps/subtitle_service/rag.py index a0e42ef..0455562 100644 --- a/src/videoroll/apps/subtitle_service/rag.py +++ b/src/videoroll/apps/subtitle_service/rag.py @@ -562,6 +562,8 @@ def _start_agent_run( domain: str, target_lang: str, query: str, + agent_type: str = "rag_term_research", + parent_agent_run_id: str | None = None, task_id: str | None = None, subtitle_job_id: str | None = None, ) -> str: @@ -571,17 +573,19 @@ def _start_agent_run( """ INSERT INTO translation_agent_runs ( id, agent_type, status, term, normalized_term, domain, target_lang, - task_id, subtitle_job_id, query, steps, result, started_at, updated_at + task_id, subtitle_job_id, query, steps, result, parent_agent_run_id, + started_at, updated_at ) VALUES ( - CAST(:id AS uuid), 'rag_term_research', 'running', :term, :normalized_term, + CAST(:id AS uuid), :agent_type, 'running', :term, :normalized_term, :domain, :target_lang, CAST(:task_id AS uuid), CAST(:subtitle_job_id AS uuid), - :query, '[]'::jsonb, '{}'::jsonb, now(), now() + :query, '[]'::jsonb, '{}'::jsonb, CAST(:parent_agent_run_id AS uuid), now(), now() ) """ ), { "id": run_id, + "agent_type": str(agent_type or "rag_term_research").strip()[:64] or "rag_term_research", "term": str(term or "").strip(), "normalized_term": normalize_term(term), "domain": str(domain or "").strip(), @@ -589,6 +593,7 @@ def _start_agent_run( "task_id": task_id, "subtitle_job_id": subtitle_job_id, "query": str(query or "").strip(), + "parent_agent_run_id": parent_agent_run_id, }, ) db.commit() @@ -1588,6 +1593,311 @@ def fetch_wikipedia_evidence( return [] +def fetch_url_evidence( + *, + url: str, + title: str = "", + timeout_seconds: float = 20.0, + db: Session | None = None, + agent_run_id: str | None = None, +) -> dict[str, Any] | None: + clean_url = str(url or "").strip() + if not clean_url or not _is_fetchable_url(clean_url): + return None + started = time.perf_counter() + try: + with httpx.Client( + timeout=timeout_seconds, + follow_redirects=True, + headers={"User-Agent": "VideoRoll-RAG-Agent/1.0", "Accept": "text/html, text/plain;q=0.9,*/*;q=0.8"}, + ) as client: + resp = client.get(clean_url) + resp.raise_for_status() + ctype = resp.headers.get("content-type", "").lower() + body = resp.text[:500_000] + if "html" in ctype or " dict[str, Any]: + compact_observations = observations[-8:] + compact_evidence = [ + { + "title": str(item.get("title") or "")[:180], + "url": str(item.get("url") or "")[:500], + "snippet": str(item.get("snippet") or "")[:500], + "has_content": bool(str(item.get("content") or "").strip()), + "tool": str(item.get("tool") or ""), + } + for item in evidence[-8:] + if isinstance(item, dict) + ] + data = request_openai_json_object( + config=config, + system_prompt="You are a tool-using translation research sub-agent. Return ONLY valid JSON.", + user_prompt=( + "你是一个字幕翻译术语研究子 Agent。你需要像 coding agent 一样根据已有 observation 自己决定下一步调用哪个工具。\n" + "目标：找到术语在当前字幕上下文中最贴切的含义和中文译法，并收集足够证据。\n\n" + "可用工具：\n" + "- rag_lookup: 查询本地知识库是否已有该术语。输入 term。\n" + "- wiki_search: 查询 English Wikipedia。输入 query。\n" + "- search_web: 调用配置的 SearXNG 搜索。输入 query。\n" + "- fetch_url: 打开一个 URL 抽取正文。输入 url。\n" + "- finish: 认为证据足够或无需继续。\n\n" + "决策要求：\n" + "- 不要一开始机械调用所有工具；根据 observation 判断下一步。\n" + "- Wikipedia 不足、无中文译名支撑、或上下文不一致时，应继续 search_web 或 fetch_url。\n" + "- 如果本地知识库已命中，通常 finish。\n" + "- 如果证据明显和字幕无关，也可以 finish 并说明无法入库。\n" + "- 只输出 JSON，不要解释性文本。\n\n" + f"step_no: {step_no}\n" + f"术语: {term}\n" + f"目标语言: {target_lang or 'zh'}\n" + f"领域提示: {domain_hint or '未知'}\n" + f"可用工具: {json.dumps(available_tools, ensure_ascii=False)}\n\n" + f"字幕上下文:\n{context[:2600]}\n\n" + f"observations JSON:\n{json.dumps(compact_observations, ensure_ascii=False)[:5000]}\n\n" + f"evidence JSON:\n{json.dumps(compact_evidence, ensure_ascii=False)[:5000]}\n\n" + '输出 JSON：{"action":"rag_lookup|wiki_search|search_web|fetch_url|finish",' + '"query":"","url":"","reason":"","final_answer_ready":false}' + ), + client=client, + ) + action = str(data.get("action") or "").strip() + if action not in {"rag_lookup", "wiki_search", "search_web", "fetch_url", "finish"}: + action = "finish" + return { + "action": action, + "query": str(data.get("query") or "").strip()[:240], + "url": str(data.get("url") or "").strip()[:1000], + "reason": str(data.get("reason") or "").strip()[:1000], + "final_answer_ready": _json_bool(data.get("final_answer_ready")), + } + + +def _dedupe_evidence(items: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + seen: set[str] = set() + for item in items: + if not isinstance(item, dict): + continue + url = str(item.get("url") or "").strip() + title = str(item.get("title") or "").strip() + key = url or title.lower() + if not key or key in seen: + continue + seen.add(key) + out.append(item) + return out + + +def _collect_evidence_with_tool_agent( + db: Session, + *, + agent_run_id: str | None, + term: str, + domain_hint: str, + target_lang: str, + rag_settings: RagSettings, + chat_config: OpenAIChatConfig, + llm_context: str, + search_queries: list[str], + max_steps: int = 6, +) -> tuple[list[dict[str, Any]], list[str], int]: + available_tools = ["rag_lookup"] + if rag_settings.wiki_enabled: + available_tools.append("wiki_search") + if rag_settings.search_enabled: + available_tools.append("search_web") + available_tools.append("fetch_url") + available_tools.append("finish") + + observations: list[dict[str, Any]] = [] + evidence: list[dict[str, Any]] = [] + tools_used: list[str] = [] + used_actions: set[str] = set() + rounds = 0 + + for step_no in range(1, max(1, min(10, int(max_steps))) + 1): + rounds = step_no + try: + started = time.perf_counter() + decision = research_agent_next_action_openai( + term=term, + context=llm_context, + target_lang=target_lang, + domain_hint=domain_hint, + available_tools=available_tools, + observations=observations, + evidence=evidence, + config=chat_config, + step_no=step_no, + ) + _append_llm_step( + db, + agent_run_id, + action="agent_tool_decision", + config=chat_config, + input_value={ + "term": term, + "step_no": step_no, + "available_tools": available_tools, + "observation_count": len(observations), + "evidence_count": len(evidence), + }, + output_value=decision, + duration_ms=_duration_ms(started), + ) + except Exception as e: + decision = {"action": "finish", "query": "", "url": "", "reason": f"tool decision failed: {e}", "final_answer_ready": False} + _append_llm_step( + db, + agent_run_id, + action="agent_tool_decision_failed", + config=chat_config, + error=str(e)[:300], + error_type=type(e).__name__, + ) + + action = str(decision.get("action") or "finish") + reason = str(decision.get("reason") or "") + if action == "finish": + observations.append({"action": "finish", "reason": reason, "evidence_count": len(evidence)}) + _append_agent_step(db, agent_run_id, {"kind": "agent", "action": "agent_finish_decision", "reason": reason, "evidence_count": len(evidence)}) + break + + if action == "rag_lookup": + tools_used.append("rag_lookup") + used_actions.add(action) + norm_value = normalize_term(term) + exists = norm_value in existing_term_norms(db, terms=[term], target_lang=target_lang) + observations.append({"action": "rag_lookup", "term": term, "exists": exists, "normalized_term": norm_value}) + _append_agent_step( + db, + agent_run_id, + {"kind": "tool", "action": "rag_lookup", "tool": "rag_lookup", "term": term, "exists": exists, "ok": True}, + ) + if exists: + break + continue + + query = str(decision.get("query") or "").strip() + if not query: + query = (search_queries[0] if search_queries else " ".join([p for p in [domain_hint, term] if p]).strip()) or term + + if action == "wiki_search" and rag_settings.wiki_enabled: + tools_used.append("wikipedia") + used_actions.add(action) + extra = fetch_wikipedia_evidence(term, domain=domain_hint, queries=[query], db=db, agent_run_id=agent_run_id) + evidence = _dedupe_evidence([*evidence, *extra]) + observations.append({"action": action, "query": query, "count": len(extra), "total_evidence": len(evidence)}) + continue + + if action == "search_web" and rag_settings.search_enabled: + tools_used.append("search") + used_actions.add(action) + extra = fetch_search_evidence( + term, + domain=domain_hint, + search_url=rag_settings.search_url, + queries=[query], + context=llm_context, + target_lang=target_lang, + config=chat_config, + db=db, + agent_run_id=agent_run_id, + ) + evidence = _dedupe_evidence([*evidence, *extra]) + observations.append({"action": action, "query": query, "count": len(extra), "total_evidence": len(evidence)}) + continue + + if action == "fetch_url": + tools_used.append("fetch") + used_actions.add(action) + url = str(decision.get("url") or "").strip() + if not url: + for item in reversed(evidence): + candidate = str(item.get("url") or "").strip() + if candidate and not str(item.get("content") or "").strip(): + url = candidate + break + page = fetch_url_evidence(url=url, db=db, agent_run_id=agent_run_id) if url else None + if page: + evidence = _dedupe_evidence([*evidence, page]) + observations.append({"action": action, "url": url, "ok": bool(page), "total_evidence": len(evidence)}) + continue + + observations.append({"action": action, "reason": f"tool unavailable or disabled: {action}"}) + + # Safety fallback: if the model stopped too early, try enabled tools once. + if not evidence: + if rag_settings.wiki_enabled and "wiki_search" not in used_actions: + tools_used.append("wikipedia") + evidence = _dedupe_evidence(fetch_wikipedia_evidence(term, domain=domain_hint, queries=search_queries, db=db, agent_run_id=agent_run_id)) + if not evidence and rag_settings.search_enabled and "search_web" not in used_actions: + tools_used.append("search") + evidence = _dedupe_evidence( + fetch_search_evidence( + term, + domain=domain_hint, + search_url=rag_settings.search_url, + queries=search_queries, + context=llm_context, + target_lang=target_lang, + config=chat_config, + db=db, + agent_run_id=agent_run_id, + ) + ) + return evidence, tools_used, rounds + + def explain_term_from_evidence_openai( *, term: str, @@ -1758,6 +2068,7 @@ def upsert_knowledge_item( created_by: str = "manual", embedding: list[float] | None = None, embedding_model: str = "", + dedupe_any_domain: bool = False, ) -> str: item_id = str(uuid.uuid4()) clean_item_type = str(item_type or "document").strip() or "document" @@ -1794,6 +2105,27 @@ def upsert_knowledge_item( ), {"target_lang": clean_target_lang, "domain": clean_domain, "normalized_term": norm}, ).first() + if not existing and dedupe_any_domain: + existing = db.execute( + text( + """ + SELECT id FROM translation_knowledge_items + WHERE item_type = 'term' + AND target_lang = :target_lang + AND normalized_term = :normalized_term + AND status <> 'archived' + ORDER BY + CASE + WHEN domain = :domain THEN 0 + WHEN domain = '' THEN 1 + ELSE 2 + END, + updated_at DESC + LIMIT 1 + """ + ), + {"target_lang": clean_target_lang, "domain": clean_domain, "normalized_term": norm}, + ).first() if existing: item_id = str(existing[0]) db.execute( @@ -1959,6 +2291,71 @@ def exact_term_hits( return [_row_to_hit(row) for row in rows] +def existing_term_norms( + db: Session, + *, + terms: Iterable[str], + target_lang: str, + include_archived: bool = False, + limit: int = 100, +) -> set[str]: + norms: list[str] = [] + seen: set[str] = set() + for term in terms: + norm = normalize_term(str(term or "")) + if not norm or norm in seen: + continue + seen.add(norm) + norms.append(norm) + if not norms: + return set() + + clauses = [ + "item_type = 'term'", + "target_lang = :target_lang", + "normalized_term = ANY(:norms)", + ] + if not include_archived: + clauses.append("status <> 'archived'") + try: + rows = db.execute( + text( + f""" + SELECT DISTINCT normalized_term + FROM translation_knowledge_items + WHERE {' AND '.join(clauses)} + LIMIT :limit + """ + ), + { + "target_lang": str(target_lang or "zh").strip() or "zh", + "norms": norms, + "limit": max(1, min(500, int(limit))), + }, + ).all() + except Exception: + try: + db.rollback() + except Exception: + pass + return set() + + out: set[str] = set() + for row in rows: + mapping = getattr(row, "_mapping", None) + if mapping is not None: + value = mapping.get("normalized_term") + else: + try: + value = row[0] + except Exception: + value = None + norm = normalize_term(str(value or "")) + if norm: + out.add(norm) + return out + + def _research_discovered_term( db: Session, *, @@ -1972,8 +2369,9 @@ def _research_discovered_term( previous_summary: str, existing_term_cards: list[dict[str, Any]], gate_duration_ms: int | None, - task_id: str | None, - subtitle_job_id: str | None, + parent_agent_run_id: str | None = None, + task_id: str | None = None, + subtitle_job_id: str | None = None, ) -> AgentResearchResult | None: term = str(item.get("term") or "").strip() norm = normalize_term(term) @@ -1990,6 +2388,8 @@ def _research_discovered_term( domain=domain_hint, target_lang=target_lang, query=query, + agent_type="rag_term_research", + parent_agent_run_id=parent_agent_run_id, task_id=task_id, subtitle_job_id=subtitle_job_id, ) @@ -2080,6 +2480,21 @@ def _research_discovered_term( ) return AgentResearchResult(term=term, normalized_term=norm) + if norm in existing_term_norms(db, terms=[term], target_lang=target_lang): + _finish_agent_run( + db, + agent_run_id, + status="skipped", + result={ + "term": term, + "domain": domain_hint, + "knowledge_status": "already_exists", + "failure_category": "existing_knowledge", + "reason": "term already exists in the knowledge base; skipping external research", + }, + ) + return AgentResearchResult(term=term, normalized_term=norm) + external_lookup_enabled = bool(rag_settings.wiki_enabled or rag_settings.search_enabled) search_queries: list[str] = [] if external_lookup_enabled: @@ -2128,17 +2543,21 @@ def _research_discovered_term( "reason": "LLM query generation failed; using deterministic fallback queries.", }, ) - evidence = [] - if rag_settings.wiki_enabled: - evidence = fetch_wikipedia_evidence( + tools_used: list[str] = [] + + def _fetch_wiki_evidence_round() -> list[dict[str, Any]]: + tools_used.append("wikipedia") + return fetch_wikipedia_evidence( term, domain=domain_hint, queries=search_queries, db=db, agent_run_id=agent_run_id, ) - if not evidence and rag_settings.search_enabled: - evidence = fetch_search_evidence( + + def _fetch_search_evidence_round() -> list[dict[str, Any]]: + tools_used.append("search") + return fetch_search_evidence( term, domain=domain_hint, search_url=rag_settings.search_url, @@ -2149,6 +2568,48 @@ def _research_discovered_term( db=db, agent_run_id=agent_run_id, ) + + def _fetch_fallback_evidence(reason: str) -> list[dict[str, Any]]: + """Try the next enabled-but-unused evidence tool instead of giving up. + + Design: when one tool fails or its evidence is judged insufficient, the agent + must fall back to the other enabled tools (e.g. wiki -> web search) before + finishing the run. + """ + if rag_settings.search_enabled and "search" not in tools_used: + fallback_tool = "search" + extra = _fetch_search_evidence_round() + elif rag_settings.wiki_enabled and "wikipedia" not in tools_used: + fallback_tool = "wikipedia" + extra = _fetch_wiki_evidence_round() + else: + return [] + _append_agent_step( + db, + agent_run_id, + { + "kind": "policy", + "action": "evidence_tool_fallback", + "tool": fallback_tool, + "reason": reason, + "extra_evidence_count": len(extra), + }, + ) + return extra + + evidence, agent_tools_used, evidence_rounds = _collect_evidence_with_tool_agent( + db, + agent_run_id=agent_run_id, + term=term, + domain_hint=domain_hint, + target_lang=target_lang, + rag_settings=rag_settings, + chat_config=chat_config, + llm_context=llm_context, + search_queries=search_queries, + max_steps=6, + ) + tools_used.extend(agent_tools_used) if not evidence: _finish_agent_run( db, @@ -2163,107 +2624,167 @@ def _research_discovered_term( "wikipedia": bool(rag_settings.wiki_enabled), "search": bool(rag_settings.search_enabled), }, + "tools_used": tools_used, }, error="no search evidence", ) return None - try: - summarize_started = time.perf_counter() - explained = explain_term_from_evidence_openai( - term=term, - context=llm_context, - target_lang=target_lang, - domain_hint=domain_hint, - evidence=evidence, - config=chat_config, - ) - _append_llm_step( - db, - agent_run_id, - action="summarize_evidence", - config=chat_config, - input_value={"term": term, "domain": domain_hint, "evidence_count": len(evidence)}, - output_value={ - "term": str((explained or {}).get("term") or term), - "translation": str((explained or {}).get("translation") or ""), - "confidence": float((explained or {}).get("confidence") or 0.0), - }, - duration_ms=_duration_ms(summarize_started), - ) - except Exception as e: - _append_llm_step( - db, - agent_run_id, - action="summarize_failed", - config=chat_config, - error=str(e)[:300], - error_type=type(e).__name__, - ) - explained = None - if not explained: - _finish_agent_run( - db, - agent_run_id, - status="failed", - result={"term": term, "domain": domain_hint, "evidence_count": len(evidence), "failure_category": "summarize_failed"}, - error="LLM did not produce a usable glossary entry", + explained: dict[str, Any] | None = None + verification: dict[str, Any] = { + "supported": False, + "context_consistent": False, + "should_write": False, + "should_auto_approve": False, + "confidence": 0.0, + "reason": "verification did not run", + "failure_category": "verify_not_run", + } + valid_sources: list[dict[str, str]] = [] + final_confidence = 0.0 + should_write = False + max_evidence_rounds = max(2, evidence_rounds + 1) + while True: + summarized: dict[str, Any] | None = None + try: + summarize_started = time.perf_counter() + summarized = explain_term_from_evidence_openai( + term=term, + context=llm_context, + target_lang=target_lang, + domain_hint=domain_hint, + evidence=evidence, + config=chat_config, + ) + _append_llm_step( + db, + agent_run_id, + action="summarize_evidence", + config=chat_config, + input_value={ + "term": term, + "domain": domain_hint, + "evidence_count": len(evidence), + "evidence_round": evidence_rounds, + }, + output_value={ + "term": str((summarized or {}).get("term") or term), + "translation": str((summarized or {}).get("translation") or ""), + "confidence": float((summarized or {}).get("confidence") or 0.0), + }, + duration_ms=_duration_ms(summarize_started), + ) + except Exception as e: + _append_llm_step( + db, + agent_run_id, + action="summarize_failed", + config=chat_config, + error=str(e)[:300], + error_type=type(e).__name__, + ) + summarized = None + if summarized is not None: + explained = summarized + elif explained is not None: + # Summarizer failed on a fallback round; keep the previous candidate and stop retrying. + break + else: + if evidence_rounds < max_evidence_rounds: + extra = _fetch_fallback_evidence("summarizer produced no usable glossary entry; trying another evidence tool") + if extra: + evidence = extra + evidence + evidence_rounds += 1 + continue + _finish_agent_run( + db, + agent_run_id, + status="failed", + result={ + "term": term, + "domain": domain_hint, + "evidence_count": len(evidence), + "failure_category": "summarize_failed", + "tools_used": tools_used, + "evidence_rounds": evidence_rounds, + }, + error="LLM did not produce a usable glossary entry", + ) + return None + + valid_sources = _valid_external_evidence(evidence, search_url=rag_settings.search_url) + try: + verify_started = time.perf_counter() + verification = verify_glossary_entry_openai( + term=term, + context=llm_context, + target_lang=target_lang, + domain_hint=domain_hint, + evidence=evidence, + candidate=explained, + config=chat_config, + ) + _append_llm_step( + db, + agent_run_id, + action="verify_glossary_entry", + config=chat_config, + input_value={ + "term": term, + "candidate_confidence": float(explained.get("confidence") or 0.0), + "valid_source_count": len(valid_sources), + "evidence_round": evidence_rounds, + }, + output_value=verification, + duration_ms=_duration_ms(verify_started), + ) + except Exception as e: + verification = { + "supported": False, + "context_consistent": False, + "should_write": False, + "should_auto_approve": False, + "confidence": 0.0, + "reason": "verifier failed", + "failure_category": "verify_failed", + } + _append_llm_step( + db, + agent_run_id, + action="verify_failed", + config=chat_config, + error=str(e)[:300], + error_type=type(e).__name__, + ) + + candidate_confidence = float(explained.get("confidence") or 0.0) + verifier_confidence = float(verification.get("confidence") or 0.0) + final_confidence = min(candidate_confidence, verifier_confidence) if verifier_confidence > 0 else candidate_confidence + should_write = ( + bool(verification.get("supported")) + and bool(verification.get("context_consistent")) + and bool(verification.get("should_write")) + and bool(valid_sources) ) + if should_write: + break + # Before finishing a rejected candidate, try another enabled evidence tool. + # context_only is a semantic rejection rather than an evidence gap. + failure_category = str(verification.get("failure_category") or "").strip() + should_try_more_evidence = failure_category != "context_only" + if should_try_more_evidence and evidence_rounds < max_evidence_rounds: + extra = _fetch_fallback_evidence( + "verifier rejected the candidate " + f"(failure_category={failure_category or 'unknown'}); trying another evidence tool" + ) + if extra: + evidence = extra + evidence + evidence_rounds += 1 + continue + break + + if explained is None: return None - - valid_sources = _valid_external_evidence(evidence, search_url=rag_settings.search_url) - try: - verify_started = time.perf_counter() - verification = verify_glossary_entry_openai( - term=term, - context=llm_context, - target_lang=target_lang, - domain_hint=domain_hint, - evidence=evidence, - candidate=explained, - config=chat_config, - ) - _append_llm_step( - db, - agent_run_id, - action="verify_glossary_entry", - config=chat_config, - input_value={ - "term": term, - "candidate_confidence": float(explained.get("confidence") or 0.0), - "valid_source_count": len(valid_sources), - }, - output_value=verification, - duration_ms=_duration_ms(verify_started), - ) - except Exception as e: - verification = { - "supported": False, - "context_consistent": False, - "should_write": False, - "should_auto_approve": False, - "confidence": 0.0, - "reason": "verifier failed", - "failure_category": "verify_failed", - } - _append_llm_step( - db, - agent_run_id, - action="verify_failed", - config=chat_config, - error=str(e)[:300], - error_type=type(e).__name__, - ) - - candidate_confidence = float(explained.get("confidence") or 0.0) - verifier_confidence = float(verification.get("confidence") or 0.0) - final_confidence = min(candidate_confidence, verifier_confidence) if verifier_confidence > 0 else candidate_confidence - should_write = ( - bool(verification.get("supported")) - and bool(verification.get("context_consistent")) - and bool(verification.get("should_write")) - and bool(valid_sources) - ) if not should_write: failure_category = str(verification.get("failure_category") or "") if not valid_sources: @@ -2282,6 +2803,8 @@ def _research_discovered_term( "verification": verification, "valid_source_count": len(valid_sources), "search_queries": search_queries, + "tools_used": tools_used, + "evidence_rounds": evidence_rounds, }, error=str(verification.get("reason") or "verifier rejected glossary entry")[:4000], ) @@ -2333,6 +2856,7 @@ def _research_discovered_term( created_by="agent", embedding=emb, embedding_model=embedding_model_key(rag_settings), + dedupe_any_domain=True, ) db.commit() _finish_agent_run( @@ -2405,8 +2929,9 @@ def _run_research_agent( previous_summary: str, existing_term_cards: list[dict[str, Any]], gate_duration_ms: int | None, - task_id: str | None, - subtitle_job_id: str | None, + parent_agent_run_id: str | None = None, + task_id: str | None = None, + subtitle_job_id: str | None = None, ) -> AgentResearchResult | None: if session_factory is None: return _research_discovered_term( @@ -2421,6 +2946,7 @@ def _run_research_agent( previous_summary=previous_summary, existing_term_cards=existing_term_cards, gate_duration_ms=gate_duration_ms, + parent_agent_run_id=parent_agent_run_id, task_id=task_id, subtitle_job_id=subtitle_job_id, ) @@ -2438,6 +2964,7 @@ def _run_research_agent( previous_summary=previous_summary, existing_term_cards=existing_term_cards, gate_duration_ms=gate_duration_ms, + parent_agent_run_id=parent_agent_run_id, task_id=task_id, subtitle_job_id=subtitle_job_id, ) @@ -2459,8 +2986,9 @@ def _run_research_agents( previous_summary: str, existing_term_cards: list[dict[str, Any]], gate_duration_ms: int | None, - task_id: str | None, - subtitle_job_id: str | None, + parent_agent_run_id: str | None = None, + task_id: str | None = None, + subtitle_job_id: str | None = None, ) -> list[AgentResearchResult]: if not items: return [] @@ -2487,6 +3015,7 @@ def _run_research_agents( previous_summary=previous_summary, existing_term_cards=existing_term_cards, gate_duration_ms=gate_duration_ms, + parent_agent_run_id=parent_agent_run_id, task_id=task_id, subtitle_job_id=subtitle_job_id, ) @@ -2513,6 +3042,7 @@ def _run_research_agents( previous_summary=previous_summary, existing_term_cards=existing_term_cards, gate_duration_ms=gate_duration_ms, + parent_agent_run_id=parent_agent_run_id, task_id=task_id, subtitle_job_id=subtitle_job_id, ) @@ -2554,29 +3084,44 @@ def build_rag_context( if not text_value: return RagContext(term_cards=[], knowledge_cards=[], hits=[]) llm_context = _context_for_llm(text_value, previous_summary=previous_summary, limit=9000) + master_agent_run_id: str | None = None + try: + master_agent_run_id = _start_agent_run( + db, + term="", + domain=rag_settings.domain, + target_lang=target_lang, + query=text_value[:240], + agent_type="rag_master", + task_id=task_id, + subtitle_job_id=subtitle_job_id, + ) + _append_agent_step( + db, + master_agent_run_id, + { + "kind": "agent", + "action": "master_start", + "context_excerpt": text_value[:500], + "previous_summary": str(previous_summary or "")[:500], + "tools": { + "rag_lookup": True, + "wikipedia": bool(rag_settings.wiki_enabled), + "search": bool(rag_settings.search_enabled), + "fetch_url": True, + }, + }, + ) + except Exception: + db.rollback() + master_agent_run_id = None hits: list[RagHit] = [] seen_ids: set[str] = set() gate_terms: list[dict[str, Any]] | None = None gate_norms: set[str] = set() gate_duration_ms: int | None = None - - if rag_settings.auto_discover_terms: - gate_started = time.perf_counter() - try: - gate_terms = pretranslation_rag_gate_openai( - text_value, - target_lang=target_lang, - domain_hint=rag_settings.domain, - existing_terms=[], - previous_summary=previous_summary, - config=chat_config, - ) - gate_norms = {normalize_term(str(item.get("term") or "")) for item in gate_terms if isinstance(item, dict)} - gate_norms.discard("") - except Exception: - gate_terms = None - gate_duration_ms = _duration_ms(gate_started) + existing_gate_terms: list[dict[str, Any]] = [] for hit in exact_term_hits( db, @@ -2585,11 +3130,70 @@ def build_rag_context( domain=rag_settings.domain, limit=max(rag_settings.top_k * 2, 10), ): - if gate_terms is not None and normalize_term(hit.term) not in gate_norms: - continue if hit.id not in seen_ids: seen_ids.add(hit.id) hits.append(hit) + if hit.item_type == "term": + existing_gate_terms.append( + { + "term": hit.term, + "translation": hit.translation, + "domain": hit.domain, + "score": round(hit.score, 4), + } + ) + if existing_gate_terms: + _append_agent_step( + db, + master_agent_run_id, + { + "kind": "tool", + "action": "master_rag_lookup", + "tool": "rag_lookup", + "count": len(existing_gate_terms), + "results": existing_gate_terms[:10], + "ok": True, + }, + ) + + if rag_settings.auto_discover_terms: + gate_started = time.perf_counter() + try: + gate_terms = pretranslation_rag_gate_openai( + text_value, + target_lang=target_lang, + domain_hint=rag_settings.domain, + existing_terms=existing_gate_terms, + previous_summary=previous_summary, + config=chat_config, + ) + gate_norms = {normalize_term(str(item.get("term") or "")) for item in gate_terms if isinstance(item, dict)} + gate_norms.discard("") + _append_llm_step( + db, + master_agent_run_id, + action="master_pretranslation_rag_gate", + config=chat_config, + input_value={ + "domain": rag_settings.domain, + "context_excerpt": text_value[:500], + "previous_summary": str(previous_summary or "")[:500], + "existing_hit_count": len(existing_gate_terms), + }, + output_value={"terms": gate_terms}, + duration_ms=_duration_ms(gate_started), + ) + except Exception: + gate_terms = None + _append_llm_step( + db, + master_agent_run_id, + action="master_pretranslation_rag_gate_failed", + config=chat_config, + duration_ms=_duration_ms(gate_started), + error_type="gate_failed", + ) + gate_duration_ms = _duration_ms(gate_started) try: rag_query_text = text_value[:8000] @@ -2628,6 +3232,7 @@ def build_rag_context( query_embedding = [] context_only_cards: list[dict[str, Any]] = [] + subagent_count = 0 if rag_settings.auto_discover_terms: known_norms = {normalize_term(h.term) for h in hits if h.term} @@ -2661,6 +3266,12 @@ def build_rag_context( except Exception: discovered = [] gate_duration_ms = _duration_ms(fallback_started) + discovered_terms = [ + str(item.get("term") or "").strip() + for item in discovered + if isinstance(item, dict) and str(item.get("term") or "").strip() + ] + known_norms.update(existing_term_norms(db, terms=discovered_terms, target_lang=target_lang)) research_items: list[dict[str, Any]] = [] for item in discovered: term = str(item.get("term") or "").strip() @@ -2671,6 +3282,19 @@ def build_rag_context( continue research_items.append(item) known_norms.add(norm) + subagent_count = len(research_items) + _append_agent_step( + db, + master_agent_run_id, + { + "kind": "agent", + "action": "master_dispatch_subagents", + "candidate_count": len(discovered), + "research_count": len(research_items), + "terms": [str(item.get("term") or "") for item in research_items[:20] if isinstance(item, dict)], + "parallelism": rag_settings.agent_parallelism, + }, + ) for result in _run_research_agents( db=db, session_factory=session_factory, @@ -2684,6 +3308,7 @@ def build_rag_context( previous_summary=previous_summary, existing_term_cards=existing_term_cards, gate_duration_ms=gate_duration_ms, + parent_agent_run_id=master_agent_run_id, task_id=task_id, subtitle_job_id=subtitle_job_id, ): @@ -2732,6 +3357,18 @@ def build_rag_context( context=text_value[:2000], ) + _finish_agent_run( + db, + master_agent_run_id, + status="succeeded", + result={ + "term_cards": len(term_cards), + "knowledge_cards": len(knowledge_cards), + "hits": len(hits), + "target_lang": target_lang, + "subagents": subagent_count, + }, + ) return RagContext(term_cards=term_cards, knowledge_cards=knowledge_cards, hits=hits) @@ -2855,7 +3492,7 @@ def list_agent_runs(db: Session, *, status: str | None = None, limit: int = 50) text( f""" SELECT id, agent_type, status, term, domain, target_lang, task_id, subtitle_job_id, - query, steps, result, error, knowledge_item_id, + query, steps, result, error, knowledge_item_id, parent_agent_run_id, started_at, finished_at, created_at, updated_at FROM translation_agent_runs WHERE {' AND '.join(clauses)} @@ -2895,6 +3532,7 @@ def list_agent_runs(db: Session, *, status: str | None = None, limit: int = 50) "result": result if isinstance(result, dict) else {}, "error": m["error"], "knowledge_item_id": str(m["knowledge_item_id"]) if m["knowledge_item_id"] else None, + "parent_agent_run_id": str(m["parent_agent_run_id"]) if m["parent_agent_run_id"] else None, "started_at": m["started_at"], "finished_at": m["finished_at"], "created_at": m["created_at"], diff --git a/src/videoroll/apps/subtitle_service/schemas.py b/src/videoroll/apps/subtitle_service/schemas.py index 43baa6c..0bc79e2 100644 --- a/src/videoroll/apps/subtitle_service/schemas.py +++ b/src/videoroll/apps/subtitle_service/schemas.py @@ -201,6 +201,7 @@ class TranslateSettingsRead(BaseModel): openai_model: str openai_temperature: float openai_timeout_seconds: float + openai_max_retries: int = 3 rag_enabled: bool = False rag_top_k: int = 8 @@ -237,6 +238,7 @@ class TranslateSettingsUpdate(BaseModel): openai_model: Optional[str] = None openai_temperature: Optional[float] = None openai_timeout_seconds: Optional[float] = None + openai_max_retries: Optional[int] = Field(default=None, ge=1, le=10) rag_enabled: Optional[bool] = None rag_top_k: Optional[int] = Field(default=None, ge=0, le=30) @@ -341,6 +343,7 @@ class AgentRunRead(BaseModel): result: dict[str, Any] = Field(default_factory=dict) error: str = "" knowledge_item_id: Optional[uuid.UUID] = None + parent_agent_run_id: Optional[uuid.UUID] = None started_at: datetime finished_at: Optional[datetime] = None created_at: datetime diff --git a/src/videoroll/apps/subtitle_service/translate_settings_store.py b/src/videoroll/apps/subtitle_service/translate_settings_store.py index 602de68..2eeac79 100644 --- a/src/videoroll/apps/subtitle_service/translate_settings_store.py +++ b/src/videoroll/apps/subtitle_service/translate_settings_store.py @@ -65,6 +65,7 @@ def get_translate_settings(db: Session, defaults: SubtitleServiceSettings) -> di "openai_model": str(openai.get("model") or defaults.openai_model), "openai_temperature": float(openai.get("temperature") or defaults.openai_temperature), "openai_timeout_seconds": float(openai.get("timeout_seconds") or defaults.openai_timeout_seconds), + "openai_max_retries": int(openai.get("max_retries") or stored.get("openai_max_retries") or 3), "rag_enabled": bool(stored.get("rag_enabled") if "rag_enabled" in stored else defaults.rag_enabled), "rag_top_k": int(stored.get("rag_top_k") or defaults.rag_top_k), "rag_min_score": float(stored.get("rag_min_score") or defaults.rag_min_score), @@ -148,6 +149,7 @@ def update_translate_settings(db: Session, defaults: SubtitleServiceSettings, up ("openai_model", "model"), ("openai_temperature", "temperature"), ("openai_timeout_seconds", "timeout_seconds"), + ("openai_max_retries", "max_retries"), ]: if key not in update: continue @@ -239,6 +241,11 @@ def update_translate_settings(db: Session, defaults: SubtitleServiceSettings, up openai["timeout_seconds"] = float(openai.get("timeout_seconds")) except Exception: openai["timeout_seconds"] = defaults.openai_timeout_seconds + try: + if "max_retries" in openai and openai["max_retries"] is not None: + openai["max_retries"] = max(1, min(10, int(openai.get("max_retries") or 3))) + except Exception: + openai["max_retries"] = 3 row.value_json = stored db.add(row) diff --git a/src/videoroll/db/auto_migrate.py b/src/videoroll/db/auto_migrate.py index c61afe3..b2db024 100644 --- a/src/videoroll/db/auto_migrate.py +++ b/src/videoroll/db/auto_migrate.py @@ -201,6 +201,7 @@ def _ensure_pgvector_rag_tables(engine: Engine) -> None: result JSONB NOT NULL DEFAULT '{}'::jsonb, error TEXT NOT NULL DEFAULT '', knowledge_item_id UUID REFERENCES translation_knowledge_items(id) ON DELETE SET NULL, + parent_agent_run_id UUID, started_at TIMESTAMPTZ NOT NULL DEFAULT now(), finished_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), @@ -209,6 +210,14 @@ def _ensure_pgvector_rag_tables(engine: Engine) -> None: """ ) ) + conn.execute( + text( + """ + ALTER TABLE translation_agent_runs + ADD COLUMN IF NOT EXISTS parent_agent_run_id UUID + """ + ) + ) conn.execute( text( """ @@ -226,6 +235,14 @@ def _ensure_pgvector_rag_tables(engine: Engine) -> None: """ ) ) + conn.execute( + text( + """ + CREATE INDEX IF NOT EXISTS ix_translation_agent_runs_parent + ON translation_agent_runs (parent_agent_run_id, updated_at DESC) + """ + ) + ) conn.execute( text( """ diff --git a/src/web/src/components/ui.tsx b/src/web/src/components/ui.tsx index 6d18226..4001c1b 100644 --- a/src/web/src/components/ui.tsx +++ b/src/web/src/components/ui.tsx @@ -61,10 +61,14 @@ export function Section({ children, className = "" }: PropsWithChildren<{ classN return
{children}
; } -export function DataTable({ children }: PropsWithChildren) { +export function DataTable({ + children, + className = "", + wrapClassName = "", +}: PropsWithChildren<{ className?: string; wrapClassName?: string }>) { return ( -
- {children}
+
+ {children}
); } diff --git a/src/web/src/pages/DashboardPage.tsx b/src/web/src/pages/DashboardPage.tsx index 98e2a6c..c63ab8a 100644 --- a/src/web/src/pages/DashboardPage.tsx +++ b/src/web/src/pages/DashboardPage.tsx @@ -56,6 +56,7 @@ type AgentRun = { result: Record; error: string; knowledge_item_id?: string | null; + parent_agent_run_id?: string | null; started_at: string; finished_at?: string | null; created_at: string; @@ -177,7 +178,17 @@ function AgentStep({ step, index }: { step: Record; index: numb ); } -function AgentRunDetail({ run, onClose }: { run: AgentRun; onClose: () => void }) { +function AgentRunDetail({ + run, + childrenRuns = [], + onSelectRun, + onClose, +}: { + run: AgentRun; + childrenRuns?: AgentRun[]; + onSelectRun: (run: AgentRun) => void; + onClose: () => void; +}) { const translation = textValue(run.result?.translation); const confidence = numberValue(run.result?.confidence); const knowledgeStatus = textValue(run.result?.knowledge_status); @@ -234,6 +245,27 @@ function AgentRunDetail({ run, onClose }: { run: AgentRun; onClose: () => void } 
{prettyJson(run.result)}
) : null} + {childrenRuns.length ? ( +
+
子 Agent
+
+ {childrenRuns.map((child) => ( + + ))} +
+
+ ) : null}
对话流
@@ -248,7 +280,7 @@ function AgentRunDetail({ run, onClose }: { run: AgentRun; onClose: () => void } ); } -function AgentRunCard({ run, onSelect }: { run: AgentRun; onSelect: (run: AgentRun) => void }) { +function AgentRunCard({ run, onSelect, childrenRuns = [] }: { run: AgentRun; onSelect: (run: AgentRun) => void; childrenRuns?: AgentRun[] }) { const translation = textValue(run.result?.translation); const confidence = numberValue(run.result?.confidence); const knowledgeStatus = textValue(run.result?.knowledge_status); @@ -268,6 +300,7 @@ function AgentRunCard({ run, onSelect }: { run: AgentRun; onSelect: (run: AgentR {run.domain ?
domain: {run.domain}
: null} {run.query ?
query: {run.query}
: null} {opened ?
opened/read pages: {opened}
: null} + {childrenRuns.length ?
subagents: {childrenRuns.length}
: null} {latestStep?.action ?
latest: {String(latestStep.action)}
: null} {translation ? (
@@ -279,6 +312,16 @@ function AgentRunCard({ run, onSelect }: { run: AgentRun; onSelect: (run: AgentR {failureCategory ?
failure: {failureCategory}
: null} {run.error ?
{run.error}
: null}
+ {childrenRuns.length ? ( +
+ {childrenRuns.slice(0, 5).map((child) => ( +
+ {child.term || child.query || child.id} + {child.status} +
+ ))} +
+ ) : null} ); } @@ -385,8 +428,19 @@ export default function DashboardPage() { const total = tasks?.length ?? 0; const failedTasks = useMemo(() => (tasks ?? []).filter((t) => t.status === "FAILED").slice(0, 6), [tasks]); const runningTasks = useMemo(() => (tasks ?? []).filter((t) => runningStatuses.has(t.status)).slice(0, 6), [tasks]); - const runningAgents = useMemo(() => (agentRuns ?? []).filter((run) => run.status === "running"), [agentRuns]); - const recentFinishedAgents = useMemo(() => (agentRuns ?? []).filter((run) => run.status !== "running").slice(0, 6), [agentRuns]); + const agentChildren = useMemo(() => { + const out = new Map(); + for (const run of agentRuns ?? []) { + if (!run.parent_agent_run_id) continue; + const rows = out.get(run.parent_agent_run_id) ?? []; + rows.push(run); + out.set(run.parent_agent_run_id, rows); + } + return out; + }, [agentRuns]); + const topLevelAgents = useMemo(() => (agentRuns ?? []).filter((run) => !run.parent_agent_run_id), [agentRuns]); + const runningAgents = useMemo(() => topLevelAgents.filter((run) => run.status === "running"), [topLevelAgents]); + const recentFinishedAgents = useMemo(() => topLevelAgents.filter((run) => run.status !== "running").slice(0, 6), [topLevelAgents]); const selectedAgent = useMemo(() => (agentRuns ?? []).find((run) => run.id === selectedAgentId) ?? null, [agentRuns, selectedAgentId]); const publishedCount = counts.find(([status]) => status === "PUBLISHED")?.[1] ?? 0; const failedCount = counts.find(([status]) => status === "FAILED")?.[1] ?? 0; @@ -522,14 +576,14 @@ export default function DashboardPage() {
正在工作
{runningAgents.length === 0 ?
暂无运行中的 agent。
: null} - {runningAgents.map((run) => setSelectedAgentId(item.id)} />)} + {runningAgents.map((run) => setSelectedAgentId(item.id)} />)}
最近结果
{recentFinishedAgents.length === 0 ?
暂无 agent 结果。
: null} - {recentFinishedAgents.map((run) => setSelectedAgentId(item.id)} />)} + {recentFinishedAgents.map((run) => setSelectedAgentId(item.id)} />)}
@@ -648,7 +702,14 @@ export default function DashboardPage() { ) : null} - {selectedAgent ? setSelectedAgentId(null)} /> : null} + {selectedAgent ? ( + setSelectedAgentId(run.id)} + onClose={() => setSelectedAgentId(null)} + /> + ) : null} ); } diff --git a/src/web/src/pages/KnowledgeBasePage.tsx b/src/web/src/pages/KnowledgeBasePage.tsx index f2be615..c06fd61 100644 --- a/src/web/src/pages/KnowledgeBasePage.tsx +++ b/src/web/src/pages/KnowledgeBasePage.tsx @@ -235,7 +235,7 @@ export default function KnowledgeBasePage() { {items.length === 0 ? ( 暂无知识条目 ) : ( - + Type diff --git a/src/web/src/pages/SettingsTranslatePage.tsx b/src/web/src/pages/SettingsTranslatePage.tsx index a3b99db..934dc7f 100644 --- a/src/web/src/pages/SettingsTranslatePage.tsx +++ b/src/web/src/pages/SettingsTranslatePage.tsx @@ -17,6 +17,7 @@ type TranslateSettings = { openai_model: string; openai_temperature: number; openai_timeout_seconds: number; + openai_max_retries: number; rag_enabled: boolean; rag_top_k: number; rag_min_score: number; @@ -67,6 +68,7 @@ export default function SettingsTranslatePage() { const [openaiModel, setOpenaiModel] = useState("gpt-4o-mini"); const [openaiTemperature, setOpenaiTemperature] = useState(0.2); const [openaiTimeoutSeconds, setOpenaiTimeoutSeconds] = useState(60); + const [openaiMaxRetries, setOpenaiMaxRetries] = useState(3); const [openaiApiKey, setOpenaiApiKey] = useState(""); const [ragEnabled, setRagEnabled] = useState(false); @@ -119,6 +121,7 @@ export default function SettingsTranslatePage() { setOpenaiModel(s.openai_model); setOpenaiTemperature(s.openai_temperature); setOpenaiTimeoutSeconds(s.openai_timeout_seconds); + setOpenaiMaxRetries(s.openai_max_retries ?? 3); setRagEnabled(s.rag_enabled); setRagTopK(s.rag_top_k); setRagMinScore(s.rag_min_score); @@ -161,6 +164,7 @@ export default function SettingsTranslatePage() { openai_model: openaiModel, openai_temperature: openaiTemperature, openai_timeout_seconds: openaiTimeoutSeconds, + openai_max_retries: openaiMaxRetries, rag_enabled: ragEnabled, rag_top_k: ragTopK, rag_min_score: ragMinScore, @@ -397,6 +401,11 @@ export default function SettingsTranslatePage() {
openai_timeout_seconds
setOpenaiTimeoutSeconds(parseFloat(e.target.value || "1"))} /> + diff --git a/tests/test_translation_rag.py b/tests/test_translation_rag.py index 0cdf2ce..340573a 100644 --- a/tests/test_translation_rag.py +++ b/tests/test_translation_rag.py @@ -1,11 +1,17 @@ from __future__ import annotations +import json + +import httpx + from videoroll.ai.client import OpenAIChatConfig +from videoroll.ai.client import request_openai_embedding from videoroll.apps.subtitle_service.rag import ( _run_research_agents, _fallback_search_queries, _search_url_with_params, build_knowledge_embedding_text, + build_rag_context, embedding_model_key, fetch_search_evidence, fetch_wikipedia_evidence, @@ -17,6 +23,7 @@ from videoroll.apps.subtitle_service.rag import ( search_knowledge, should_research_term, ) +from videoroll.apps.subtitle_service.processing import Segment from videoroll.apps.subtitle_service.embeddings import ( assert_embedding_dimensions, embed_text, @@ -220,7 +227,7 @@ def test_run_research_agents_uses_session_factory_for_parallel_terms(monkeypatch from videoroll.apps.subtitle_service import rag as rag_module created_sessions: list[object] = [] - calls: list[tuple[object, str]] = [] + calls: list[tuple[object, str, str | None]] = [] class _Session: def close(self) -> None: @@ -231,9 +238,9 @@ def test_run_research_agents_uses_session_factory_for_parallel_terms(monkeypatch created_sessions.append(session) return session - def fake_research(db, *, item, **_kwargs): + def fake_research(db, *, item, **kwargs): term = item["term"] - calls.append((db, term)) + calls.append((db, term, kwargs.get("parent_agent_run_id"))) return rag_module.AgentResearchResult(term=term, normalized_term=normalize_term(term)) monkeypatch.setattr(rag_module, "_research_discovered_term", fake_research) @@ -259,13 +266,15 @@ def test_run_research_agents_uses_session_factory_for_parallel_terms(monkeypatch previous_summary="", existing_term_cards=[], gate_duration_ms=1, + parent_agent_run_id="master-run", task_id=None, subtitle_job_id=None, ) assert {item.term for item in results} == {"AWP", "Rush B"} assert len(created_sessions) == 2 - assert {id(db) for db, _term in calls} == {id(session) for session in created_sessions} + assert {id(db) for db, _term, _parent in calls} == {id(session) for session in created_sessions} + assert {parent for _db, _term, parent in calls} == {"master-run"} def test_fallback_search_queries_include_definition_and_target_language() -> None: @@ -371,6 +380,33 @@ def test_assert_embedding_dimensions() -> None: raise AssertionError("expected dimension mismatch") +def test_openai_embedding_request_sends_dimensions() -> None: + requests: list[dict[str, object]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + data = dict(json.loads(request.content.decode("utf-8"))) + requests.append(data) + assert data["model"] == "Qwen3-Embedding-8B" + assert data["input"] == "纳米" + assert data["dimensions"] == 4096 + return httpx.Response(200, json={"data": [{"embedding": [0.1, 0.2]}]}) + + client = httpx.Client(transport=httpx.MockTransport(handler)) + vector = request_openai_embedding( + config=OpenAIChatConfig( + api_key="embed-key", + base_url="https://embedding.example/v1", + model="Qwen3-Embedding-8B", + embedding_dimensions=4096, + ), + text="纳米", + client=client, + ) + + assert vector == [0.1, 0.2] + assert requests + + class _Result: def __init__(self, rows: list[object]) -> None: self._rows = rows @@ -408,6 +444,176 @@ class _FakeKnowledgeDb: return _NestedTransaction() +def test_research_falls_back_to_search_when_wiki_verification_rejects(monkeypatch) -> None: + from videoroll.apps.subtitle_service import rag as rag_module + + tool_calls: list[str] = [] + steps: list[dict[str, object]] = [] + finished: dict[str, object] = {} + + monkeypatch.setattr(rag_module, "_start_agent_run", lambda *_args, **_kwargs: "run-1") + monkeypatch.setattr(rag_module, "_append_llm_step", lambda *_args, **_kwargs: None) + monkeypatch.setattr(rag_module, "_append_agent_step", lambda _db, _run_id, step: steps.append(step)) + monkeypatch.setattr( + rag_module, + "_finish_agent_run", + lambda _db, _run_id, **kwargs: finished.update(kwargs), + ) + monkeypatch.setattr(rag_module, "existing_term_norms", lambda *_args, **_kwargs: set()) + monkeypatch.setattr(rag_module, "generate_search_queries_openai", lambda **_kwargs: ["hobby knife definition"]) + decisions = iter( + [ + {"action": "wiki_search", "query": "hobby knife definition", "reason": "start with encyclopedic source"}, + {"action": "finish", "reason": "wiki evidence collected", "final_answer_ready": True}, + ] + ) + monkeypatch.setattr(rag_module, "research_agent_next_action_openai", lambda **_kwargs: next(decisions)) + + def fake_wiki(*_args, **_kwargs): + tool_calls.append("wiki") + return [ + { + "title": "Hobby knife", + "url": "https://en.wikipedia.org/wiki/Utility_knife", + "snippet": "A precision knife used for hobbies.", + "content": "A precision knife used for hobbies and model making.", + } + ] + + def fake_search(*_args, **_kwargs): + tool_calls.append("search") + return [ + { + "title": "Hobby knife definition", + "url": "https://example.com/hobby-knife", + "snippet": "Hobby knives are precision cutting tools.", + "content": "Hobby knives are precision cutting tools used by model makers.", + } + ] + + monkeypatch.setattr(rag_module, "fetch_wikipedia_evidence", fake_wiki) + monkeypatch.setattr(rag_module, "fetch_search_evidence", fake_search) + monkeypatch.setattr( + rag_module, + "explain_term_from_evidence_openai", + lambda **_kwargs: { + "term": "hobby knife", + "translation": "模型刀", + "domain": "手工模型", + "aliases": [], + "description": "精细切割工具。", + "sources": [], + "confidence": 0.8, + }, + ) + monkeypatch.setattr( + rag_module, + "verify_glossary_entry_openai", + lambda **_kwargs: { + "supported": False, + "context_consistent": False, + "should_write": False, + "should_auto_approve": False, + "confidence": 0.0, + "reason": "unsupported", + "failure_category": "unsupported_translation_or_no_context", + }, + ) + + result = rag_module._research_discovered_term( + object(), # type: ignore[arg-type] + item={"term": "hobby knife", "category": "domain_jargon", "need_rag": True, "need_search": True, "scope": "global"}, + target_lang="zh", + rag_settings=rag_settings_from_translate_settings( + { + "rag_enabled": True, + "rag_auto_learn_terms": True, + "rag_wiki_enabled": True, + "rag_search_enabled": True, + "rag_search_url": "https://search.example/search", + "rag_embedding_dimensions": 2, + } + ), + embedding_settings=embedding_settings_from_translate_settings({"rag_embedding_provider": "local", "rag_embedding_dimensions": 2}), + chat_config=OpenAIChatConfig(api_key="x", base_url="https://example.invalid/v1", model="demo"), + text_value="Use a hobby knife to cut the tape.", + llm_context="Use a hobby knife to cut the tape.", + previous_summary="", + existing_term_cards=[], + gate_duration_ms=1, + task_id=None, + subtitle_job_id=None, + ) + + assert result is not None + assert tool_calls == ["wiki", "search"] + assert any(step.get("action") == "evidence_tool_fallback" and step.get("tool") == "search" for step in steps) + assert finished["status"] == "skipped" + assert finished["result"]["tools_used"] == ["wikipedia", "search"] # type: ignore[index] + + +def test_build_rag_context_skips_research_for_existing_pending_term(monkeypatch) -> None: + from videoroll.apps.subtitle_service import rag as rag_module + + research_items: list[dict[str, object]] = [] + + class _Db: + def execute(self, stmt: object, params: dict[str, object] | None = None) -> _Result: + sql = str(stmt) + if "SELECT DISTINCT normalized_term" in sql: + return _Result([_Row(normalized_term="hobby knife")]) + return _Result([]) + + def rollback(self) -> None: + return None + + monkeypatch.setattr( + rag_module, + "pretranslation_rag_gate_openai", + lambda *_args, **_kwargs: [ + { + "term": "hobby knife", + "category": "domain_jargon", + "domain": "手工模型", + "need_rag": True, + "need_search": True, + "scope": "global", + "priority": 0.8, + "reason": "tool name", + } + ], + ) + monkeypatch.setattr(rag_module, "exact_term_hits", lambda *_args, **_kwargs: []) + monkeypatch.setattr(rag_module, "embed_text", lambda *_args, **_kwargs: [0.1, 0.2]) + monkeypatch.setattr(rag_module, "assert_embedding_dimensions", lambda *_args, **_kwargs: None) + monkeypatch.setattr(rag_module, "search_knowledge", lambda *_args, **_kwargs: []) + + def fake_run_research_agents(*_args, **kwargs): + research_items.extend(kwargs["items"]) + return [] + + monkeypatch.setattr(rag_module, "_run_research_agents", fake_run_research_agents) + + ctx = build_rag_context( + _Db(), # type: ignore[arg-type] + segments=[Segment(0, 1, "Use a hobby knife to cut the tape.")], + target_lang="zh", + rag_settings=rag_settings_from_translate_settings( + { + "rag_enabled": True, + "rag_auto_discover_terms": True, + "rag_auto_learn_terms": True, + "rag_embedding_dimensions": 2, + } + ), + embedding_settings=embedding_settings_from_translate_settings({"rag_embedding_provider": "local", "rag_embedding_dimensions": 2}), + chat_config=OpenAIChatConfig(api_key="x", base_url="https://example.invalid/v1", model="demo"), + ) + + assert ctx.term_cards == [] + assert research_items == [] + + def test_search_knowledge_filters_by_embedding_model() -> None: db = _FakeKnowledgeDb([]) search_knowledge(