diff --git a/README.md b/README.md index 1555741..aec7803 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,17 @@ 详细规格见 [docs/PROJECT_SPEC.md](docs/PROJECT_SPEC.md)。 +## 显著功能 + +- **RAG 翻译 Agent**:字幕翻译前先扫描当前 block 和前文 summary,判断哪些术语真的需要查知识库或联网检索,减少把基础词、局部变量写入长期知识库。 +- **PostgreSQL + pgvector 知识库**:术语、译法、来源、置信度和使用次数统一入库;支持向量检索、手动新增/删除条目、重建 embedding。 +- **Wikipedia Tool + SearXNG Tool**:Agent 优先查固定 English Wikipedia API,必要时再调用自建 SearXNG,并可读取网页正文作为证据。 +- **并发术语研究 Agent**:可在 `Settings -> Translate` 设置同时研究的术语数量;每个 Agent 都有独立工具调用 trace。 +- **保守入库策略**:summarize -> verify -> write_rag;只有 verifier 判断来源支撑、上下文一致且置信度足够时才正式入库,否则标记为 pending、context_only 或 skipped。 +- **独立 embedding 配置**:翻译 API key/base URL 与 embedding API key/base URL 分开;embedding 支持 OpenAI 兼容服务和本地模型。 +- **可观测 Dashboard**:显示任务状态、CPU/内存/Intel GPU 资源监控,以及 RAG Agent 当前运行状态和历史对话流。 +- **完整 Web 配置**:ASR、翻译、RAG、YouTube、Bilibili、自动模式、知识库均可在 Web UI 中配置,减少部署环境变量依赖。 + ## 当前状态 - Web UI 已接入管理密码登录,首次打开需要设置 admin password。 @@ -15,7 +26,7 @@ - Bilibili 投稿支持在 Web UI 中保存 Cookie、测试登录、维护默认投稿模板。 - Web UI 支持浅色 / 暗色模式切换,Dashboard 增加 CPU、内存和 Intel GPU 资源监控。 - 字幕翻译支持真正的 RAG 术语知识库:可使用 PostgreSQL + pgvector 检索术语,支持本地 embedding 模型、手动下载模型、重建向量,以及 Agent 自动发现/搜索/验证/入库术语。 -- RAG Agent 支持 SearXNG 搜索、网页正文读取、独立 verifier、保守入库策略,并可在 Dashboard 查看每个 Agent 的工具调用 trace。 +- RAG Agent 支持固定 Wikipedia Tool、SearXNG 搜索、网页正文读取、独立 verifier、保守入库策略,并可在 Dashboard 查看每个 Agent 的工具调用 trace。 ## 架构概览 @@ -316,10 +327,13 @@ RAG 用于解决视频字幕翻译中的上下文和术语问题,例如游戏 ```text 字幕片段 - -> 术语发现 + + 前文翻译 summary + -> RAG gate 判断是否需要检索 + -> 术语发现与噪声过滤 -> 查询 PostgreSQL/pgvector 知识库 - -> 未命中时生成多条搜索 query - -> 调用 SearXNG 搜索 + -> 未命中时并发启动术语研究 Agent + -> 优先调用 Wikipedia Tool + -> 必要时调用 SearXNG 搜索 -> 必要时读取网页正文 -> LLM 总结候选术语 -> verifier 检查来源和上下文 @@ -332,6 +346,8 @@ RAG 用于解决视频字幕翻译中的上下文和术语问题,例如游戏 - 单字母变量(如逻辑课里的 `P/Q/R`)默认视为局部变量,不搜索、不写入长期知识库。 - 常见基础术语(如 `truth table`)可以作为本次翻译提示,但默认不自动污染长期知识库。 - 自动正式入库要求 verifier 通过、上下文一致、有有效外部来源,并达到较高置信度;否则进入待审核或跳过。 +- Wikipedia Tool 使用固定 English Wikipedia API,不需要额外配置 wiki 地址;它适合通用百科类专有名词,不替代垂直领域资料库。 +- 可在 `Settings -> Translate` 调整 Agent 并发数和超时时间。并发越高,术语补全越快,但外部 API、LLM 和搜索服务压力也越大。 - Dashboard 中可以点击 RAG Agent 查看每步工具调用、耗时、模型、错误类型和 JSON 输入输出。 PostgreSQL 需要启用 pgvector: diff --git a/docs/PROJECT_SPEC.md b/docs/PROJECT_SPEC.md index bc984b2..804498c 100644 --- a/docs/PROJECT_SPEC.md +++ b/docs/PROJECT_SPEC.md @@ -6,9 +6,41 @@ --- -## 1. 总体拆分:三大模块 + 一个编排层(Orchestrator) +## 1. 当前显著能力 -### 1.1 模块列表与边界 +当前实现已经从“普通字幕翻译流水线”升级为带可观测 RAG Agent 的视频翻译系统,重点能力如下: + +- **字幕翻译 RAG gate**:每个翻译 block 开始前,LLM 先结合当前字幕和前文 summary 判断哪些术语真的需要检索,避免把基础词、局部变量、普通短语都塞进知识库。 +- **PostgreSQL + pgvector 知识库**:保存术语、译法、领域、别名、解释、来源、置信度、状态和 embedding,翻译时按向量相似度召回并注入上下文。 +- **单 Agent + 多 Tool 架构**:术语研究 Agent 统一调度 RAG、Wikipedia、SearXNG、网页读取、总结、验证和写库,不一开始拆成多个松散 Agent。 +- **Wikipedia Tool**:固定使用 English Wikipedia API,适合百科型专有名词和通用背景知识;不用在配置里维护各种不通用的 wiki 地址。 +- **SearXNG Tool**:可接入自建 SearXNG,配置填写 base URL,系统自动请求 `/search?q=...&format=json` 并过滤搜索引擎自身页面等无效结果。 +- **网页正文读取 Tool**:当搜索摘要不足时,Agent 可以打开候选链接抽取正文,再交给 LLM 总结证据。 +- **Verifier 入库保护**:候选术语会经过 verifier 判断来源是否支撑、是否和字幕上下文一致、置信度是否足够;不满足条件时不会污染长期知识库。 +- **并发术语研究**:翻译设置里可调整 RAG Agent 并发数和超时时间,让多个需要查证的术语同时研究,最后合并回当前翻译上下文。 +- **独立 embedding 配置**:翻译模型和 embedding 模型的 API key/base URL 分开;embedding 支持 OpenAI 兼容接口和本地模型目录。 +- **Web 管理与可观测性**:知识库页面支持查看、新增、删除条目;Dashboard 可查看资源监控、正在运行的 Agent、历史 trace 和每步 JSON 输入输出。 + +核心翻译流程: + +```text +字幕 block + 前文 summary + -> RAG gate 判断需要检索的术语 + -> 查询 pgvector 知识库 + -> 未命中时并发启动术语研究 Agent + -> Wikipedia / SearXNG / fetch tools 收集证据 + -> LLM summarize 生成候选译法 + -> verifier 校验来源和上下文 + -> 写入知识库或标记为 pending/context_only/skipped + -> 将命中知识和本次研究结果注入翻译 prompt + -> 输出当前 block 翻译并更新 summary +``` + +--- + +## 2. 总体拆分:三大模块 + 一个编排层(Orchestrator) + +### 2.1 模块列表与边界 1) **subtitle-service(字幕/翻译模块)** - 只负责:从视频/音频生成字幕(转写 + 可选翻译 + 可选润色/术语表 + 排版对齐),并按配置输出字幕文件;可选产生“带字幕的视频”(硬字幕 burn-in / 软字幕 soft-sub)。 @@ -26,7 +58,7 @@ - 负责:任务状态机、队列分发、产物索引(assets)、人工审核、模板管理、凭据加密存储、审计日志与告警。 - 原则:不写平台/ASR 的细节逻辑,只调度模块并记录结果。 -### 1.2 两种落地方式(推荐先 1 后 2) +### 2.2 两种落地方式(推荐先 1 后 2) **方式 A:Monorepo + 进程级解耦(推荐起步)** - 同仓库放四个服务/包,但每个模块都有自己的入口、配置、依赖、接口。 @@ -38,13 +70,13 @@ --- -## 2. 统一“产物契约”(Artifact Contract):模块解耦关键 +## 3. 统一“产物契约”(Artifact Contract):模块解耦关键 模块之间只通过两类东西交互: 1) **DB 里 task 的状态与元信息** 2) **对象存储里的产物 key(S3/MinIO)** -### 2.1 存储 Key 规范(示例) +### 3.1 存储 Key 规范(示例) 建议按 `tenant/account/task_id` 分层,便于多账号与隔离: @@ -62,7 +94,7 @@ - `meta/{task_id}/publish_result.json` - `logs/{task_id}/stage_{name}.log` -### 2.2 产物对象最小字段(建议) +### 3.2 产物对象最小字段(建议) 每个产物在 DB 的 `assets` 表里记录: - `task_id` @@ -75,11 +107,11 @@ --- -## 3. 任务状态机(由 Orchestrator 维护) +## 4. 任务状态机(由 Orchestrator 维护) 建议把任务拆成“内容接入/下载”、“字幕制作”、“发布”三段,每段内部再细分 stage;每个 stage 都是**幂等**、可重试、可断点续跑。 -### 3.1 建议状态枚举(可按需要裁剪) +### 4.1 建议状态枚举(可按需要裁剪) - `CREATED`(任务创建) - `INGESTED`(已拉取元信息/已入库) @@ -96,7 +128,7 @@ - `FAILED`(带 error_code,可重试标记) - `CANCELED` -### 3.2 幂等/重试原则 +### 4.2 幂等/重试原则 - 每个 stage 以 `task_id + stage_name` 作为幂等键;产物存在则跳过或做一致性校验。 - `FAILED` 不代表终止,保存 `error_code` 和 `is_retryable`,支持手动/自动重试。 @@ -104,9 +136,9 @@ --- -## 4. subtitle-service(字幕/翻译模块)设计(重点) +## 5. subtitle-service(字幕/翻译模块)设计(重点) -### 4.1 支持的业务场景 +### 5.1 支持的业务场景 1) **仅生成字幕文件** - 输入:你的本地视频(上传到存储)、或系统已有 `raw/{task_id}/video.mp4` @@ -123,7 +155,7 @@ 4) **双语字幕** - 输出:`subtitle_bi.srt` 或 `subtitle_bi.ass`(上/下行) -### 4.2 内部流水线(可拆分 job) +### 5.2 内部流水线(可拆分 job) **S1 音频提取(FFmpeg)** - 输入:视频文件 @@ -136,7 +168,10 @@ **S3 翻译(可选)** - 输入:`segments.json` 或 `subtitle_src.srt` - 输出:`subtitle_zh.srt` -- 可插拔 provider:LLM / 翻译引擎;支持 `glossary`(术语表)和 `do_not_translate`(专有名词表)。 +- 可插拔 provider:LLM / 翻译引擎。 +- 翻译前可启用 RAG gate:结合当前 block 和前文 summary 发现需要查询的术语,先查 PostgreSQL/pgvector,未命中时由 Agent 调用 Wikipedia、SearXNG 和网页读取工具补充证据。 +- RAG 结果不会无条件写入长期知识库;必须经过 verifier 校验来源、上下文一致性和置信度。 +- 支持手动知识库条目、自动学习条目、context_only 临时提示和 pending 待确认条目。 **S4 字幕排版与对齐(强烈建议做)** 目标:可读性与节奏一致 @@ -149,9 +184,9 @@ - burn-in:FFmpeg + ASS(推荐用 ASS 样式作为输入,效果更好) - soft-sub:容器封装(建议 MKV,MP4 的字幕支持更有限) -### 4.3 对外接口(建议:HTTP + Job 模式) +### 5.3 对外接口(建议:HTTP + Job 模式) -#### 4.3.1 创建字幕任务 +#### 5.3.1 创建字幕任务 `POST /subtitle/jobs` 请求(示例): @@ -180,7 +215,7 @@ { "job_id": "uuid", "status": "queued" } ``` -#### 4.3.2 查询任务状态 +#### 5.3.2 查询任务状态 `GET /subtitle/jobs/{job_id}` 返回(示例): @@ -196,7 +231,7 @@ } ``` -### 4.4 “只出字幕 vs 压制字幕”怎么落地(推荐的参数化) +### 5.4 “只出字幕 vs 压制字幕”怎么落地(推荐的参数化) 在同一条 API 里用 `output.render` 控制即可: - 只出字幕:`burn_in=false, soft_sub=false` @@ -208,11 +243,11 @@ --- -## 5. youtube-ingest(YouTube 接入/抓取模块)设计(与发布解耦) +## 6. youtube-ingest(YouTube 接入/抓取模块)设计(与发布解耦) > 合规核心:**只允许白名单源**(你自己的频道/你已获授权的频道或播放列表/CC 许可明确的源),并把许可信息写入任务,支持人工补证据。 -### 5.1 能力清单 +### 6.1 能力清单 1) **Source 管理(白名单)** - `channel_id` 白名单(own/authorized) @@ -229,9 +264,9 @@ - `manual_only`:只写元信息,不自动下载(最稳、最不触碰平台条款) - `downloader_plugin`:仅对 `license=own/authorized` 开启;写清楚合规责任 -### 5.2 对外接口(示例) +### 6.2 对外接口(示例) -#### 5.2.1 手动入队单条视频 +#### 6.2.1 手动入队单条视频 `POST /youtube/ingest` ```json { @@ -242,7 +277,7 @@ } ``` -#### 5.2.2 扫描频道/播放列表 +#### 6.2.2 扫描频道/播放列表 `POST /youtube/scan` ```json { @@ -259,15 +294,15 @@ - `skipped_duplicates` - `skipped_not_whitelisted` -### 5.3 与 subtitle-service 解耦的点 +### 6.3 与 subtitle-service 解耦的点 - youtube-ingest 只产生 `raw/metadata.json`(以及可选 `raw/video.mp4`) - subtitle-service 不关心 YouTube,只关心 `raw/video.mp4` 是否存在 --- -## 6. bilibili-publisher(B 站上传投稿模块)设计(只接 final 包) +## 7. bilibili-publisher(B 站上传投稿模块)设计(只接 final 包) -### 6.1 输入契约:Publish Package +### 7.1 输入契约:Publish Package `POST /bilibili/publish` ```json @@ -288,7 +323,7 @@ } ``` -### 6.2 输出:发布结果 +### 7.2 输出:发布结果 ```json { @@ -299,7 +334,7 @@ } ``` -### 6.3 关键工程点(不涉及绕过风控) +### 7.3 关键工程点(不涉及绕过风控) - 分片/断点续传:上传过程要可恢复(网络抖动常见) - 速率限制:同账号投稿间隔、每日上限(避免触发平台限制) - 凭据安全:cookie/csrf 不在日志里打印,DB 加密存储 @@ -307,16 +342,18 @@ --- -## 7. Orchestrator(编排层)要做的“产品能力” +## 8. Orchestrator(编排层)要做的“产品能力” -### 7.1 Web 管控台核心功能 +### 8.1 Web 管控台核心功能 - 任务列表:状态筛选、搜索、批量重试/取消 - 任务详情:查看每 stage 的日志与产物;字幕文件在线预览/编辑;投稿 meta 编辑;审核通过/驳回 +- Dashboard:任务状态、队列概览、CPU/内存/Intel GPU 资源监控、RAG Agent 运行状态和历史 trace +- 知识库管理:查看、新增、删除 RAG 条目;触发 embedding 重建 - 模板中心:标题/简介/标签/分区/动态文案模板,支持变量(如 `{source_title}`、`{publish_date}`) - 账号与凭据:B 站凭据录入与轮换提醒;权限控制 - 系统设置:并发数、重试策略、队列开关、通知(Webhook/邮件等) -### 7.2 队列编排建议 +### 8.2 队列编排建议 如果用队列(推荐): - `queue:ingest`(youtube-ingest 相关) - `queue:subtitle`(subtitle-service 相关) @@ -327,7 +364,7 @@ --- -## 8. 最小可运行里程碑(建议) +## 9. 最小可运行里程碑(建议) **M1:本地视频 → 中文字幕 SRT** - Web 创建任务 + 上传本地视频到存储 @@ -344,9 +381,16 @@ **M4:YouTube 授权源扫描** - 只做白名单源发现 + 入队(download 先关) +**M5:RAG Agent 翻译增强** +- PostgreSQL + pgvector 知识库 +- RAG gate + 术语噪声过滤 +- Wikipedia / SearXNG / fetch tools +- verifier 保守入库 +- Dashboard trace 与知识库管理页面 + --- -## 9. 你需要提前确定的 5 个关键选项(决定代码形态) +## 10. 你需要提前确定的 5 个关键选项(决定代码形态) 1) 技术栈:**Python(FastAPI + Celery)** 还是 **Node(NestJS + BullMQ)** 2) ASR:本地(faster-whisper)还是云(可插拔) diff --git a/src/videoroll/apps/subtitle_service/embeddings.py b/src/videoroll/apps/subtitle_service/embeddings.py index 6672e46..32bb3c8 100644 --- a/src/videoroll/apps/subtitle_service/embeddings.py +++ b/src/videoroll/apps/subtitle_service/embeddings.py @@ -255,6 +255,9 @@ def assert_embedding_dimensions(vector: list[float], expected: int) -> None: def embedding_settings_from_translate_settings(settings: dict[str, Any]) -> EmbeddingSettings: provider = normalize_embedding_provider(str(settings.get("rag_embedding_provider") or "openai")) + embedding_api_key = str(settings.get("rag_embedding_api_key") or "").strip() or None + embedding_base_url = str(settings.get("rag_embedding_base_url") or "").strip() + embedding_timeout = float(settings.get("rag_embedding_timeout_seconds") or 60.0) return EmbeddingSettings( provider=provider, model=str(settings.get("rag_embedding_model") or "text-embedding-3-small").strip() or "text-embedding-3-small", @@ -262,10 +265,10 @@ def embedding_settings_from_translate_settings(settings: dict[str, Any]) -> Embe model_dir=str(settings.get("rag_embedding_model_dir") or "/models/embeddings").strip() or "/models/embeddings", device=str(settings.get("rag_embedding_device") or "cpu").strip() or "cpu", openai_config=OpenAIChatConfig( - api_key=str(settings.get("openai_api_key") or "").strip() or None, - base_url=str(settings.get("openai_base_url") or "").strip(), + api_key=embedding_api_key, + base_url=embedding_base_url, model=str(settings.get("rag_embedding_model") or "text-embedding-3-small").strip() or "text-embedding-3-small", temperature=0.0, - timeout_seconds=float(settings.get("openai_timeout_seconds") or 60.0), + timeout_seconds=embedding_timeout, ), ) diff --git a/src/videoroll/apps/subtitle_service/main.py b/src/videoroll/apps/subtitle_service/main.py index 851e757..e88cf9a 100644 --- a/src/videoroll/apps/subtitle_service/main.py +++ b/src/videoroll/apps/subtitle_service/main.py @@ -77,6 +77,7 @@ from videoroll.apps.subtitle_service.embeddings import ( ) from videoroll.apps.subtitle_service.rag import ( build_knowledge_embedding_text, + delete_knowledge_item, list_agent_runs, rebuild_knowledge_embeddings, list_knowledge_items, @@ -343,11 +344,17 @@ def get_translate_settings_view( rag_embedding_dimensions=cfg["rag_embedding_dimensions"], rag_embedding_model_dir=cfg["rag_embedding_model_dir"], rag_embedding_device=cfg["rag_embedding_device"], + rag_embedding_api_key_set=cfg["rag_embedding_api_key_set"], + rag_embedding_base_url=cfg["rag_embedding_base_url"], + rag_embedding_timeout_seconds=cfg["rag_embedding_timeout_seconds"], rag_auto_discover_terms=cfg["rag_auto_discover_terms"], rag_auto_learn_terms=cfg["rag_auto_learn_terms"], + rag_wiki_enabled=cfg["rag_wiki_enabled"], rag_search_enabled=cfg["rag_search_enabled"], rag_search_url=cfg["rag_search_url"], rag_domain=cfg["rag_domain"], + rag_agent_parallelism=cfg["rag_agent_parallelism"], + rag_agent_timeout_seconds=cfg["rag_agent_timeout_seconds"], ) @@ -378,11 +385,17 @@ def put_translate_settings_view( rag_embedding_dimensions=cfg["rag_embedding_dimensions"], rag_embedding_model_dir=cfg["rag_embedding_model_dir"], rag_embedding_device=cfg["rag_embedding_device"], + rag_embedding_api_key_set=cfg["rag_embedding_api_key_set"], + rag_embedding_base_url=cfg["rag_embedding_base_url"], + rag_embedding_timeout_seconds=cfg["rag_embedding_timeout_seconds"], rag_auto_discover_terms=cfg["rag_auto_discover_terms"], rag_auto_learn_terms=cfg["rag_auto_learn_terms"], + rag_wiki_enabled=cfg["rag_wiki_enabled"], rag_search_enabled=cfg["rag_search_enabled"], rag_search_url=cfg["rag_search_url"], rag_domain=cfg["rag_domain"], + rag_agent_parallelism=cfg["rag_agent_parallelism"], + rag_agent_timeout_seconds=cfg["rag_agent_timeout_seconds"], ) @@ -484,6 +497,31 @@ def upsert_knowledge_item_view( return KnowledgeItemUpsertResponse(id=uuid.UUID(item_id)) +@app.delete("/subtitle/knowledge/items/{item_id}") +def delete_knowledge_item_view( + item_id: uuid.UUID, + settings: SubtitleServiceSettings = Depends(get_settings), + db: Session = Depends(get_db), +) -> dict[str, bool]: + try: + deleted = delete_knowledge_item(db, str(item_id)) + db.commit() + except Exception as e: + db.rollback() + if not _is_missing_knowledge_table_error(e): + raise HTTPException(status_code=500, detail=f"knowledge item delete failed: {e}") from e + _ensure_rag_schema(settings) + try: + deleted = delete_knowledge_item(db, str(item_id)) + db.commit() + except Exception as retry_error: + db.rollback() + raise HTTPException(status_code=503, detail=f"knowledge item delete failed after schema migration: {retry_error}") from retry_error + if not deleted: + raise HTTPException(status_code=404, detail="knowledge item not found") + return {"deleted": True} + + @app.post("/subtitle/knowledge/rebuild-embeddings", response_model=KnowledgeEmbeddingRebuildResponse) def rebuild_knowledge_embeddings_view( payload: KnowledgeEmbeddingRebuildRequest, @@ -618,6 +656,12 @@ def test_embedding( cfg["rag_embedding_provider"] = payload.provider if payload.model is not None: cfg["rag_embedding_model"] = payload.model + if payload.api_key is not None: + cfg["rag_embedding_api_key"] = payload.api_key + if payload.base_url is not None: + cfg["rag_embedding_base_url"] = payload.base_url + if payload.timeout_seconds is not None: + cfg["rag_embedding_timeout_seconds"] = payload.timeout_seconds if payload.model_dir is not None: cfg["rag_embedding_model_dir"] = payload.model_dir if payload.dimensions is not None: diff --git a/src/videoroll/apps/subtitle_service/rag.py b/src/videoroll/apps/subtitle_service/rag.py index 047e63f..a0e42ef 100644 --- a/src/videoroll/apps/subtitle_service/rag.py +++ b/src/videoroll/apps/subtitle_service/rag.py @@ -4,14 +4,16 @@ import hashlib import html import ipaddress import json +import math import re import socket import time import uuid +from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError, as_completed from dataclasses import dataclass from datetime import datetime, timezone -from typing import Any, Iterable -from urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlunparse +from typing import Any, Callable, Iterable +from urllib.parse import parse_qsl, quote, urlencode, urljoin, urlparse, urlunparse import httpx from sqlalchemy import text @@ -28,9 +30,31 @@ _SCRIPT_STYLE_RE = re.compile(r"<(script|style|noscript)\b[^>]*>.*?", re.IG _TAG_RE = re.compile(r"<[^>]+>") _ARTICLE_RE = re.compile(r"<(?:article|div)\b[^>]*class=[\"'][^\"']*\bresult\b[^\"']*[\"'][^>]*>.*?", re.IGNORECASE | re.DOTALL) _ANCHOR_RE = re.compile(r"]*href=[\"']([^\"']+)[\"'][^>]*>(.*?)", re.IGNORECASE | re.DOTALL) +_SEARXNG_UI_TEXT_RE = re.compile( + r"(my searxng|about preferences|preferences\s+search syntax|default language|clear\s+search\s+general)", + re.IGNORECASE, +) _SINGLE_LETTER_RE = re.compile(r"^[A-Za-z]$") _MATH_LOGIC_DOMAIN_RE = re.compile(r"(logic|math|数学|逻辑|命题|proposition|propositional)", re.IGNORECASE) _AUTO_APPROVE_CONFIDENCE_THRESHOLD = 0.9 +_SEARCHABLE_GATE_CATEGORIES = { + "proper_noun", + "acronym", + "domain_jargon", + "work_specific_term", + "ambiguous_term", + "community_meme", + "technical_standard", +} +_NON_SEARCH_GATE_CATEGORIES = { + "basic_dictionary", + "common_word", + "local_variable", + "unit_or_number", + "full_sentence", + "generic_action", + "greeting", +} _COMMON_CONTEXT_TRANSLATIONS: dict[str, dict[str, str]] = { "truth table": { "translation": "真值表", @@ -43,6 +67,9 @@ _COMMON_CONTEXT_TRANSLATIONS: dict[str, dict[str, str]] = { "description": "研究命题及命题连接词推理关系的逻辑分支。", }, } +_WIKIPEDIA_API_URL = "https://en.wikipedia.org/w/api.php" +_WIKIPEDIA_SOURCE_NAME = "Wikipedia" +_WIKIPEDIA_USER_AGENT = "VideoRoll-RAG-Agent/1.0 (https://github.com/kelriclink/videoroll)" @dataclass(frozen=True) @@ -59,7 +86,10 @@ class RagSettings: auto_learn_terms: bool = False search_enabled: bool = False search_url: str = "" + wiki_enabled: bool = False domain: str = "" + agent_parallelism: int = 1 + agent_timeout_seconds: float = 120.0 @dataclass(frozen=True) @@ -87,6 +117,14 @@ class RagContext: hits: list[RagHit] +@dataclass(frozen=True) +class AgentResearchResult: + term: str + normalized_term: str + context_card: dict[str, Any] | None = None + hit: RagHit | None = None + + @dataclass(frozen=True) class ToolSpec: tool_name: str @@ -148,6 +186,24 @@ _FETCH_TOOL_SPEC = ToolSpec( cost={"network_requests": 1}, ) +_WIKI_SEARCH_TOOL_SPEC = ToolSpec( + tool_name="wiki_search", + input_schema={"type": "object", "required": ["query", "api_url"], "properties": {"query": {"type": "string"}, "api_url": {"type": "string"}}}, + output_schema={"type": "object", "properties": {"count": {"type": "integer"}, "results": {"type": "array"}}}, + timeout_seconds=20.0, + retry_count=0, + cost={"network_requests": 1}, +) + +_WIKI_READ_TOOL_SPEC = ToolSpec( + tool_name="wiki_read", + input_schema={"type": "object", "required": ["pageid", "api_url"], "properties": {"pageid": {"type": "integer"}, "api_url": {"type": "string"}}}, + output_schema={"type": "object", "properties": {"title": {"type": "string"}, "chars": {"type": "integer"}, "excerpt": {"type": "string"}}}, + timeout_seconds=20.0, + retry_count=0, + cost={"network_requests": 1}, +) + def rag_settings_from_translate_settings(settings: dict[str, Any]) -> RagSettings: return RagSettings( @@ -163,7 +219,10 @@ def rag_settings_from_translate_settings(settings: dict[str, Any]) -> RagSetting auto_learn_terms=bool(settings.get("rag_auto_learn_terms")), search_enabled=bool(settings.get("rag_search_enabled")), search_url=str(settings.get("rag_search_url") or "").strip(), + wiki_enabled=bool(settings.get("rag_wiki_enabled")), domain=str(settings.get("rag_domain") or "").strip(), + agent_parallelism=max(1, min(8, int(settings.get("rag_agent_parallelism") or 1))), + agent_timeout_seconds=max(10.0, min(900.0, float(settings.get("rag_agent_timeout_seconds") or 120.0))), ) @@ -173,6 +232,42 @@ def normalize_term(term: str) -> str: return s +def normalize_wiki_api_url(raw_url: str) -> str: + raw = str(raw_url or "").strip() + if not raw: + return "" + if "://" not in raw: + raw = f"https://{raw}" + parsed = urlparse(raw) + path = parsed.path.rstrip("/") + if not path: + path = "/w/api.php" + elif path.endswith("/api.php"): + path = path + elif "/wiki/" in path: + prefix = path.split("/wiki/", 1)[0].rstrip("/") + path = f"{prefix}/w/api.php" if prefix else "/w/api.php" + elif path.endswith("/wiki"): + prefix = path[: -len("/wiki")].rstrip("/") + path = f"{prefix}/w/api.php" if prefix else "/w/api.php" + else: + path = f"{path}/w/api.php" + return urlunparse(parsed._replace(path=path, query="", fragment="")) + + +def _wiki_page_url(api_url: str, title: str) -> str: + parsed = urlparse(normalize_wiki_api_url(api_url)) + path = parsed.path + if path.endswith("/w/api.php"): + root = path[: -len("/w/api.php")] + elif path.endswith("/api.php"): + root = path[: -len("/api.php")] + else: + root = "" + page_path = f"{root.rstrip('/')}/wiki/{quote(str(title or '').strip().replace(' ', '_'))}" + return urlunparse(parsed._replace(path=page_path, query="", fragment="")) + + def build_knowledge_embedding_text( *, item_type: str, @@ -251,6 +346,16 @@ def _collapse_text(value: str, *, limit: int = 12000) -> str: return text_value[:limit] +def _context_for_llm(text_value: str, *, previous_summary: str = "", limit: int = 9000) -> str: + current = str(text_value or "").strip() + summary = str(previous_summary or "").strip() + if summary: + combined = f"前文摘要:\n{summary[:800]}\n\n当前字幕 block:\n{current}" + else: + combined = current + return combined[:limit] + + def _search_endpoint_from_base(search_base_url: str) -> str: raw = str(search_base_url or "").strip() if not raw: @@ -294,6 +399,9 @@ def _normalize_result_url(raw_url: str, *, base_url: str) -> str: def _is_search_engine_internal_url(url: str, *, search_url: str) -> bool: parsed = urlparse(str(url or "").strip()) search_parsed = urlparse(_search_endpoint_from_base(search_url)) + hostname = (parsed.hostname or "").lower() + if hostname in {"searx.space", "www.searx.space"}: + return True if not parsed.netloc or not search_parsed.netloc: return False if parsed.netloc.lower() != search_parsed.netloc.lower(): @@ -362,6 +470,8 @@ def _parse_search_json(data: Any) -> list[dict[str, Any]]: snippet = str(item.get("snippet") or item.get("content") or item.get("description") or "").strip() if not title and not snippet: continue + if _SEARXNG_UI_TEXT_RE.search(" ".join([title, snippet])): + continue out.append({"title": _strip_html(title), "url": url, "snippet": _strip_html(snippet)[:800]}) return out @@ -377,6 +487,8 @@ def _filter_search_results(results: list[dict[str, Any]], *, search_url: str) -> continue if _is_search_engine_internal_url(url, search_url=search_url): continue + if _SEARXNG_UI_TEXT_RE.search(" ".join([title, snippet])): + continue if title.lower() in {"about", "preferences", "search syntax"} and "searxng" in snippet.lower(): continue if url in seen: @@ -393,8 +505,6 @@ def _parse_search_html(html_value: str, *, base_url: str) -> list[dict[str, Any] chunks = _ARTICLE_RE.findall(source) if not chunks: chunks = re.findall(r"]*>.*?", source, flags=re.IGNORECASE | re.DOTALL) - if not chunks: - chunks = source.split(" list[str] return out -def should_research_term(term: str, *, domain: str = "", context: str = "") -> dict[str, Any]: +def should_research_term( + term: str, + *, + domain: str = "", + context: str = "", + gate_item: dict[str, Any] | None = None, +) -> dict[str, Any]: clean = str(term or "").strip() norm = normalize_term(clean) domain_text = str(domain or "").strip() + category_hint = str((gate_item or {}).get("category") or "").strip() + scope_hint = str((gate_item or {}).get("scope") or "").strip() + need_rag_hint = (gate_item or {}).get("need_rag") + need_search_hint = (gate_item or {}).get("need_search") if not clean or not norm: return { "should_research": False, @@ -668,11 +788,45 @@ def should_research_term(term: str, *, domain: str = "", context: str = "") -> d "category": "common_word", "reason": "common function word", } + if category_hint in _NON_SEARCH_GATE_CATEGORIES: + return { + "should_research": False, + "should_persist": False, + "category": category_hint, + "scope": scope_hint or "none", + "reason": str((gate_item or {}).get("reason") or "RAG gate classified this as not requiring external knowledge"), + } + if need_rag_hint is False: + return { + "should_research": False, + "should_persist": False, + "category": category_hint or "gate_rejected", + "scope": scope_hint or "none", + "reason": str((gate_item or {}).get("reason") or "RAG gate decided this term does not need RAG"), + } + if category_hint and category_hint not in _SEARCHABLE_GATE_CATEGORIES and need_search_hint is not True: + return { + "should_research": False, + "should_persist": scope_hint in {"task", "series", "global"}, + "category": category_hint, + "scope": scope_hint or "task", + "reason": str((gate_item or {}).get("reason") or "RAG gate did not classify this as external-search-worthy"), + } + if need_search_hint is False and category_hint not in _SEARCHABLE_GATE_CATEGORIES: + return { + "should_research": False, + "should_persist": scope_hint in {"task", "series", "global"}, + "category": category_hint or "task_context", + "scope": scope_hint or "task", + "reason": str((gate_item or {}).get("reason") or "RAG gate requested local/task context only"), + } return { "should_research": True, - "should_persist": True, - "category": "research", - "reason": "term may need external context", + "should_persist": scope_hint != "task", + "category": category_hint or "research", + "scope": scope_hint or "global", + "reason": str((gate_item or {}).get("reason") or "term may need external context"), + "need_search": bool(need_search_hint) if need_search_hint is not None else True, } @@ -696,12 +850,14 @@ def discover_terms_openai( *, target_lang: str, domain_hint: str, + previous_summary: str = "", config: OpenAIChatConfig, client: httpx.Client | None = None, ) -> list[dict[str, Any]]: source = str(text_value or "").strip() if not source: return [] + llm_context = _context_for_llm(source, previous_summary=previous_summary, limit=8000) data = request_openai_json_object( config=config, system_prompt="You identify translation terms in subtitles. Return ONLY valid JSON.", @@ -714,7 +870,7 @@ def discover_terms_openai( "- reason 用中文简短说明为什么它是术语。\n" f"- 目标语言:{target_lang or 'zh'}\n" f"- 领域提示:{domain_hint or '未知'}\n\n" - f"字幕片段:\n{source[:8000]}\n\n" + f"上下文:\n{llm_context}\n\n" '输出 JSON:{"terms":[{"term":"","domain":"","reason":""}]}' ), client=client, @@ -742,6 +898,100 @@ def discover_terms_openai( return out[:20] +def pretranslation_rag_gate_openai( + text_value: str, + *, + target_lang: str, + domain_hint: str, + existing_terms: list[dict[str, Any]] | None = None, + previous_summary: str = "", + config: OpenAIChatConfig, + client: httpx.Client | None = None, +) -> list[dict[str, Any]]: + source = str(text_value or "").strip() + if not source: + return [] + llm_context = _context_for_llm(source, previous_summary=previous_summary, limit=9000) + compact_existing = [ + { + "term": str(item.get("term") or "")[:160], + "translation": str(item.get("translation") or "")[:160], + "domain": str(item.get("domain") or "")[:120], + "score": item.get("score"), + } + for item in (existing_terms or [])[:20] + if isinstance(item, dict) + ] + data = request_openai_json_object( + config=config, + system_prompt="You are a pre-translation RAG gate. Return ONLY valid JSON.", + user_prompt=( + "请在翻译这个字幕 block 之前,判断哪些词/短语如果不查资料或不查本地知识库,可能会翻错。\n" + "这不是普通 NER,也不是词典抽取;目标是筛出高价值 RAG 候选。\n\n" + "只返回满足以下至少一项的候选:\n" + "- 专有名词、人名、组织、作品、角色、品牌;\n" + "- 缩写或代号,且在当前语境可能有特定含义;\n" + "- 游戏/动漫/技术/社区黑话、梗、固定译法;\n" + "- 作品设定、领域标准、框架名、工具名、论文/协议/技术名词;\n" + "- 普通词但当前语境可能有特殊含义或歧义。\n\n" + "不要返回这些:\n" + "- 基础词典词、常见动词/名词、寒暄句;\n" + "- 数字、单位、尺寸、普通食材;\n" + "- true/false/condition/validate/pixel/salt 这类模型能稳定翻译的基础词;\n" + "- P/Q/R/x/y 这类局部变量;\n" + "- 整句普通表达,除非它是固定梗或固定术语。\n\n" + "category 只能使用:proper_noun, acronym, domain_jargon, work_specific_term, " + "ambiguous_term, community_meme, technical_standard, task_context, basic_dictionary, " + "common_word, local_variable, unit_or_number, full_sentence, generic_action, greeting。\n" + "scope 只能使用:global, series, task, none。\n" + "need_rag 表示是否需要本地 RAG/上下文辅助;need_search 表示本地未命中时是否值得外部搜索。\n" + "如果你认为没有值得查的词,返回空数组。不要为了凑数返回基础词。返回数量不设上限,但必须是高价值项。\n\n" + f"目标语言:{target_lang or 'zh'}\n" + f"领域提示:{domain_hint or '未知'}\n" + f"已有本地命中 JSON:\n{json.dumps(compact_existing, ensure_ascii=False)}\n\n" + f"上下文:\n{llm_context}\n\n" + '输出 JSON:{"terms":[{"term":"","category":"","domain":"","need_rag":true,' + '"need_search":true,"scope":"global","priority":0.0,"reason":""}]}' + ), + client=client, + ) + terms = data.get("terms") + if not isinstance(terms, list): + return [] + out: list[dict[str, Any]] = [] + seen: set[str] = set() + for item in terms: + if not isinstance(item, dict): + continue + term = str(item.get("term") or "").strip() + norm = normalize_term(term) + if not term or norm in seen: + continue + seen.add(norm) + try: + priority = float(item.get("priority") or 0.0) + except Exception: + priority = 0.0 + category = str(item.get("category") or "").strip() + scope = str(item.get("scope") or "").strip() + need_rag_raw = item.get("need_rag") + need_search_raw = item.get("need_search") + out.append( + { + "term": term, + "category": category, + "domain": str(item.get("domain") or domain_hint or "").strip(), + "need_rag": True if need_rag_raw is None else _json_bool(need_rag_raw), + "need_search": (category in _SEARCHABLE_GATE_CATEGORIES) if need_search_raw is None else _json_bool(need_search_raw), + "scope": scope if scope in {"global", "series", "task", "none"} else "global", + "priority": max(0.0, min(1.0, priority)), + "reason": str(item.get("reason") or "").strip(), + } + ) + out.sort(key=lambda x: float(x.get("priority") or 0.0), reverse=True) + return out + + def _dedupe_search_queries(queries: Iterable[str], *, limit: int = 3) -> list[str]: out: list[str] = [] seen: set[str] = set() @@ -920,28 +1170,38 @@ def fetch_search_evidence( with httpx.Client( timeout=timeout_seconds, follow_redirects=True, - headers={"User-Agent": "VideoRoll-RAG-Agent/1.0"}, + headers={"User-Agent": "VideoRoll-RAG-Agent/1.0", "Accept": "application/json, text/html;q=0.9,*/*;q=0.8"}, ) as client: for query in search_queries: json_url = _search_url_with_params(endpoint, query=query, json_format=True) html_url = _search_url_with_params(endpoint, query=query, json_format=False) started = time.perf_counter() filtered_results: list[dict[str, Any]] = [] + json_error = "" try: - resp = client.get(json_url) - resp.raise_for_status() - content_type = resp.headers.get("content-type", "") - if "json" in content_type.lower(): - filtered_results = _parse_search_json(resp.json()) - else: - try: - filtered_results = _parse_search_json(resp.json()) - except Exception: - filtered_results = _parse_search_html(resp.text, base_url=str(resp.url)) - if not filtered_results: - resp = client.get(html_url) + try: + resp = client.get(json_url) resp.raise_for_status() - filtered_results = _parse_search_html(resp.text, base_url=str(resp.url)) + content_type = resp.headers.get("content-type", "") + if "json" in content_type.lower(): + filtered_results = _parse_search_json(resp.json()) + else: + try: + filtered_results = _parse_search_json(resp.json()) + except Exception: + filtered_results = _parse_search_html(resp.text, base_url=str(resp.url)) + except Exception as e: + json_error = str(e)[:300] + filtered_results = [] + if not filtered_results: + try: + resp = client.get(html_url) + resp.raise_for_status() + filtered_results = _parse_search_html(resp.text, base_url=str(resp.url)) + except Exception as e: + if json_error: + raise RuntimeError(f"json search failed: {json_error}; html search failed: {e}") from e + raise filtered_results = _filter_search_results(filtered_results, search_url=endpoint) compact_results = [ {"title": r.get("title", ""), "url": r.get("url", ""), "snippet": str(r.get("snippet") or "")[:240]} @@ -950,7 +1210,7 @@ def fetch_search_evidence( step = ToolResult( spec=_SEARCH_TOOL_SPEC, input={"query": query, "url": json_url}, - output={"count": len(filtered_results), "results": compact_results}, + output={"count": len(filtered_results), "results": compact_results, "json_error": json_error}, ok=True, duration_ms=_duration_ms(started), ).to_step(action="search") @@ -1137,6 +1397,197 @@ def fetch_search_evidence( return [] +def fetch_wikipedia_evidence( + term: str, + *, + domain: str, + queries: list[str] | None = None, + timeout_seconds: float = 20.0, + max_pages: int = 3, + db: Session | None = None, + agent_run_id: str | None = None, +) -> list[dict[str, Any]]: + clean_term = str(term or "").strip() + if not clean_term: + return [] + raw_queries = [clean_term] + raw_queries.extend(queries or []) + search_queries = _dedupe_search_queries(raw_queries, limit=4) + if not search_queries: + return [] + + search_results: list[dict[str, Any]] = [] + seen_pages: set[str] = set() + try: + with httpx.Client( + timeout=timeout_seconds, + follow_redirects=True, + headers={"User-Agent": _WIKIPEDIA_USER_AGENT, "Accept": "application/json"}, + ) as client: + for query in search_queries: + started = time.perf_counter() + params = { + "action": "query", + "list": "search", + "srsearch": query, + "srlimit": 5, + "format": "json", + "formatversion": "2", + } + try: + resp = client.get(_WIKIPEDIA_API_URL, params=params) + resp.raise_for_status() + data = resp.json() + raw_items = data.get("query", {}).get("search", []) if isinstance(data, dict) else [] + results: list[dict[str, Any]] = [] + if isinstance(raw_items, list): + for raw in raw_items[:5]: + if not isinstance(raw, dict): + continue + title = _strip_html(str(raw.get("title") or "")).strip() + pageid = raw.get("pageid") + snippet = _strip_html(str(raw.get("snippet") or "")).strip() + if not title or pageid is None: + continue + page_key = str(pageid) + result = { + "title": title, + "pageid": pageid, + "url": _wiki_page_url(_WIKIPEDIA_API_URL, title), + "snippet": snippet[:800], + "source": _WIKIPEDIA_SOURCE_NAME, + "tool": "wiki", + } + results.append(result) + if page_key not in seen_pages: + seen_pages.add(page_key) + search_results.append(result) + compact_results = [ + {"title": r["title"], "pageid": r["pageid"], "url": r["url"], "snippet": str(r.get("snippet") or "")[:240]} + for r in results[:5] + ] + step = ToolResult( + spec=_WIKI_SEARCH_TOOL_SPEC, + input={"query": query, "api_url": _WIKIPEDIA_API_URL}, + output={"count": len(results), "results": compact_results}, + ok=True, + duration_ms=_duration_ms(started), + ).to_step(action="wiki_search") + step.update({"query": query, "api_url": _WIKIPEDIA_API_URL, "count": len(results), "results": compact_results}) + if db is not None: + _append_agent_step(db, agent_run_id, step) + except Exception as e: + step = ToolResult( + spec=_WIKI_SEARCH_TOOL_SPEC, + input={"query": query, "api_url": _WIKIPEDIA_API_URL}, + output={"count": 0, "results": []}, + ok=False, + duration_ms=_duration_ms(started), + error_type=type(e).__name__, + error=str(e)[:300], + ).to_step(action="wiki_search_failed") + step.update({"query": query, "api_url": _WIKIPEDIA_API_URL}) + if db is not None: + _append_agent_step(db, agent_run_id, step) + continue + if len(search_results) >= 8: + break + + if not search_results: + if db is not None: + _append_agent_step( + db, + agent_run_id, + { + "kind": "tool", + "action": "wiki_no_results", + "tool": "wiki_search", + "tool_name": "wiki_search", + "api_url": _WIKIPEDIA_API_URL, + "queries": search_queries, + }, + ) + return [] + + out: list[dict[str, Any]] = [] + read_limit = max(1, min(5, int(max_pages))) + for item in search_results[:read_limit]: + pageid = item.get("pageid") + title = str(item.get("title") or "").strip() + page: dict[str, Any] = { + "title": title, + "url": str(item.get("url") or "").strip(), + "snippet": str(item.get("snippet") or "").strip()[:800], + "source": _WIKIPEDIA_SOURCE_NAME, + "tool": "wiki", + } + started = time.perf_counter() + params = { + "action": "query", + "prop": "extracts", + "exintro": "1", + "explaintext": "1", + "pageids": str(pageid), + "format": "json", + "formatversion": "2", + } + try: + resp = client.get(_WIKIPEDIA_API_URL, params=params) + resp.raise_for_status() + data = resp.json() + pages = data.get("query", {}).get("pages", []) if isinstance(data, dict) else [] + wiki_page = pages[0] if isinstance(pages, list) and pages and isinstance(pages[0], dict) else {} + extract = _collapse_text(str(wiki_page.get("extract") or ""), limit=5000) + if extract: + page["content"] = extract + if wiki_page.get("title"): + page["title"] = str(wiki_page.get("title") or title) + page["url"] = _wiki_page_url(_WIKIPEDIA_API_URL, page["title"]) + step = ToolResult( + spec=_WIKI_READ_TOOL_SPEC, + input={"pageid": int(pageid), "api_url": _WIKIPEDIA_API_URL, "title": title}, + output={"title": page["title"], "chars": len(extract), "excerpt": extract[:360]}, + ok=True, + duration_ms=_duration_ms(started), + ).to_step(action="wiki_read") + step.update({"pageid": pageid, "title": page["title"], "url": page["url"], "chars": len(extract), "excerpt": extract[:360]}) + if db is not None: + _append_agent_step(db, agent_run_id, step) + except Exception as e: + page["fetch_error"] = str(e)[:300] + step = ToolResult( + spec=_WIKI_READ_TOOL_SPEC, + input={"pageid": pageid, "api_url": _WIKIPEDIA_API_URL, "title": title}, + output={"title": title, "chars": 0, "excerpt": ""}, + ok=False, + duration_ms=_duration_ms(started), + error_type=type(e).__name__, + error=str(e)[:300], + ).to_step(action="wiki_read_failed") + step.update({"pageid": pageid, "title": title, "url": page["url"]}) + if db is not None: + _append_agent_step(db, agent_run_id, step) + if page.get("snippet") or page.get("content"): + out.append(page) + return out + except Exception: + if db is not None: + _append_agent_step( + db, + agent_run_id, + { + "kind": "tool", + "action": "wiki_failed", + "tool": "wiki_search", + "tool_name": "wiki_search", + "api_url": _WIKIPEDIA_API_URL, + "queries": search_queries, + "error_type": "unexpected_error", + }, + ) + return [] + + def explain_term_from_evidence_openai( *, term: str, @@ -1508,6 +1959,581 @@ def exact_term_hits( return [_row_to_hit(row) for row in rows] +def _research_discovered_term( + db: Session, + *, + item: dict[str, Any], + target_lang: str, + rag_settings: RagSettings, + embedding_settings: EmbeddingSettings, + chat_config: OpenAIChatConfig, + text_value: str, + llm_context: str, + previous_summary: str, + existing_term_cards: list[dict[str, Any]], + gate_duration_ms: int | None, + task_id: str | None, + subtitle_job_id: str | None, +) -> AgentResearchResult | None: + term = str(item.get("term") or "").strip() + norm = normalize_term(term) + if not term or not norm: + return None + + domain_hint = str(item.get("domain") or rag_settings.domain or "").strip() + query = " ".join([p for p in [domain_hint, term] if p]).strip() + agent_run_id: str | None = None + try: + agent_run_id = _start_agent_run( + db, + term=term, + domain=domain_hint, + target_lang=target_lang, + query=query, + task_id=task_id, + subtitle_job_id=subtitle_job_id, + ) + _append_llm_step( + db, + agent_run_id, + action="pretranslation_rag_gate", + config=chat_config, + input_value={ + "domain": rag_settings.domain, + "context_excerpt": text_value[:500], + "previous_summary": str(previous_summary or "").strip()[:500], + "existing_hit_count": len(existing_term_cards), + }, + output_value={ + "term": term, + "domain": domain_hint, + "category": str(item.get("category") or ""), + "need_rag": bool(item.get("need_rag")), + "need_search": bool(item.get("need_search")), + "scope": str(item.get("scope") or ""), + "priority": float(item.get("priority") or 0.0), + "reason": str(item.get("reason") or "").strip(), + }, + duration_ms=gate_duration_ms, + ) + except Exception: + db.rollback() + agent_run_id = None + + research_policy = should_research_term(term, domain=domain_hint, context=text_value, gate_item=item) + _append_agent_step( + db, + agent_run_id, + { + "kind": "policy", + "action": "term_research_policy", + "term": term, + "domain": domain_hint, + "decision": research_policy, + }, + ) + if not research_policy.get("should_research"): + context_card: dict[str, Any] | None = None + if research_policy.get("category") == "context_only" and research_policy.get("translation"): + context_card = _context_only_term_card(term, research_policy, target_lang=target_lang, domain=domain_hint) + elif research_policy.get("scope") in {"task", "series"} or item.get("need_rag"): + hint = str(item.get("translation") or item.get("translation_hint") or "").strip() + if hint: + context_card = { + "term": term, + "translation": hint, + "domain": domain_hint, + "aliases": [], + "description": str(item.get("reason") or research_policy.get("reason") or "").strip(), + "confidence": float(item.get("priority") or 0.0) or 0.7, + "score": float(item.get("priority") or 0.0) or 0.7, + "sources": [], + "target_lang": target_lang, + "status": str(research_policy.get("scope") or "task_context"), + } + _finish_agent_run( + db, + agent_run_id, + status="skipped", + result={ + "term": term, + "domain": domain_hint, + "knowledge_status": str(research_policy.get("category") or "skipped"), + "failure_category": str(research_policy.get("category") or "skipped"), + "reason": str(research_policy.get("reason") or ""), + }, + ) + return AgentResearchResult(term=term, normalized_term=norm, context_card=context_card) + + if not bool(research_policy.get("need_search", True)): + _finish_agent_run( + db, + agent_run_id, + status="skipped", + result={ + "term": term, + "domain": domain_hint, + "knowledge_status": "local_rag_only", + "failure_category": "no_external_search_requested", + "reason": str(research_policy.get("reason") or ""), + }, + ) + 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: + started = time.perf_counter() + try: + search_queries = generate_search_queries_openai( + term=term, + context=llm_context, + target_lang=target_lang, + domain_hint=domain_hint, + config=chat_config, + max_queries=3, + ) + _append_llm_step( + db, + agent_run_id, + action="generate_search_queries", + config=chat_config, + input_value={ + "term": term, + "domain": domain_hint, + "context_excerpt": text_value[:500], + "previous_summary": str(previous_summary or "").strip()[:500], + }, + output_value={"queries": search_queries}, + duration_ms=_duration_ms(started), + ) + except Exception as e: + search_queries = _fallback_search_queries(term, domain=domain_hint, target_lang=target_lang, limit=3) + _append_llm_step( + db, + agent_run_id, + action="generate_search_queries_failed", + config=chat_config, + duration_ms=_duration_ms(started), + error=str(e)[:300], + error_type=type(e).__name__, + ) + _append_agent_step( + db, + agent_run_id, + { + "kind": "policy", + "action": "search_query_fallback", + "queries": search_queries, + "reason": "LLM query generation failed; using deterministic fallback queries.", + }, + ) + evidence = [] + if rag_settings.wiki_enabled: + 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: + 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, + ) + if not evidence: + _finish_agent_run( + db, + agent_run_id, + status="failed", + result={ + "term": term, + "domain": domain_hint, + "search_queries": search_queries, + "failure_category": "no_search_evidence", + "tools": { + "wikipedia": bool(rag_settings.wiki_enabled), + "search": bool(rag_settings.search_enabled), + }, + }, + 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", + ) + 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: + failure_category = failure_category or "no_valid_external_source" + _finish_agent_run( + db, + agent_run_id, + status="skipped", + result={ + "term": str(explained.get("term") or term), + "translation": str(explained.get("translation") or ""), + "domain": str(explained.get("domain") or domain_hint or ""), + "confidence": final_confidence, + "knowledge_status": "context_only" if failure_category == "context_only" else "not_written", + "failure_category": failure_category or "verify_rejected", + "verification": verification, + "valid_source_count": len(valid_sources), + "search_queries": search_queries, + }, + error=str(verification.get("reason") or "verifier rejected glossary entry")[:4000], + ) + return AgentResearchResult(term=term, normalized_term=norm) + + status = ( + "auto_approved" + if bool(verification.get("should_auto_approve")) and final_confidence >= _AUTO_APPROVE_CONFIDENCE_THRESHOLD + else "pending" + ) + source_list = [x for x in explained.get("sources") or [] if isinstance(x, dict)] or valid_sources + try: + _append_agent_step( + db, + agent_run_id, + { + "kind": "policy", + "action": "knowledge_write_decision", + "term": str(explained.get("term") or term), + "translation": str(explained.get("translation") or ""), + "confidence": final_confidence, + "status": status, + "auto_threshold": _AUTO_APPROVE_CONFIDENCE_THRESHOLD, + "valid_source_count": len(valid_sources), + }, + ) + emb_text = build_knowledge_embedding_text( + item_type="term", + term=str(explained.get("term") or term), + translation=str(explained.get("translation") or ""), + domain=str(explained.get("domain") or ""), + aliases=[str(x) for x in explained.get("aliases") or []], + description=str(explained.get("description") or ""), + ) + emb = embed_text(emb_text, settings=embedding_settings) + assert_embedding_dimensions(emb, rag_settings.embedding_dimensions) + learned_id = upsert_knowledge_item( + db, + item_type="term", + target_lang=target_lang, + term=str(explained.get("term") or term), + translation=str(explained.get("translation") or ""), + domain=str(explained.get("domain") or rag_settings.domain or ""), + aliases=[str(x) for x in explained.get("aliases") or []], + description=str(explained.get("description") or ""), + sources=source_list, + confidence=final_confidence, + status=status, + created_by="agent", + embedding=emb, + embedding_model=embedding_model_key(rag_settings), + ) + db.commit() + _finish_agent_run( + db, + agent_run_id, + status="succeeded", + result={ + "term": str(explained.get("term") or term), + "translation": str(explained.get("translation") or ""), + "domain": str(explained.get("domain") or rag_settings.domain or ""), + "confidence": final_confidence, + "knowledge_status": status, + "verification": verification, + "valid_source_count": len(valid_sources), + "search_queries": search_queries, + "sources": source_list[:5], + }, + knowledge_item_id=learned_id, + ) + if status == "auto_approved": + return AgentResearchResult( + term=term, + normalized_term=norm, + hit=RagHit( + id=learned_id, + item_type="term", + term=str(explained.get("term") or term), + translation=str(explained.get("translation") or ""), + target_lang=target_lang, + domain=str(explained.get("domain") or rag_settings.domain or ""), + aliases=[str(x) for x in explained.get("aliases") or []], + title="", + content="", + description=str(explained.get("description") or ""), + sources=source_list, + confidence=final_confidence, + status=status, + score=final_confidence, + ), + ) + return AgentResearchResult(term=term, normalized_term=norm) + except Exception as e: + db.rollback() + _finish_agent_run( + db, + agent_run_id, + status="failed", + result={ + "term": str(explained.get("term") or term), + "translation": str(explained.get("translation") or ""), + "confidence": final_confidence, + "failure_category": "knowledge_save_failed", + }, + error=f"knowledge save failed: {e}", + ) + return None + + +def _run_research_agent( + *, + db: Session, + session_factory: Callable[[], Session] | None, + item: dict[str, Any], + target_lang: str, + rag_settings: RagSettings, + embedding_settings: EmbeddingSettings, + chat_config: OpenAIChatConfig, + text_value: str, + llm_context: str, + previous_summary: str, + existing_term_cards: list[dict[str, Any]], + gate_duration_ms: int | None, + task_id: str | None, + subtitle_job_id: str | None, +) -> AgentResearchResult | None: + if session_factory is None: + return _research_discovered_term( + db, + item=item, + target_lang=target_lang, + rag_settings=rag_settings, + embedding_settings=embedding_settings, + chat_config=chat_config, + text_value=text_value, + llm_context=llm_context, + previous_summary=previous_summary, + existing_term_cards=existing_term_cards, + gate_duration_ms=gate_duration_ms, + task_id=task_id, + subtitle_job_id=subtitle_job_id, + ) + worker_db = session_factory() + try: + return _research_discovered_term( + worker_db, + item=item, + target_lang=target_lang, + rag_settings=rag_settings, + embedding_settings=embedding_settings, + chat_config=chat_config, + text_value=text_value, + llm_context=llm_context, + previous_summary=previous_summary, + existing_term_cards=existing_term_cards, + gate_duration_ms=gate_duration_ms, + task_id=task_id, + subtitle_job_id=subtitle_job_id, + ) + finally: + worker_db.close() + + +def _run_research_agents( + *, + db: Session, + session_factory: Callable[[], Session] | None, + items: list[dict[str, Any]], + target_lang: str, + rag_settings: RagSettings, + embedding_settings: EmbeddingSettings, + chat_config: OpenAIChatConfig, + text_value: str, + llm_context: str, + previous_summary: str, + existing_term_cards: list[dict[str, Any]], + gate_duration_ms: int | None, + task_id: str | None, + subtitle_job_id: str | None, +) -> list[AgentResearchResult]: + if not items: + return [] + parallelism = max(1, min(8, int(rag_settings.agent_parallelism or 1))) + if session_factory is None: + parallelism = 1 + timeout_seconds = max(10.0, min(900.0, float(rag_settings.agent_timeout_seconds or 120.0))) + if parallelism <= 1 or len(items) <= 1: + out: list[AgentResearchResult] = [] + deadline = time.monotonic() + timeout_seconds * max(1, len(items)) + for item in items: + if time.monotonic() >= deadline: + break + result = _run_research_agent( + db=db, + session_factory=None, + item=item, + target_lang=target_lang, + rag_settings=rag_settings, + embedding_settings=embedding_settings, + chat_config=chat_config, + text_value=text_value, + llm_context=llm_context, + previous_summary=previous_summary, + existing_term_cards=existing_term_cards, + gate_duration_ms=gate_duration_ms, + task_id=task_id, + subtitle_job_id=subtitle_job_id, + ) + if result is not None: + out.append(result) + return out + + out: list[AgentResearchResult] = [] + max_workers = min(parallelism, len(items)) + executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="rag-agent") + try: + futures = [ + executor.submit( + _run_research_agent, + db=db, + session_factory=session_factory, + item=item, + target_lang=target_lang, + rag_settings=rag_settings, + embedding_settings=embedding_settings, + chat_config=chat_config, + text_value=text_value, + llm_context=llm_context, + previous_summary=previous_summary, + existing_term_cards=existing_term_cards, + gate_duration_ms=gate_duration_ms, + task_id=task_id, + subtitle_job_id=subtitle_job_id, + ) + for item in items + ] + try: + for future in as_completed(futures, timeout=timeout_seconds * max(1, math.ceil(len(items) / max_workers))): + try: + result = future.result() + except Exception: + result = None + if result is not None: + out.append(result) + except FuturesTimeoutError: + for future in futures: + future.cancel() + finally: + executor.shutdown(wait=False, cancel_futures=True) + return out + + def build_rag_context( db: Session, *, @@ -1516,6 +2542,8 @@ def build_rag_context( rag_settings: RagSettings, embedding_settings: EmbeddingSettings, chat_config: OpenAIChatConfig, + previous_summary: str = "", + session_factory: Callable[[], Session] | None = None, task_id: str | None = None, subtitle_job_id: str | None = None, ) -> RagContext: @@ -1525,9 +2553,31 @@ def build_rag_context( text_value = "\n".join([str(s.text or "").strip() for s in segments if str(s.text or "").strip()]) if not text_value: return RagContext(term_cards=[], knowledge_cards=[], hits=[]) + llm_context = _context_for_llm(text_value, previous_summary=previous_summary, limit=9000) 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) + for hit in exact_term_hits( db, text_value=text_value, @@ -1535,41 +2585,83 @@ 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) try: - query_embedding = embed_text(text_value[:8000], settings=embedding_settings) - assert_embedding_dimensions(query_embedding, rag_settings.embedding_dimensions) - for hit in search_knowledge( - db, - query_embedding=query_embedding, - target_lang=target_lang, - domain=rag_settings.domain, - embedding_model=embedding_model_key(rag_settings), - top_k=rag_settings.top_k, - min_score=rag_settings.min_score, - ): - if hit.id not in seen_ids: - seen_ids.add(hit.id) - hits.append(hit) + rag_query_text = text_value[:8000] + if gate_terms is not None: + rag_query_text = "\n".join( + [ + " | ".join( + [ + str(item.get("term") or "").strip(), + str(item.get("category") or "").strip(), + str(item.get("reason") or "").strip(), + ] + ).strip(" |") + for item in gate_terms + if isinstance(item, dict) and str(item.get("term") or "").strip() + ] + ) + if rag_query_text.strip(): + query_embedding = embed_text(rag_query_text[:8000], settings=embedding_settings) + assert_embedding_dimensions(query_embedding, rag_settings.embedding_dimensions) + for hit in search_knowledge( + db, + query_embedding=query_embedding, + target_lang=target_lang, + domain=rag_settings.domain, + embedding_model=embedding_model_key(rag_settings), + top_k=rag_settings.top_k, + min_score=rag_settings.min_score, + ): + if gate_terms is not None and hit.item_type == "term" 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) except Exception: query_embedding = [] context_only_cards: list[dict[str, Any]] = [] if rag_settings.auto_discover_terms: - try: - discovered = discover_terms_openai( - text_value, - target_lang=target_lang, - domain_hint=rag_settings.domain, - config=chat_config, - ) - except Exception: - discovered = [] known_norms = {normalize_term(h.term) for h in hits if h.term} + existing_term_cards = [ + { + "term": hit.term, + "translation": hit.translation, + "domain": hit.domain, + "score": round(hit.score, 4), + } + for hit in hits + if hit.item_type == "term" + ] + discovered = gate_terms + if discovered is None: + fallback_started = time.perf_counter() + try: + discovered = discover_terms_openai( + text_value, + target_lang=target_lang, + domain_hint=rag_settings.domain, + previous_summary=previous_summary, + config=chat_config, + ) + for item in discovered: + item.setdefault("need_rag", True) + item.setdefault("need_search", True) + item.setdefault("scope", "global") + item.setdefault("category", "legacy_discovery") + item.setdefault("priority", 0.5) + except Exception: + discovered = [] + gate_duration_ms = _duration_ms(fallback_started) + research_items: list[dict[str, Any]] = [] for item in discovered: term = str(item.get("term") or "").strip() norm = normalize_term(term) @@ -1577,352 +2669,32 @@ def build_rag_context( continue if not rag_settings.auto_learn_terms: continue - domain_hint = str(item.get("domain") or rag_settings.domain or "").strip() - query = " ".join([p for p in [domain_hint, term] if p]).strip() - agent_run_id: str | None = None - try: - agent_run_id = _start_agent_run( - db, - term=term, - domain=domain_hint, - target_lang=target_lang, - query=query, - task_id=task_id, - subtitle_job_id=subtitle_job_id, - ) - _append_llm_step( - db, - agent_run_id, - action="term_discovered", - config=chat_config, - output_value={ - "term": term, - "domain": domain_hint, - "reason": str(item.get("reason") or "").strip(), - }, - ) - except Exception: - db.rollback() - agent_run_id = None - - research_policy = should_research_term(term, domain=domain_hint, context=text_value) - _append_agent_step( - db, - agent_run_id, - { - "kind": "policy", - "action": "term_research_policy", - "term": term, - "domain": domain_hint, - "decision": research_policy, - }, - ) - if not research_policy.get("should_research"): - if research_policy.get("category") == "context_only" and research_policy.get("translation"): - context_only_cards.append(_context_only_term_card(term, research_policy, target_lang=target_lang, domain=domain_hint)) - _finish_agent_run( - db, - agent_run_id, - status="skipped", - result={ - "term": term, - "domain": domain_hint, - "knowledge_status": str(research_policy.get("category") or "skipped"), - "failure_category": str(research_policy.get("category") or "skipped"), - "reason": str(research_policy.get("reason") or ""), - }, - ) - known_norms.add(norm) - continue - - search_queries: list[str] = [] - if rag_settings.search_enabled: - started = time.perf_counter() - try: - search_queries = generate_search_queries_openai( - term=term, - context=text_value, - target_lang=target_lang, - domain_hint=domain_hint, - config=chat_config, - max_queries=3, - ) - _append_llm_step( - db, - agent_run_id, - action="generate_search_queries", - config=chat_config, - input_value={"term": term, "domain": domain_hint, "context_excerpt": text_value[:500]}, - output_value={"queries": search_queries}, - duration_ms=_duration_ms(started), - ) - except Exception as e: - search_queries = _fallback_search_queries(term, domain=domain_hint, target_lang=target_lang, limit=3) - _append_llm_step( - db, - agent_run_id, - action="generate_search_queries_failed", - config=chat_config, - duration_ms=_duration_ms(started), - error=str(e)[:300], - error_type=type(e).__name__, - ) - _append_agent_step( - db, - agent_run_id, - { - "kind": "policy", - "action": "search_query_fallback", - "queries": search_queries, - "reason": "LLM query generation failed; using deterministic fallback queries.", - }, - ) - evidence = [] - if rag_settings.search_enabled: - evidence = fetch_search_evidence( - term, - domain=domain_hint, - search_url=rag_settings.search_url, - queries=search_queries, - context=text_value, - target_lang=target_lang, - config=chat_config, - db=db, - agent_run_id=agent_run_id, - ) - if not evidence: - _finish_agent_run( - db, - agent_run_id, - status="failed", - result={ - "term": term, - "domain": domain_hint, - "search_queries": search_queries, - "failure_category": "no_search_evidence", - }, - error="no search evidence", - ) - continue - try: - summarize_started = time.perf_counter() - explained = explain_term_from_evidence_openai( - term=term, - context=text_value, - 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", - ) - continue - - 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=text_value, - 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: - failure_category = failure_category or "no_valid_external_source" - _finish_agent_run( - db, - agent_run_id, - status="skipped", - result={ - "term": str(explained.get("term") or term), - "translation": str(explained.get("translation") or ""), - "domain": str(explained.get("domain") or domain_hint or ""), - "confidence": final_confidence, - "knowledge_status": "context_only" if failure_category == "context_only" else "not_written", - "failure_category": failure_category or "verify_rejected", - "verification": verification, - "valid_source_count": len(valid_sources), - "search_queries": search_queries, - }, - error=str(verification.get("reason") or "verifier rejected glossary entry")[:4000], - ) - known_norms.add(norm) - continue - - status = ( - "auto_approved" - if bool(verification.get("should_auto_approve")) and final_confidence >= _AUTO_APPROVE_CONFIDENCE_THRESHOLD - else "pending" - ) - source_list = [x for x in explained.get("sources") or [] if isinstance(x, dict)] or valid_sources - try: - _append_agent_step( - db, - agent_run_id, - { - "kind": "policy", - "action": "knowledge_write_decision", - "term": str(explained.get("term") or term), - "translation": str(explained.get("translation") or ""), - "confidence": final_confidence, - "status": status, - "auto_threshold": _AUTO_APPROVE_CONFIDENCE_THRESHOLD, - "valid_source_count": len(valid_sources), - }, - ) - emb_text = build_knowledge_embedding_text( - item_type="term", - term=str(explained.get("term") or term), - translation=str(explained.get("translation") or ""), - domain=str(explained.get("domain") or ""), - aliases=[str(x) for x in explained.get("aliases") or []], - description=str(explained.get("description") or ""), - ) - emb = embed_text(emb_text, settings=embedding_settings) - assert_embedding_dimensions(emb, rag_settings.embedding_dimensions) - learned_id = upsert_knowledge_item( - db, - item_type="term", - target_lang=target_lang, - term=str(explained.get("term") or term), - translation=str(explained.get("translation") or ""), - domain=str(explained.get("domain") or rag_settings.domain or ""), - aliases=[str(x) for x in explained.get("aliases") or []], - description=str(explained.get("description") or ""), - sources=source_list, - confidence=final_confidence, - status=status, - created_by="agent", - embedding=emb, - embedding_model=embedding_model_key(rag_settings), - ) - db.commit() - _finish_agent_run( - db, - agent_run_id, - status="succeeded", - result={ - "term": str(explained.get("term") or term), - "translation": str(explained.get("translation") or ""), - "domain": str(explained.get("domain") or rag_settings.domain or ""), - "confidence": final_confidence, - "knowledge_status": status, - "verification": verification, - "valid_source_count": len(valid_sources), - "search_queries": search_queries, - "sources": source_list[:5], - }, - knowledge_item_id=learned_id, - ) - known_norms.add(norm) - if status == "auto_approved": - hits.append( - RagHit( - id=learned_id, - item_type="term", - term=str(explained.get("term") or term), - translation=str(explained.get("translation") or ""), - target_lang=target_lang, - domain=str(explained.get("domain") or rag_settings.domain or ""), - aliases=[str(x) for x in explained.get("aliases") or []], - title="", - content="", - description=str(explained.get("description") or ""), - sources=source_list, - confidence=final_confidence, - status=status, - score=final_confidence, - ) - ) - except Exception as e: - db.rollback() - _finish_agent_run( - db, - agent_run_id, - status="failed", - result={ - "term": str(explained.get("term") or term), - "translation": str(explained.get("translation") or ""), - "confidence": final_confidence, - "failure_category": "knowledge_save_failed", - }, - error=f"knowledge save failed: {e}", - ) + research_items.append(item) + known_norms.add(norm) + for result in _run_research_agents( + db=db, + session_factory=session_factory, + items=research_items, + target_lang=target_lang, + rag_settings=rag_settings, + embedding_settings=embedding_settings, + chat_config=chat_config, + text_value=text_value, + llm_context=llm_context, + previous_summary=previous_summary, + existing_term_cards=existing_term_cards, + gate_duration_ms=gate_duration_ms, + task_id=task_id, + subtitle_job_id=subtitle_job_id, + ): + if result.normalized_term: + known_norms.add(result.normalized_term) + if result.context_card: + context_only_cards.append(result.context_card) + if result.hit and result.hit.id not in seen_ids: + seen_ids.add(result.hit.id) + hits.append(result.hit) + discovered = [] term_cards: list[dict[str, Any]] = list(context_only_cards) knowledge_cards: list[dict[str, Any]] = [] @@ -2065,6 +2837,14 @@ def list_knowledge_items(db: Session, *, item_type: str | None = None, status: s return out +def delete_knowledge_item(db: Session, item_id: str) -> bool: + result = db.execute( + text("DELETE FROM translation_knowledge_items WHERE id = CAST(:id AS uuid)"), + {"id": str(item_id or "").strip()}, + ) + return int(getattr(result, "rowcount", 0) or 0) > 0 + + def list_agent_runs(db: Session, *, status: str | None = None, limit: int = 50) -> list[dict[str, Any]]: clauses = ["1=1"] params: dict[str, Any] = {"limit": max(1, min(100, int(limit)))} diff --git a/src/videoroll/apps/subtitle_service/schemas.py b/src/videoroll/apps/subtitle_service/schemas.py index f026700..43baa6c 100644 --- a/src/videoroll/apps/subtitle_service/schemas.py +++ b/src/videoroll/apps/subtitle_service/schemas.py @@ -210,11 +210,17 @@ class TranslateSettingsRead(BaseModel): rag_embedding_dimensions: int = 1536 rag_embedding_model_dir: str = "/models/embeddings" rag_embedding_device: str = "cpu" + rag_embedding_api_key_set: bool = False + rag_embedding_base_url: str = "" + rag_embedding_timeout_seconds: float = 60.0 rag_auto_discover_terms: bool = False rag_auto_learn_terms: bool = False + rag_wiki_enabled: bool = False rag_search_enabled: bool = False rag_search_url: str = "" rag_domain: str = "" + rag_agent_parallelism: int = 1 + rag_agent_timeout_seconds: float = 120.0 class TranslateSettingsUpdate(BaseModel): @@ -240,11 +246,17 @@ class TranslateSettingsUpdate(BaseModel): rag_embedding_dimensions: Optional[int] = Field(default=None, ge=1, le=4096) rag_embedding_model_dir: Optional[str] = None rag_embedding_device: Optional[str] = None + rag_embedding_api_key: Optional[str] = None + rag_embedding_base_url: Optional[str] = None + rag_embedding_timeout_seconds: Optional[float] = None rag_auto_discover_terms: Optional[bool] = None rag_auto_learn_terms: Optional[bool] = None + rag_wiki_enabled: Optional[bool] = None rag_search_enabled: Optional[bool] = None rag_search_url: Optional[str] = None rag_domain: Optional[str] = None + rag_agent_parallelism: Optional[int] = Field(default=None, ge=1, le=8) + rag_agent_timeout_seconds: Optional[float] = Field(default=None, ge=10.0, le=900.0) class TranslateTestRequest(BaseModel): @@ -357,6 +369,9 @@ class EmbeddingTestRequest(BaseModel): text: str = "hello world" provider: Optional[str] = None model: Optional[str] = None + api_key: Optional[str] = None + base_url: Optional[str] = None + timeout_seconds: Optional[float] = None model_dir: Optional[str] = None dimensions: Optional[int] = None device: Optional[str] = None diff --git a/src/videoroll/apps/subtitle_service/translate_settings_store.py b/src/videoroll/apps/subtitle_service/translate_settings_store.py index 1132a48..602de68 100644 --- a/src/videoroll/apps/subtitle_service/translate_settings_store.py +++ b/src/videoroll/apps/subtitle_service/translate_settings_store.py @@ -40,6 +40,15 @@ def get_translate_settings(db: Session, defaults: SubtitleServiceSettings) -> di api_key = decrypt_str(api_key_enc) except Exception: api_key = "" + embedding_openai = _as_dict(stored.get("rag_embedding_openai")) + embedding_api_key = "" + embedding_api_key_enc = embedding_openai.get("api_key_enc") + if isinstance(embedding_api_key_enc, str) and embedding_api_key_enc.strip(): + try: + embedding_api_key = decrypt_str(embedding_api_key_enc) + except Exception: + embedding_api_key = "" + default_embedding_base_url = str(defaults.rag_embedding_base_url or "").strip() return { "default_provider": str(stored.get("default_provider") or defaults.translate_default_provider), @@ -64,13 +73,24 @@ def get_translate_settings(db: Session, defaults: SubtitleServiceSettings) -> di "rag_embedding_dimensions": int(stored.get("rag_embedding_dimensions") or defaults.rag_embedding_dimensions), "rag_embedding_model_dir": str(stored.get("rag_embedding_model_dir") or defaults.rag_embedding_model_dir), "rag_embedding_device": str(stored.get("rag_embedding_device") or defaults.rag_embedding_device), + "rag_embedding_api_key": embedding_api_key or str(defaults.rag_embedding_api_key or ""), + "rag_embedding_api_key_set": bool(embedding_api_key), + "rag_embedding_base_url": normalize_openai_base_url(str(embedding_openai.get("base_url") or default_embedding_base_url)), + "rag_embedding_timeout_seconds": float( + embedding_openai.get("timeout_seconds") + or stored.get("rag_embedding_timeout_seconds") + or defaults.rag_embedding_timeout_seconds + ), "rag_auto_discover_terms": bool( stored.get("rag_auto_discover_terms") if "rag_auto_discover_terms" in stored else defaults.rag_auto_discover_terms ), "rag_auto_learn_terms": bool(stored.get("rag_auto_learn_terms") if "rag_auto_learn_terms" in stored else defaults.rag_auto_learn_terms), + "rag_wiki_enabled": bool(stored.get("rag_wiki_enabled") if "rag_wiki_enabled" in stored else False), "rag_search_enabled": bool(stored.get("rag_search_enabled") if "rag_search_enabled" in stored else defaults.rag_search_enabled), "rag_search_url": str(stored.get("rag_search_url") or defaults.rag_search_url), "rag_domain": str(stored.get("rag_domain") or defaults.rag_domain), + "rag_agent_parallelism": int(stored.get("rag_agent_parallelism") or 1), + "rag_agent_timeout_seconds": float(stored.get("rag_agent_timeout_seconds") or 120.0), } @@ -93,11 +113,15 @@ def update_translate_settings(db: Session, defaults: SubtitleServiceSettings, up "rag_embedding_dimensions", "rag_embedding_model_dir", "rag_embedding_device", + "rag_embedding_timeout_seconds", "rag_auto_discover_terms", "rag_auto_learn_terms", + "rag_wiki_enabled", "rag_search_enabled", "rag_search_url", "rag_domain", + "rag_agent_parallelism", + "rag_agent_timeout_seconds", ]: if key not in update: continue @@ -137,6 +161,21 @@ def update_translate_settings(db: Session, defaults: SubtitleServiceSettings, up stored["openai"] = openai + embedding_openai = dict(_as_dict(stored.get("rag_embedding_openai"))) + if "rag_embedding_api_key" in update: + key = update.get("rag_embedding_api_key") + if key is not None: + key = str(key).strip() + if not key: + embedding_openai.pop("api_key_enc", None) + else: + embedding_openai["api_key_enc"] = encrypt_str(key) + if "rag_embedding_base_url" in update and update.get("rag_embedding_base_url") is not None: + embedding_openai["base_url"] = normalize_openai_base_url(str(update.get("rag_embedding_base_url") or "")) + if "rag_embedding_timeout_seconds" in update and update.get("rag_embedding_timeout_seconds") is not None: + embedding_openai["timeout_seconds"] = update.get("rag_embedding_timeout_seconds") + stored["rag_embedding_openai"] = embedding_openai + # Normalize a few types. try: if "default_batch_size" in stored and stored["default_batch_size"] is not None: @@ -160,7 +199,22 @@ def update_translate_settings(db: Session, defaults: SubtitleServiceSettings, up stored["rag_embedding_dimensions"] = max(1, min(4096, int(stored.get("rag_embedding_dimensions") or defaults.rag_embedding_dimensions))) except Exception: stored["rag_embedding_dimensions"] = defaults.rag_embedding_dimensions - for bool_key in ["rag_enabled", "rag_auto_discover_terms", "rag_auto_learn_terms", "rag_search_enabled"]: + try: + stored["rag_agent_parallelism"] = max(1, min(8, int(stored.get("rag_agent_parallelism") or 1))) + except Exception: + stored["rag_agent_parallelism"] = 1 + try: + stored["rag_agent_timeout_seconds"] = max(10.0, min(900.0, float(stored.get("rag_agent_timeout_seconds") or 120.0))) + except Exception: + stored["rag_agent_timeout_seconds"] = 120.0 + try: + stored["rag_embedding_timeout_seconds"] = max( + 1.0, + min(600.0, float(stored.get("rag_embedding_timeout_seconds") or defaults.rag_embedding_timeout_seconds)), + ) + except Exception: + stored["rag_embedding_timeout_seconds"] = defaults.rag_embedding_timeout_seconds + for bool_key in ["rag_enabled", "rag_auto_discover_terms", "rag_auto_learn_terms", "rag_wiki_enabled", "rag_search_enabled"]: if bool_key in stored: stored[bool_key] = bool(stored[bool_key]) provider = str(stored.get("rag_embedding_provider") or defaults.rag_embedding_provider).strip().lower() @@ -168,6 +222,13 @@ def update_translate_settings(db: Session, defaults: SubtitleServiceSettings, up for str_key in ["rag_embedding_model", "rag_embedding_model_dir", "rag_embedding_device", "rag_search_url", "rag_domain"]: if str_key in stored and stored[str_key] is not None: stored[str_key] = str(stored[str_key]).strip() + if "base_url" in embedding_openai: + embedding_openai["base_url"] = normalize_openai_base_url(str(embedding_openai.get("base_url") or "")) + try: + if "timeout_seconds" in embedding_openai and embedding_openai["timeout_seconds"] is not None: + embedding_openai["timeout_seconds"] = max(1.0, min(600.0, float(embedding_openai.get("timeout_seconds")))) + except Exception: + embedding_openai["timeout_seconds"] = defaults.rag_embedding_timeout_seconds try: if "openai_temperature" in update and update["openai_temperature"] is not None: openai["temperature"] = float(openai.get("temperature")) diff --git a/src/videoroll/apps/subtitle_service/worker.py b/src/videoroll/apps/subtitle_service/worker.py index 41536ae..2d6b73f 100644 --- a/src/videoroll/apps/subtitle_service/worker.py +++ b/src/videoroll/apps/subtitle_service/worker.py @@ -1374,7 +1374,7 @@ def process_job(self: Any, job_id: str) -> dict[str, str]: ) def _rag_context_provider(batch_segments: list[Segment], start_idx: int, summary: str) -> dict[str, Any] | None: - del start_idx, summary + del start_idx if not rag_settings.enabled: return None ctx = build_rag_context( @@ -1384,6 +1384,8 @@ def process_job(self: Any, job_id: str) -> dict[str, str]: rag_settings=rag_settings, embedding_settings=rag_embedding_settings, chat_config=rag_chat_config, + previous_summary=summary, + session_factory=lambda: get_sessionmaker(settings.database_url)(), task_id=str(task.id), subtitle_job_id=str(job.id), ) diff --git a/src/videoroll/config.py b/src/videoroll/config.py index 87f656c..d343748 100644 --- a/src/videoroll/config.py +++ b/src/videoroll/config.py @@ -89,6 +89,9 @@ class SubtitleServiceSettings(CommonSettings): rag_embedding_dimensions: int = Field(1536, alias="SUBTITLE_RAG_EMBEDDING_DIMENSIONS") rag_embedding_model_dir: str = Field("/models/embeddings", alias="SUBTITLE_RAG_EMBEDDING_MODEL_DIR") rag_embedding_device: str = Field("cpu", alias="SUBTITLE_RAG_EMBEDDING_DEVICE") + rag_embedding_api_key: str = Field("", alias="SUBTITLE_RAG_EMBEDDING_API_KEY") + rag_embedding_base_url: str = Field("", alias="SUBTITLE_RAG_EMBEDDING_BASE_URL") + rag_embedding_timeout_seconds: float = Field(60.0, alias="SUBTITLE_RAG_EMBEDDING_TIMEOUT_SECONDS") rag_auto_discover_terms: bool = Field(False, alias="SUBTITLE_RAG_AUTO_DISCOVER_TERMS") rag_auto_learn_terms: bool = Field(False, alias="SUBTITLE_RAG_AUTO_LEARN_TERMS") rag_search_enabled: bool = Field(False, alias="SUBTITLE_RAG_SEARCH_ENABLED") diff --git a/src/web/src/App.tsx b/src/web/src/App.tsx index a212aa8..9feaf54 100644 --- a/src/web/src/App.tsx +++ b/src/web/src/App.tsx @@ -19,6 +19,7 @@ import SettingsBilibiliPage from "./pages/SettingsBilibiliPage"; import SettingsAutoPage from "./pages/SettingsAutoPage"; import SettingsReviewPage from "./pages/SettingsReviewPage"; import RenderQueuePage from "./pages/RenderQueuePage"; +import KnowledgeBasePage from "./pages/KnowledgeBasePage"; function NavItem({ to, label }: { to: string; label: string }) { return ( @@ -130,6 +131,7 @@ export default function App() { + @@ -160,6 +162,7 @@ export default function App() { } /> } /> } /> + } /> } /> } /> } /> diff --git a/src/web/src/pages/DashboardPage.tsx b/src/web/src/pages/DashboardPage.tsx index 327c73a..98e2a6c 100644 --- a/src/web/src/pages/DashboardPage.tsx +++ b/src/web/src/pages/DashboardPage.tsx @@ -218,7 +218,7 @@ function AgentRunDetail({ run, onClose }: { run: AgentRun; onClose: () => void } {knowledgeStatus ?
knowledge: {knowledgeStatus}
: null} {failureCategory ?
failure: {failureCategory}
: null} {run.knowledge_item_id ? ( - + item: {run.knowledge_item_id} ) : null} diff --git a/src/web/src/pages/KnowledgeBasePage.tsx b/src/web/src/pages/KnowledgeBasePage.tsx new file mode 100644 index 0000000..f2be615 --- /dev/null +++ b/src/web/src/pages/KnowledgeBasePage.tsx @@ -0,0 +1,277 @@ +import { useCallback, useEffect, useMemo, useState } from "react"; +import { Link } from "react-router-dom"; +import { useConfirm } from "../components/feedbackContext"; +import { Button, DataTable, EmptyState, PageHeader, Section } from "../components/ui"; +import { fetchJson } from "../lib/http"; +import { SUBTITLE_SERVICE_URL } from "../lib/urls"; + +type KnowledgeItem = { + id: string; + item_type: string; + term: string; + translation: string; + target_lang: string; + domain: string; + aliases: unknown[]; + title: string; + content: string; + description: string; + sources: unknown[]; + confidence: number; + status: string; + created_by: string; + usage_count: number; + embedding_model: string; + updated_at: string; +}; + +function formatDate(value: string) { + return value ? new Date(value).toLocaleString() : "-"; +} + +function statusClass(status: string) { + if (status === "auto_approved" || status === "approved") return "border-emerald-200 bg-emerald-50 text-emerald-800"; + if (status === "pending") return "border-amber-200 bg-amber-50 text-amber-800"; + if (status === "archived") return "border-slate-200 bg-slate-50 text-slate-600"; + return "border-slate-200 bg-slate-50 text-slate-700"; +} + +export default function KnowledgeBasePage() { + const confirm = useConfirm(); + const [items, setItems] = useState([]); + const [error, setError] = useState(null); + const [busy, setBusy] = useState(false); + const [itemTypeFilter, setItemTypeFilter] = useState(""); + const [statusFilter, setStatusFilter] = useState(""); + + const [itemType, setItemType] = useState<"term" | "document">("term"); + const [targetLang, setTargetLang] = useState("zh"); + const [term, setTerm] = useState(""); + const [translation, setTranslation] = useState(""); + const [domain, setDomain] = useState(""); + const [title, setTitle] = useState(""); + const [description, setDescription] = useState(""); + const [content, setContent] = useState(""); + const [status, setStatus] = useState("approved"); + const [confidence, setConfidence] = useState(1); + + const query = useMemo(() => { + const params = new URLSearchParams(); + params.set("limit", "200"); + if (itemTypeFilter) params.set("item_type", itemTypeFilter); + if (statusFilter) params.set("status", statusFilter); + return params.toString(); + }, [itemTypeFilter, statusFilter]); + + const refresh = useCallback(async () => { + setError(null); + try { + const rows = await fetchJson(`${SUBTITLE_SERVICE_URL}/subtitle/knowledge/items?${query}`); + setItems(rows); + } catch (e: unknown) { + setError(e instanceof Error ? e.message : String(e)); + } + }, [query]); + + useEffect(() => { + refresh(); + }, [refresh]); + + async function saveItem() { + setBusy(true); + setError(null); + try { + await fetchJson(`${SUBTITLE_SERVICE_URL}/subtitle/knowledge/items`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + item_type: itemType, + target_lang: targetLang, + term, + translation, + domain, + title, + content, + description, + confidence, + status, + created_by: "manual", + }), + }); + setTerm(""); + setTranslation(""); + setTitle(""); + setDescription(""); + setContent(""); + await refresh(); + } catch (e: unknown) { + setError(e instanceof Error ? e.message : String(e)); + } finally { + setBusy(false); + } + } + + async function deleteItem(item: KnowledgeItem) { + const ok = await confirm({ + title: "删除知识条目", + message: `确定删除「${item.term || item.title || item.id}」吗?此操作不会删除任务,只会移除知识库条目。`, + confirmLabel: "删除", + tone: "danger", + }); + if (!ok) return; + setBusy(true); + setError(null); + try { + await fetchJson(`${SUBTITLE_SERVICE_URL}/subtitle/knowledge/items/${item.id}`, { method: "DELETE" }); + await refresh(); + } catch (e: unknown) { + setError(e instanceof Error ? e.message : String(e)); + } finally { + setBusy(false); + } + } + + return ( +
+ + + RAG 设置 + + + + } + /> + {error ?
{error}
: null} + +
+
+
+
新增条目
+
手动添加的条目会直接生成向量;如果 embedding 配置不可用,术语仍可保存但不会参与向量检索。
+
+
+ + +
+
+
+ + + {itemType === "term" ? ( + <> + + + + ) : ( + + )} + +