From ecbaf444253abcd6f01314aaa728d638c6be79cc Mon Sep 17 00:00:00 2001 From: kelriclink Date: Sat, 4 Jul 2026 22:58:48 +0800 Subject: [PATCH] feat: improve RAG agent search configuration --- README.md | 17 + docs/PROJECT_SPEC.md | 4 + .../apps/subtitle_service/agent_runtime.py | 228 +++++ src/videoroll/apps/subtitle_service/main.py | 14 + src/videoroll/apps/subtitle_service/rag.py | 824 ++++++++++++++++-- .../apps/subtitle_service/retrieval.py | 126 +++ .../apps/subtitle_service/schemas.py | 14 + .../translate_settings_store.py | 84 +- src/videoroll/config.py | 7 + src/web/src/pages/SettingsTranslatePage.tsx | 148 ++++ tests/test_translation_rag.py | 102 +++ 11 files changed, 1487 insertions(+), 81 deletions(-) create mode 100644 src/videoroll/apps/subtitle_service/agent_runtime.py create mode 100644 src/videoroll/apps/subtitle_service/retrieval.py diff --git a/README.md b/README.md index f784389..239578b 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,7 @@ - **主 Agent + 子 Agent 翻译 RAG**：字幕翻译前由主 Agent 扫描当前 block 和前文 summary，决定哪些术语需要研究；每个术语由独立子 Agent 维护 observation/evidence 上下文并选择工具。 - **PostgreSQL + pgvector 知识库**：术语、译法、来源、置信度和使用次数统一入库；支持向量检索、手动新增/删除条目、重建 embedding。 - **真正的工具调用循环**：子 Agent 可按需调用 `rag_lookup`、`wiki_search`、`search_web`、`fetch_url`、`finish`，而不是固定按 wiki -> search 的流水线机械执行。 +- **轻量 Agent Runtime**：内部运行时统一管理 tool schema、LLM/tool/fetch 预算、状态迁移、错误类型和 trace 字段，方便后续扩展 Dictionary/Wiki/GitHub/Bilibili 等工具。 - **Wikipedia Tool + SearXNG Tool**：固定 English Wikipedia API 用于通用百科知识；自建 SearXNG 用于补充网络搜索；搜索摘要不足时可继续读取网页正文。 - **并发术语研究 Agent**：可在 `Settings -> Translate` 设置并发数和超时；主 Agent 等待子 Agent 返回结构化结果后再合并进当前 block 翻译。 - **保守入库策略**：summarize -> verify -> write_rag；只有 verifier 判断来源支撑、上下文一致且置信度足够时才正式入库，否则标记为 pending、context_only 或 skipped。 @@ -371,8 +372,10 @@ RAG 用于解决视频字幕翻译中的上下文和术语问题，例如游戏 - 自动正式入库要求 verifier 通过、上下文一致、有有效外部来源，并达到较高置信度；否则进入待审核或跳过。 - Wikipedia Tool 使用固定 English Wikipedia API，不需要额外配置 wiki 地址；它适合通用百科类专有名词，不替代垂直领域资料库。 - SearXNG Tool 使用 base URL 配置，系统自动拼接 `/search?q=...&format=json`，并过滤搜索引擎 About/Preferences 等内部页面。 +- SearXNG 支持在 Web UI 配置 `categories`、`engines`、`fallback_engines`、`language`、`safesearch`、`time_range` 和 `pageno`；默认使用 `categories=general`、`language=all`、`safesearch=0`、`pageno=1`，默认搜索为空时 fallback 到 `bing,baidu`。 - 可在 `Settings -> Translate` 调整 Agent 并发数、Agent 超时、大模型失败重试次数。并发越高，术语补全越快，但外部 API、LLM 和搜索服务压力也越大。 - Dashboard 中可以点击主 Agent 查看下属子 Agent；点击子 Agent 可查看它自己的工具调用对话流。 +- 每个 Agent trace 都会带 `event_id`、`span_id`、`status`、工具 schema、耗时、错误类型和预算计数，便于排查哪个节点失败或耗时过长。 子 Agent 返回主 Agent 的数据是结构化结果，而不是把整段 trace 直接塞回 prompt： @@ -401,6 +404,20 @@ https://search.example.com 系统会自动请求 `/search?q=...&format=json`，不需要在配置里手写 `?q=`。 +可选搜索参数在 `Settings -> Translate -> RAG Agent` 配置： + +```text +categories=general +engines= +fallback_engines=bing,baidu +language=all +safesearch=0 +time_range= +pageno=1 +``` + +`engines` 为空时使用 SearXNG 实例默认引擎；常见值包括 `bing`、`baidu`、`brave`、`duckduckgo`、`wikipedia`、`arxiv`、`pubmed`、`github`、`stackoverflow`、`youtube` 等。最终可用引擎以你自己的 SearXNG `settings.yml` 为准。 + Embedding 支持两种方式： - OpenAI 兼容 embedding：填写 provider 支持的 embedding 模型名、base URL 和独立 API key。 diff --git a/docs/PROJECT_SPEC.md b/docs/PROJECT_SPEC.md index 1875881..06df28a 100644 --- a/docs/PROJECT_SPEC.md +++ b/docs/PROJECT_SPEC.md @@ -14,8 +14,10 @@ - **PostgreSQL + pgvector 知识库**：保存术语、译法、领域、别名、解释、来源、置信度、状态和 embedding，翻译时按向量相似度召回并注入上下文。 - **主 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 和写库安全控制。 +- **轻量 Agent Runtime**：借鉴 LangGraph/OpenAI Agents SDK/PydanticAI 一类框架的核心做法，但不引入重依赖；内部统一 tool schema、结构化输出校验、预算控制、状态迁移、错误分类和 trace 事件。 - **Wikipedia Tool**：固定使用 English Wikipedia API，适合百科型专有名词和通用背景知识；不用在配置里维护各种不通用的 wiki 地址。 - **SearXNG Tool**：可接入自建 SearXNG，配置填写 base URL，系统自动请求 `/search?q=...&format=json` 并过滤搜索引擎自身页面等无效结果。 +- **SearXNG 参数配置**：翻译设置支持配置 `categories`、`engines`、`fallback_engines`、`language`、`safesearch`、`time_range` 和 `pageno`。不指定 `engines` 时使用实例默认引擎，默认搜索为空时使用 fallback 引擎再次尝试。 - **网页正文读取 Tool**：当搜索摘要不足时，Agent 可以打开候选链接抽取正文，再交给 LLM 总结证据。 - **Verifier 入库保护**：候选术语会经过 verifier 判断来源是否支撑、是否和字幕上下文一致、置信度是否足够；不满足条件时不会污染长期知识库。 - **并发术语研究**：翻译设置里可调整 RAG Agent 并发数和超时时间，让多个需要查证的术语同时研究；子 Agent 结束后返回结构化 `AgentResearchResult`，由主 Agent 合并回当前翻译上下文。 @@ -47,6 +49,8 @@ - `knowledge_cards`：文档型背景知识。 - `hits`：原始知识库命中，用于记录匹配和使用次数。 +Agent trace 是规范化事件流，每步包含 `event_id`、`span_id`、`kind`、`action`、`status`、`duration_ms`、`error_type`、工具 schema、预算计数和输入输出摘要。Dashboard 以这些字段展示主 Agent/子 Agent 树和每步 JSON。 + --- ## 2. 总体拆分：三大模块 + 一个编排层（Orchestrator） diff --git a/src/videoroll/apps/subtitle_service/agent_runtime.py b/src/videoroll/apps/subtitle_service/agent_runtime.py new file mode 100644 index 0000000..114fc36 --- /dev/null +++ b/src/videoroll/apps/subtitle_service/agent_runtime.py @@ -0,0 +1,228 @@ +from __future__ import annotations + +import json +import time +import uuid +from dataclasses import dataclass, field +from typing import Any, Callable, Generic, Literal, TypeVar + +from pydantic import BaseModel, Field, ValidationError + + +AgentStepKind = Literal["agent", "llm", "tool", "policy", "retrieval", "error"] + + +class AgentBudget(BaseModel): + max_llm_calls: int = Field(default=12, ge=0, le=100) + max_tool_calls: int = Field(default=12, ge=0, le=100) + max_fetch_calls: int = Field(default=4, ge=0, le=50) + timeout_seconds: float = Field(default=120.0, ge=1.0, le=3600.0) + + +class AgentDecision(BaseModel): + action: Literal["rag_lookup", "wiki_search", "search_web", "fetch_url", "finish"] + query: str = "" + url: str = "" + reason: str = "" + final_answer_ready: bool = False + + +class SearchQueryPlan(BaseModel): + queries: list[str] = Field(default_factory=list) + + +class GlossaryCandidate(BaseModel): + term: str + translation: str + domain: str = "" + aliases: list[str] = Field(default_factory=list) + description: str = "" + sources: list[dict[str, Any]] = Field(default_factory=list) + confidence: float = Field(default=0.0, ge=0.0, le=1.0) + + +class VerificationResult(BaseModel): + supported: bool = False + context_consistent: bool = False + should_write: bool = False + should_auto_approve: bool = False + confidence: float = Field(default=0.0, ge=0.0, le=1.0) + reason: str = "" + failure_category: str = "" + + +class ToolSpec(BaseModel): + name: str + description: str = "" + input_schema: dict[str, Any] = Field(default_factory=dict) + output_schema: dict[str, Any] = Field(default_factory=dict) + timeout_seconds: float = 20.0 + retry_count: int = 0 + cost: dict[str, Any] = Field(default_factory=dict) + rate_limit: dict[str, Any] = Field(default_factory=dict) + guardrails: list[str] = Field(default_factory=list) + redact_fields: list[str] = Field(default_factory=list) + + +class ToolCallResult(BaseModel): + tool_name: str + input: dict[str, Any] = Field(default_factory=dict) + output: dict[str, Any] = Field(default_factory=dict) + ok: bool = True + duration_ms: int = 0 + error_type: str = "" + error: str = "" + + +class AgentTraceEvent(BaseModel): + event_id: str = Field(default_factory=lambda: str(uuid.uuid4())) + span_id: str = Field(default_factory=lambda: str(uuid.uuid4())) + parent_span_id: str | None = None + kind: AgentStepKind = "agent" + action: str + agent_name: str = "" + tool_name: str = "" + status: str = "ok" + duration_ms: int | None = None + model: str | None = None + tokens: dict[str, Any] | None = None + cost: dict[str, Any] = Field(default_factory=dict) + input: dict[str, Any] | None = None + output: dict[str, Any] | None = None + error_type: str = "" + error: str = "" + metadata: dict[str, Any] = Field(default_factory=dict) + at: str = "" + + def as_legacy_step(self) -> dict[str, Any]: + data = self.model_dump(exclude_none=True) + # Keep old Dashboard compatibility while exposing normalized trace fields. + data["kind"] = self.kind + data["action"] = self.action + if self.tool_name: + data["tool"] = self.tool_name + if self.input is not None: + data["input"] = self.input + if self.output is not None: + data["output"] = self.output + for key, value in self.metadata.items(): + data.setdefault(key, value) + return data + + +class AgentState(BaseModel): + run_id: str | None = None + agent_name: str + status: str = "running" + node: str = "start" + term: str = "" + observations: list[dict[str, Any]] = Field(default_factory=list) + evidence: list[dict[str, Any]] = Field(default_factory=list) + tools_used: list[str] = Field(default_factory=list) + metadata: dict[str, Any] = Field(default_factory=dict) + + +TIn = TypeVar("TIn", bound=BaseModel) +TOut = TypeVar("TOut", bound=BaseModel) + + +@dataclass(frozen=True) +class RegisteredTool(Generic[TIn, TOut]): + spec: ToolSpec + input_model: type[TIn] + output_model: type[TOut] | None = None + handler: Callable[[TIn], TOut | dict[str, Any]] | None = None + + +@dataclass +class ToolRegistry: + _tools: dict[str, RegisteredTool[Any, Any]] = field(default_factory=dict) + + def register(self, tool: RegisteredTool[Any, Any]) -> None: + self._tools[tool.spec.name] = tool + + def names(self) -> list[str]: + return sorted(self._tools) + + def spec(self, name: str) -> ToolSpec: + return self._tools[name].spec + + def specs(self) -> list[ToolSpec]: + return [tool.spec for tool in self._tools.values()] + + +class AgentBudgetExceeded(RuntimeError): + pass + + +@dataclass +class AgentRuntime: + agent_name: str + budget: AgentBudget + trace_recorder: Callable[[dict[str, Any]], None] | None = None + run_id: str | None = None + parent_span_id: str | None = None + started_monotonic: float = field(default_factory=time.monotonic) + llm_calls: int = 0 + tool_calls: int = 0 + fetch_calls: int = 0 + + def check_budget(self, *, tool_name: str = "", is_llm: bool = False) -> None: + if time.monotonic() - self.started_monotonic > self.budget.timeout_seconds: + raise AgentBudgetExceeded("agent timeout budget exceeded") + if is_llm and self.llm_calls >= self.budget.max_llm_calls: + raise AgentBudgetExceeded("agent LLM call budget exceeded") + if tool_name: + if self.tool_calls >= self.budget.max_tool_calls: + raise AgentBudgetExceeded("agent tool call budget exceeded") + if tool_name in {"fetch", "fetch_url"} and self.fetch_calls >= self.budget.max_fetch_calls: + raise AgentBudgetExceeded("agent fetch call budget exceeded") + + def before_llm(self) -> None: + self.check_budget(is_llm=True) + self.llm_calls += 1 + + def before_tool(self, tool_name: str) -> None: + self.check_budget(tool_name=tool_name) + self.tool_calls += 1 + if tool_name in {"fetch", "fetch_url"}: + self.fetch_calls += 1 + + def record(self, event: AgentTraceEvent) -> None: + if self.trace_recorder is None: + return + if not event.agent_name: + event.agent_name = self.agent_name + if self.parent_span_id and event.parent_span_id is None: + event.parent_span_id = self.parent_span_id + payload = event.as_legacy_step() + payload.setdefault("run_id", self.run_id) + payload.setdefault( + "budget", + { + "llm_calls": self.llm_calls, + "tool_calls": self.tool_calls, + "fetch_calls": self.fetch_calls, + "max_llm_calls": self.budget.max_llm_calls, + "max_tool_calls": self.budget.max_tool_calls, + "max_fetch_calls": self.budget.max_fetch_calls, + "timeout_seconds": self.budget.timeout_seconds, + }, + ) + self.trace_recorder(payload) + + +def validate_model(model: type[TOut], data: Any, *, fallback: TOut | None = None) -> TOut: + try: + if isinstance(data, model): + return data + return model.model_validate(data) + except ValidationError: + if fallback is not None: + return fallback + raise + + +def json_schema_for(model: type[BaseModel]) -> dict[str, Any]: + return json.loads(json.dumps(model.model_json_schema(), ensure_ascii=False)) + diff --git a/src/videoroll/apps/subtitle_service/main.py b/src/videoroll/apps/subtitle_service/main.py index 2c1681f..31cbd7d 100644 --- a/src/videoroll/apps/subtitle_service/main.py +++ b/src/videoroll/apps/subtitle_service/main.py @@ -353,6 +353,13 @@ def get_translate_settings_view( rag_wiki_enabled=cfg["rag_wiki_enabled"], rag_search_enabled=cfg["rag_search_enabled"], rag_search_url=cfg["rag_search_url"], + rag_search_categories=cfg["rag_search_categories"], + rag_search_engines=cfg["rag_search_engines"], + rag_search_fallback_engines=cfg["rag_search_fallback_engines"], + rag_search_language=cfg["rag_search_language"], + rag_search_safesearch=cfg["rag_search_safesearch"], + rag_search_time_range=cfg["rag_search_time_range"], + rag_search_pageno=cfg["rag_search_pageno"], rag_domain=cfg["rag_domain"], rag_agent_parallelism=cfg["rag_agent_parallelism"], rag_agent_timeout_seconds=cfg["rag_agent_timeout_seconds"], @@ -395,6 +402,13 @@ def put_translate_settings_view( rag_wiki_enabled=cfg["rag_wiki_enabled"], rag_search_enabled=cfg["rag_search_enabled"], rag_search_url=cfg["rag_search_url"], + rag_search_categories=cfg["rag_search_categories"], + rag_search_engines=cfg["rag_search_engines"], + rag_search_fallback_engines=cfg["rag_search_fallback_engines"], + rag_search_language=cfg["rag_search_language"], + rag_search_safesearch=cfg["rag_search_safesearch"], + rag_search_time_range=cfg["rag_search_time_range"], + rag_search_pageno=cfg["rag_search_pageno"], rag_domain=cfg["rag_domain"], rag_agent_parallelism=cfg["rag_agent_parallelism"], rag_agent_timeout_seconds=cfg["rag_agent_timeout_seconds"], diff --git a/src/videoroll/apps/subtitle_service/rag.py b/src/videoroll/apps/subtitle_service/rag.py index fa7d639..c4c3889 100644 --- a/src/videoroll/apps/subtitle_service/rag.py +++ b/src/videoroll/apps/subtitle_service/rag.py @@ -18,10 +18,27 @@ from urllib.parse import parse_qsl, quote, urlencode, urljoin, urlparse, urlunpa import httpx from sqlalchemy import text from sqlalchemy.orm import Session +from pydantic import BaseModel from videoroll.ai.client import OpenAIChatConfig, request_openai_json_object +from videoroll.apps.subtitle_service.agent_runtime import ( + AgentBudget, + AgentBudgetExceeded, + AgentDecision, + AgentRuntime, + AgentTraceEvent, + GlossaryCandidate, + RegisteredTool, + SearchQueryPlan, + ToolRegistry, + ToolSpec as RuntimeToolSpec, + VerificationResult, + json_schema_for, + validate_model, +) from videoroll.apps.subtitle_service.embeddings import EmbeddingSettings, assert_embedding_dimensions, embed_text from videoroll.apps.subtitle_service.processing import Segment +from videoroll.apps.subtitle_service.retrieval import RetrievalPipeline _TERM_SPLIT_RE = re.compile(r"[\s\-_]+") @@ -86,6 +103,13 @@ class RagSettings: auto_learn_terms: bool = False search_enabled: bool = False search_url: str = "" + search_categories: str = "general" + search_engines: str = "" + search_fallback_engines: str = "bing,baidu" + search_language: str = "all" + search_safesearch: int = 0 + search_time_range: str = "" + search_pageno: int = 1 wiki_enabled: bool = False domain: str = "" agent_parallelism: int = 1 @@ -130,9 +154,13 @@ class ToolSpec: tool_name: str input_schema: dict[str, Any] output_schema: dict[str, Any] + description: str = "" timeout_seconds: float = 20.0 retry_count: int = 0 cost: dict[str, Any] | None = None + rate_limit: dict[str, Any] | None = None + guardrails: list[str] | None = None + redact_fields: list[str] | None = None @dataclass(frozen=True) @@ -146,11 +174,16 @@ class ToolResult: error: str = "" def to_step(self, *, action: str) -> dict[str, Any]: + status = "ok" if self.ok else "failed" step: dict[str, Any] = { + "event_id": str(uuid.uuid4()), + "span_id": str(uuid.uuid4()), "kind": "tool", "action": action, "tool": self.spec.tool_name, "tool_name": self.spec.tool_name, + "status": status, + "description": self.spec.description, "input_schema": self.spec.input_schema, "output_schema": self.spec.output_schema, "input": self.input, @@ -160,6 +193,9 @@ class ToolResult: "timeout_seconds": self.spec.timeout_seconds, "retry_count": self.spec.retry_count, "cost": self.spec.cost or {}, + "rate_limit": self.spec.rate_limit or {}, + "guardrails": self.spec.guardrails or [], + "redact_fields": self.spec.redact_fields or [], } if self.error_type: step["error_type"] = self.error_type @@ -172,39 +208,293 @@ _SEARCH_TOOL_SPEC = ToolSpec( tool_name="search", input_schema={"type": "object", "required": ["query"], "properties": {"query": {"type": "string"}, "url": {"type": "string"}}}, output_schema={"type": "object", "properties": {"count": {"type": "integer"}, "results": {"type": "array"}}}, + description="Search the configured SearXNG instance and return filtered external result summaries.", timeout_seconds=20.0, retry_count=0, cost={"network_requests": 1}, + guardrails=["filter_search_engine_internal_pages", "dedupe_urls", "do_not_fetch_private_hosts"], ) _FETCH_TOOL_SPEC = ToolSpec( tool_name="fetch", input_schema={"type": "object", "required": ["url"], "properties": {"url": {"type": "string"}, "title": {"type": "string"}}}, output_schema={"type": "object", "properties": {"chars": {"type": "integer"}, "excerpt": {"type": "string"}}}, + description="Fetch a public URL and extract compact readable text for evidence.", timeout_seconds=20.0, retry_count=0, cost={"network_requests": 1}, + guardrails=["http_https_only", "block_private_hosts", "limit_response_chars"], ) _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"}}}, + description="Search English Wikipedia through the MediaWiki API.", timeout_seconds=20.0, retry_count=0, cost={"network_requests": 1}, + guardrails=["fixed_english_wikipedia_api", "dedupe_pageids"], ) _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"}}}, + description="Read the lead extract for a Wikipedia page.", timeout_seconds=20.0, retry_count=0, cost={"network_requests": 1}, + guardrails=["intro_extract_only", "limit_response_chars"], ) +class RagLookupInput(BaseModel): + term: str + + +class RagLookupOutput(BaseModel): + exists: bool = False + normalized_term: str = "" + + +class WikiSearchInput(BaseModel): + query: str + api_url: str = _WIKIPEDIA_API_URL + + +class WikiSearchOutput(BaseModel): + count: int = 0 + results: list[dict[str, Any]] = [] + + +class SearchWebInput(BaseModel): + query: str + url: str = "" + + +class SearchWebOutput(BaseModel): + count: int = 0 + results: list[dict[str, Any]] = [] + + +class FetchUrlInput(BaseModel): + url: str + title: str = "" + + +class FetchUrlOutput(BaseModel): + chars: int = 0 + excerpt: str = "" + + +class FinishInput(BaseModel): + reason: str = "" + final_answer_ready: bool = False + + +class FinishOutput(BaseModel): + finished: bool = True + + +def _runtime_tool_spec( + *, + name: str, + description: str, + input_model: type[BaseModel], + output_model: type[BaseModel], + timeout_seconds: float = 20.0, + retry_count: int = 0, + cost: dict[str, Any] | None = None, + rate_limit: dict[str, Any] | None = None, + guardrails: list[str] | None = None, + redact_fields: list[str] | None = None, +) -> RuntimeToolSpec: + return RuntimeToolSpec( + name=name, + description=description, + input_schema=json_schema_for(input_model), + output_schema=json_schema_for(output_model), + timeout_seconds=timeout_seconds, + retry_count=retry_count, + cost=cost or {}, + rate_limit=rate_limit or {}, + guardrails=guardrails or [], + redact_fields=redact_fields or [], + ) + + +def _research_tool_registry(rag_settings: RagSettings) -> ToolRegistry: + registry = ToolRegistry() + registry.register( + RegisteredTool( + spec=_runtime_tool_spec( + name="rag_lookup", + description="Check whether the local translation knowledge base already contains this term.", + input_model=RagLookupInput, + output_model=RagLookupOutput, + timeout_seconds=5.0, + guardrails=["read_only", "target_language_scoped"], + ), + input_model=RagLookupInput, + output_model=RagLookupOutput, + ) + ) + if rag_settings.wiki_enabled: + registry.register( + RegisteredTool( + spec=_runtime_tool_spec( + name="wiki_search", + description=_WIKI_SEARCH_TOOL_SPEC.description, + input_model=WikiSearchInput, + output_model=WikiSearchOutput, + timeout_seconds=_WIKI_SEARCH_TOOL_SPEC.timeout_seconds, + retry_count=_WIKI_SEARCH_TOOL_SPEC.retry_count, + cost=_WIKI_SEARCH_TOOL_SPEC.cost, + guardrails=_WIKI_SEARCH_TOOL_SPEC.guardrails, + ), + input_model=WikiSearchInput, + output_model=WikiSearchOutput, + ) + ) + if rag_settings.search_enabled: + registry.register( + RegisteredTool( + spec=_runtime_tool_spec( + name="search_web", + description=_SEARCH_TOOL_SPEC.description, + input_model=SearchWebInput, + output_model=SearchWebOutput, + timeout_seconds=_SEARCH_TOOL_SPEC.timeout_seconds, + retry_count=_SEARCH_TOOL_SPEC.retry_count, + cost=_SEARCH_TOOL_SPEC.cost, + guardrails=_SEARCH_TOOL_SPEC.guardrails, + ), + input_model=SearchWebInput, + output_model=SearchWebOutput, + ) + ) + registry.register( + RegisteredTool( + spec=_runtime_tool_spec( + name="fetch_url", + description=_FETCH_TOOL_SPEC.description, + input_model=FetchUrlInput, + output_model=FetchUrlOutput, + timeout_seconds=_FETCH_TOOL_SPEC.timeout_seconds, + retry_count=_FETCH_TOOL_SPEC.retry_count, + cost=_FETCH_TOOL_SPEC.cost, + guardrails=_FETCH_TOOL_SPEC.guardrails, + ), + input_model=FetchUrlInput, + output_model=FetchUrlOutput, + ) + ) + registry.register( + RegisteredTool( + spec=_runtime_tool_spec( + name="finish", + description="Stop the child agent when evidence is sufficient or further research is not useful.", + input_model=FinishInput, + output_model=FinishOutput, + timeout_seconds=1.0, + guardrails=["requires_reason"], + ), + input_model=FinishInput, + output_model=FinishOutput, + ) + ) + return registry + + +def _ordered_tool_specs(registry: ToolRegistry) -> list[dict[str, Any]]: + order = ["rag_lookup", "wiki_search", "search_web", "fetch_url", "finish"] + out: list[dict[str, Any]] = [] + for name in order: + try: + spec = registry.spec(name) + except KeyError: + continue + out.append(spec.model_dump()) + return out + + +_SEARXNG_TIME_RANGES = {"", "day", "month", "year"} + + +def _clean_searxng_csv(value: Any, *, default: str = "", limit: int = 20) -> str: + raw_items = str(value or default or "").replace("\n", ",").split(",") + out: list[str] = [] + seen: set[str] = set() + for item in raw_items: + clean = " ".join(str(item or "").strip().split()) + if not clean: + continue + key = clean.lower() + if key in seen: + continue + seen.add(key) + out.append(clean[:80]) + if len(out) >= limit: + break + return ",".join(out) + + +def _clean_searxng_language(value: Any) -> str: + clean = str(value or "all").strip() + return (clean or "all")[:32] + + +def _clean_searxng_safesearch(value: Any) -> int: + try: + return max(0, min(2, int(value if value is not None else 0))) + except Exception: + return 0 + + +def _clean_searxng_time_range(value: Any) -> str: + clean = str(value or "").strip().lower() + return clean if clean in _SEARXNG_TIME_RANGES else "" + + +def _searxng_search_params( + *, + categories: str = "general", + engines: str = "", + language: str = "all", + safesearch: int = 0, + time_range: str = "", + pageno: int = 1, +) -> dict[str, str]: + params: dict[str, str] = {} + clean_categories = _clean_searxng_csv(categories, default="general") + clean_engines = _clean_searxng_csv(engines, default="") + clean_language = _clean_searxng_language(language) + clean_time_range = _clean_searxng_time_range(time_range) + if clean_categories: + params["categories"] = clean_categories + if clean_engines: + params["engines"] = clean_engines + if clean_language: + params["language"] = clean_language + params["safesearch"] = str(_clean_searxng_safesearch(safesearch)) + if clean_time_range: + params["time_range"] = clean_time_range + try: + clean_pageno = max(1, min(100, int(pageno or 1))) + except Exception: + clean_pageno = 1 + params["pageno"] = str(clean_pageno) + return params + + +def _clean_searxng_pageno(value: Any) -> int: + try: + return max(1, min(100, int(value or 1))) + except Exception: + return 1 + + def rag_settings_from_translate_settings(settings: dict[str, Any]) -> RagSettings: return RagSettings( enabled=bool(settings.get("rag_enabled")), @@ -219,6 +509,13 @@ 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(), + search_categories=_clean_searxng_csv(settings.get("rag_search_categories"), default="general"), + search_engines=_clean_searxng_csv(settings.get("rag_search_engines"), default=""), + search_fallback_engines=_clean_searxng_csv(settings.get("rag_search_fallback_engines"), default="bing,baidu"), + search_language=_clean_searxng_language(settings.get("rag_search_language")), + search_safesearch=_clean_searxng_safesearch(settings.get("rag_search_safesearch")), + search_time_range=_clean_searxng_time_range(settings.get("rag_search_time_range")), + search_pageno=_clean_searxng_pageno(settings.get("rag_search_pageno")), 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))), @@ -621,7 +918,12 @@ def _append_agent_step(db: Session, run_id: str | None, step: dict[str, Any]) -> if not run_id: return clean_step = dict(step) + clean_step.setdefault("event_id", str(uuid.uuid4())) + clean_step.setdefault("span_id", str(uuid.uuid4())) clean_step.setdefault("at", _utc_now().isoformat()) + clean_step.setdefault("status", "failed" if clean_step.get("error") or clean_step.get("ok") is False else "ok") + if "tool" in clean_step and "tool_name" not in clean_step: + clean_step["tool_name"] = clean_step["tool"] try: db.execute( text( @@ -659,7 +961,13 @@ def _append_llm_step( ) -> None: if db is None: return - step: dict[str, Any] = {"kind": "llm", "action": action} + step: dict[str, Any] = { + "event_id": str(uuid.uuid4()), + "span_id": str(uuid.uuid4()), + "kind": "llm", + "action": action, + "status": "failed" if error or error_type else "ok", + } if config is not None: step["model"] = config.model step["tokens"] = None @@ -676,6 +984,41 @@ def _append_llm_step( _append_agent_step(db, run_id, step) +def _append_state_transition( + db: Session | None, + run_id: str | None, + *, + from_node: str, + to_node: str, + reason: str = "", + metadata: dict[str, Any] | None = None, +) -> None: + if db is None: + return + _append_agent_step( + db, + run_id, + { + "kind": "agent", + "action": "state_transition", + "from_node": from_node, + "to_node": to_node, + "reason": reason, + "metadata": metadata or {}, + }, + ) + + +def _agent_budget_for_rag(rag_settings: RagSettings, *, max_steps: int = 6) -> AgentBudget: + timeout_seconds = max(10.0, min(900.0, float(rag_settings.agent_timeout_seconds or 120.0))) + return AgentBudget( + max_llm_calls=max(4, min(24, int(max_steps) + 6)), + max_tool_calls=max(4, min(30, int(max_steps) * 2 + 4)), + max_fetch_calls=4, + timeout_seconds=timeout_seconds, + ) + + def _finish_agent_run( db: Session, run_id: str | None, @@ -1077,10 +1420,11 @@ def generate_search_queries_openai( ), client=client, ) - raw_queries = data.get("queries") - if not isinstance(raw_queries, list): + try: + plan = validate_model(SearchQueryPlan, data) + except Exception: return [] - queries = _dedupe_search_queries([str(x or "") for x in raw_queries], limit=safe_max) + queries = _dedupe_search_queries([str(x or "") for x in plan.queries], limit=safe_max) return queries or _fallback_search_queries(clean_term, domain=domain_hint, target_lang=target_lang, limit=safe_max) @@ -1167,6 +1511,13 @@ def fetch_search_evidence( *, domain: str, search_url: str, + search_categories: str = "general", + search_engines: str = "", + search_fallback_engines: str = "bing,baidu", + search_language: str = "all", + search_safesearch: int = 0, + search_time_range: str = "", + search_pageno: int = 1, queries: list[str] | None = None, context: str = "", target_lang: str = "zh", @@ -1194,12 +1545,23 @@ def fetch_search_evidence( follow_redirects=True, headers={"User-Agent": "VideoRoll-RAG-Agent/1.0", "Accept": "application/json, text/html;q=0.9,*/*;q=0.8"}, ) as client: - has_configured_engines = _search_endpoint_has_param(endpoint, "engines") + base_search_params = _searxng_search_params( + categories=search_categories, + engines=search_engines, + language=search_language, + safesearch=search_safesearch, + time_range=search_time_range, + pageno=search_pageno, + ) + fallback_engines = _clean_searxng_csv(search_fallback_engines, default="") + has_configured_engines = _search_endpoint_has_param(endpoint, "engines") or bool(base_search_params.get("engines")) for query in search_queries: filtered_results: list[dict[str, Any]] = [] - attempt_configs: list[dict[str, str] | None] = [None] - if not has_configured_engines: - attempt_configs.append({"engines": "bing,baidu"}) + attempt_configs: list[dict[str, str]] = [dict(base_search_params)] + if not has_configured_engines and fallback_engines: + fallback_params = dict(base_search_params) + fallback_params["engines"] = fallback_engines + attempt_configs.append(fallback_params) for attempt_index, extra_params in enumerate(attempt_configs): json_url = _search_url_with_params(endpoint, query=query, json_format=True, extra_params=extra_params) html_url = _search_url_with_params(endpoint, query=query, json_format=False, extra_params=extra_params) @@ -1260,8 +1622,9 @@ def fetch_search_evidence( "filtered_count": len(filtered_results), "results": compact_results, "json_error": json_error, + "search_params": extra_params, } - if extra_params: + if attempt_index > 0: output_value["fallback_params"] = extra_params if unresponsive_engines: output_value["unresponsive_engines"] = unresponsive_engines[:8] @@ -1282,9 +1645,10 @@ def fetch_search_evidence( "filtered_count": len(filtered_results), "results": compact_results, "retry_count": attempt_index, + "search_params": extra_params, } ) - if extra_params: + if attempt_index > 0: step["fallback_params"] = extra_params if unresponsive_engines: step["unresponsive_engines"] = unresponsive_engines[:8] @@ -1732,6 +2096,7 @@ def research_agent_next_action_openai( target_lang: str, domain_hint: str, available_tools: list[str], + available_tool_specs: list[dict[str, Any]] | None = None, observations: list[dict[str, Any]], evidence: list[dict[str, Any]], config: OpenAIChatConfig, @@ -1773,6 +2138,7 @@ def research_agent_next_action_openai( f"目标语言: {target_lang or 'zh'}\n" f"领域提示: {domain_hint or '未知'}\n" f"可用工具: {json.dumps(available_tools, ensure_ascii=False)}\n\n" + f"工具 schema JSON:\n{json.dumps(available_tool_specs or [], ensure_ascii=False)[:5000]}\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" @@ -1784,13 +2150,18 @@ def research_agent_next_action_openai( 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")), - } + decision = validate_model( + AgentDecision, + { + "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")), + }, + fallback=AgentDecision(action="finish", reason="invalid tool decision output"), + ) + return decision.model_dump() def _dedupe_evidence(items: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: @@ -1822,13 +2193,9 @@ def _collect_evidence_with_tool_agent( 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") + tool_registry = _research_tool_registry(rag_settings) + available_tool_specs = _ordered_tool_specs(tool_registry) + available_tools = [str(spec.get("name") or "") for spec in available_tool_specs if str(spec.get("name") or "")] observations: list[dict[str, Any]] = [] evidence: list[dict[str, Any]] = [] @@ -1836,10 +2203,35 @@ def _collect_evidence_with_tool_agent( used_actions: set[str] = set() empty_search_count = 0 rounds = 0 + runtime = AgentRuntime( + agent_name="rag_term_research", + run_id=agent_run_id, + budget=_agent_budget_for_rag(rag_settings, max_steps=max_steps), + trace_recorder=(lambda step: _append_agent_step(db, agent_run_id, step)) if db is not None else None, + ) + runtime.record( + AgentTraceEvent( + kind="agent", + action="agent_runtime_start", + output={ + "available_tools": available_tools, + "tool_specs": available_tool_specs, + "budget": runtime.budget.model_dump(), + }, + ) + ) + runtime.record( + AgentTraceEvent( + kind="agent", + action="state_transition", + output={"from_node": "start", "to_node": "decide_tool", "reason": "child agent initialized"}, + ) + ) for step_no in range(1, max(1, min(10, int(max_steps))) + 1): rounds = step_no try: + runtime.before_llm() started = time.perf_counter() decision = research_agent_next_action_openai( term=term, @@ -1847,6 +2239,7 @@ def _collect_evidence_with_tool_agent( target_lang=target_lang, domain_hint=domain_hint, available_tools=available_tools, + available_tool_specs=available_tool_specs, observations=observations, evidence=evidence, config=chat_config, @@ -1867,6 +2260,19 @@ def _collect_evidence_with_tool_agent( output_value=decision, duration_ms=_duration_ms(started), ) + except AgentBudgetExceeded as e: + observations.append({"action": "finish", "reason": str(e), "evidence_count": len(evidence)}) + runtime.record( + AgentTraceEvent( + kind="policy", + action="agent_budget_exceeded", + status="failed", + error_type=type(e).__name__, + error=str(e), + output={"evidence_count": len(evidence), "round": step_no}, + ) + ) + break except Exception as e: decision = {"action": "finish", "query": "", "url": "", "reason": f"tool decision failed: {e}", "final_answer_ready": False} _append_llm_step( @@ -1883,9 +2289,29 @@ def _collect_evidence_with_tool_agent( 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)}) + runtime.record( + AgentTraceEvent( + kind="agent", + action="state_transition", + output={"from_node": "decide_tool", "to_node": "finish", "reason": reason}, + ) + ) break if action == "rag_lookup": + try: + runtime.before_tool("rag_lookup") + except AgentBudgetExceeded as e: + runtime.record( + AgentTraceEvent( + kind="policy", + action="agent_budget_exceeded", + status="failed", + error_type=type(e).__name__, + error=str(e), + ) + ) + break tools_used.append("rag_lookup") used_actions.add(action) norm_value = normalize_term(term) @@ -1897,6 +2323,13 @@ def _collect_evidence_with_tool_agent( {"kind": "tool", "action": "rag_lookup", "tool": "rag_lookup", "term": term, "exists": exists, "ok": True}, ) if exists: + runtime.record( + AgentTraceEvent( + kind="agent", + action="state_transition", + output={"from_node": "rag_lookup", "to_node": "finish", "reason": "local knowledge already exists"}, + ) + ) break continue @@ -1905,6 +2338,13 @@ def _collect_evidence_with_tool_agent( 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: + try: + runtime.before_tool("wiki_search") + except AgentBudgetExceeded as e: + runtime.record( + AgentTraceEvent(kind="policy", action="agent_budget_exceeded", status="failed", error_type=type(e).__name__, error=str(e)) + ) + break 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) @@ -1913,12 +2353,26 @@ def _collect_evidence_with_tool_agent( continue if action == "search_web" and rag_settings.search_enabled: + try: + runtime.before_tool("search") + except AgentBudgetExceeded as e: + runtime.record( + AgentTraceEvent(kind="policy", action="agent_budget_exceeded", status="failed", error_type=type(e).__name__, error=str(e)) + ) + break tools_used.append("search") used_actions.add(action) extra = fetch_search_evidence( term, domain=domain_hint, search_url=rag_settings.search_url, + search_categories=rag_settings.search_categories, + search_engines=rag_settings.search_engines, + search_fallback_engines=rag_settings.search_fallback_engines, + search_language=rag_settings.search_language, + search_safesearch=rag_settings.search_safesearch, + search_time_range=rag_settings.search_time_range, + search_pageno=rag_settings.search_pageno, queries=[query], context=llm_context, target_lang=target_lang, @@ -1950,6 +2404,13 @@ def _collect_evidence_with_tool_agent( continue if action == "fetch_url": + try: + runtime.before_tool("fetch_url") + except AgentBudgetExceeded as e: + runtime.record( + AgentTraceEvent(kind="policy", action="agent_budget_exceeded", status="failed", error_type=type(e).__name__, error=str(e)) + ) + break tools_used.append("fetch") used_actions.add(action) url = str(decision.get("url") or "").strip() @@ -1969,6 +2430,13 @@ def _collect_evidence_with_tool_agent( # Safety fallback: if the model stopped too early, try enabled tools once. if not evidence: + runtime.record( + AgentTraceEvent( + kind="policy", + action="agent_evidence_fallback", + output={"reason": "no evidence collected in tool loop", "used_actions": sorted(used_actions)}, + ) + ) 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)) @@ -1979,6 +2447,13 @@ def _collect_evidence_with_tool_agent( term, domain=domain_hint, search_url=rag_settings.search_url, + search_categories=rag_settings.search_categories, + search_engines=rag_settings.search_engines, + search_fallback_engines=rag_settings.search_fallback_engines, + search_language=rag_settings.search_language, + search_safesearch=rag_settings.search_safesearch, + search_time_range=rag_settings.search_time_range, + search_pageno=rag_settings.search_pageno, queries=search_queries, context=llm_context, target_lang=target_lang, @@ -2033,15 +2508,22 @@ def explain_term_from_evidence_openai( confidence = float(data.get("confidence") or 0.0) except Exception: confidence = 0.0 - return { - "term": str(data.get("term") or clean_term).strip(), - "translation": translation, - "domain": str(data.get("domain") or domain_hint or "").strip(), - "aliases": aliases if isinstance(aliases, list) else [], - "description": str(data.get("description") or "").strip(), - "sources": sources if isinstance(sources, list) else [], - "confidence": max(0.0, min(1.0, confidence)), - } + try: + candidate = validate_model( + GlossaryCandidate, + { + "term": str(data.get("term") or clean_term).strip(), + "translation": translation, + "domain": str(data.get("domain") or domain_hint or "").strip(), + "aliases": aliases if isinstance(aliases, list) else [], + "description": str(data.get("description") or "").strip(), + "sources": sources if isinstance(sources, list) else [], + "confidence": max(0.0, min(1.0, confidence)), + }, + ) + except Exception: + return None + return candidate.model_dump() def _json_bool(value: Any) -> bool: @@ -2131,15 +2613,19 @@ def verify_glossary_entry_openai( confidence = float(data.get("confidence") or 0.0) except Exception: confidence = 0.0 - return { - "supported": _json_bool(data.get("supported")), - "context_consistent": _json_bool(data.get("context_consistent")), - "should_write": _json_bool(data.get("should_write")), - "should_auto_approve": _json_bool(data.get("should_auto_approve")), - "confidence": max(0.0, min(1.0, confidence)), - "reason": str(data.get("reason") or "").strip()[:1200], - "failure_category": str(data.get("failure_category") or "").strip()[:120], - } + verified = validate_model( + VerificationResult, + { + "supported": _json_bool(data.get("supported")), + "context_consistent": _json_bool(data.get("context_consistent")), + "should_write": _json_bool(data.get("should_write")), + "should_auto_approve": _json_bool(data.get("should_auto_approve")), + "confidence": max(0.0, min(1.0, confidence)), + "reason": str(data.get("reason") or "").strip()[:1200], + "failure_category": str(data.get("failure_category") or "").strip()[:120], + }, + ) + return verified.model_dump() def upsert_knowledge_item( @@ -2508,6 +2994,14 @@ def _research_discovered_term( }, duration_ms=gate_duration_ms, ) + _append_state_transition( + db, + agent_run_id, + from_node="start", + to_node="policy", + reason="child agent received gate item", + metadata={"term": term, "category": str(item.get("category") or "")}, + ) except Exception: db.rollback() agent_run_id = None @@ -2525,6 +3019,14 @@ def _research_discovered_term( }, ) if not research_policy.get("should_research"): + _append_state_transition( + db, + agent_run_id, + from_node="policy", + to_node="skipped", + reason=str(research_policy.get("reason") or "policy skipped research"), + metadata={"category": str(research_policy.get("category") or "")}, + ) 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) @@ -2558,6 +3060,14 @@ def _research_discovered_term( return AgentResearchResult(term=term, normalized_term=norm, context_card=context_card) if not bool(research_policy.get("need_search", True)): + _append_state_transition( + db, + agent_run_id, + from_node="policy", + to_node="skipped", + reason=str(research_policy.get("reason") or "external lookup was not requested"), + metadata={"category": str(research_policy.get("category") or "")}, + ) _finish_agent_run( db, agent_run_id, @@ -2573,6 +3083,14 @@ def _research_discovered_term( return AgentResearchResult(term=term, normalized_term=norm) if norm in existing_term_norms(db, terms=[term], target_lang=target_lang): + _append_state_transition( + db, + agent_run_id, + from_node="policy", + to_node="skipped", + reason="term already exists in knowledge base", + metadata={"normalized_term": norm}, + ) _finish_agent_run( db, agent_run_id, @@ -2590,6 +3108,14 @@ def _research_discovered_term( external_lookup_enabled = bool(rag_settings.wiki_enabled or rag_settings.search_enabled) search_queries: list[str] = [] if external_lookup_enabled: + _append_state_transition( + db, + agent_run_id, + from_node="policy", + to_node="query_planning", + reason="external lookup enabled", + metadata={"wiki": bool(rag_settings.wiki_enabled), "search": bool(rag_settings.search_enabled)}, + ) started = time.perf_counter() try: search_queries = generate_search_queries_openai( @@ -2653,6 +3179,13 @@ def _research_discovered_term( term, domain=domain_hint, search_url=rag_settings.search_url, + search_categories=rag_settings.search_categories, + search_engines=rag_settings.search_engines, + search_fallback_engines=rag_settings.search_fallback_engines, + search_language=rag_settings.search_language, + search_safesearch=rag_settings.search_safesearch, + search_time_range=rag_settings.search_time_range, + search_pageno=rag_settings.search_pageno, queries=search_queries, context=llm_context, target_lang=target_lang, @@ -2702,7 +3235,23 @@ def _research_discovered_term( max_steps=6, ) tools_used.extend(agent_tools_used) + _append_state_transition( + db, + agent_run_id, + from_node="query_planning", + to_node="summarize", + reason="evidence collection finished", + metadata={"evidence_count": len(evidence), "tools_used": tools_used, "rounds": evidence_rounds}, + ) if not evidence: + _append_state_transition( + db, + agent_run_id, + from_node="summarize", + to_node="failed", + reason="no search evidence", + metadata={"tools_used": tools_used}, + ) _finish_agent_run( db, agent_run_id, @@ -2805,6 +3354,14 @@ def _research_discovered_term( return None valid_sources = _valid_external_evidence(evidence, search_url=rag_settings.search_url) + _append_state_transition( + db, + agent_run_id, + from_node="summarize", + to_node="verify", + reason="candidate summary available", + metadata={"evidence_round": evidence_rounds, "valid_source_count": len(valid_sources)}, + ) try: verify_started = time.perf_counter() verification = verify_glossary_entry_openai( @@ -2859,6 +3416,14 @@ def _research_discovered_term( and bool(valid_sources) ) if should_write: + _append_state_transition( + db, + agent_run_id, + from_node="verify", + to_node="write_rag", + reason="verifier approved knowledge write", + metadata={"confidence": final_confidence, "valid_source_count": len(valid_sources)}, + ) break # Before finishing a rejected candidate, try another enabled evidence tool. # context_only is a semantic rejection rather than an evidence gap. @@ -2881,6 +3446,14 @@ def _research_discovered_term( failure_category = str(verification.get("failure_category") or "") if not valid_sources: failure_category = failure_category or "no_valid_external_source" + _append_state_transition( + db, + agent_run_id, + from_node="verify", + to_node="skipped", + reason=str(verification.get("reason") or "verifier rejected glossary entry"), + metadata={"failure_category": failure_category or "verify_rejected", "confidence": final_confidence}, + ) _finish_agent_run( db, agent_run_id, @@ -2909,6 +3482,14 @@ def _research_discovered_term( ) source_list = [x for x in explained.get("sources") or [] if isinstance(x, dict)] or valid_sources try: + _append_state_transition( + db, + agent_run_id, + from_node="write_rag", + to_node="embedding", + reason="embedding verified glossary entry", + metadata={"status": status, "confidence": final_confidence}, + ) _append_agent_step( db, agent_run_id, @@ -2951,6 +3532,14 @@ def _research_discovered_term( dedupe_any_domain=True, ) db.commit() + _append_state_transition( + db, + agent_run_id, + from_node="embedding", + to_node="succeeded", + reason="knowledge item saved", + metadata={"knowledge_item_id": learned_id, "status": status}, + ) _finish_agent_run( db, agent_run_id, @@ -2992,6 +3581,14 @@ def _research_discovered_term( return AgentResearchResult(term=term, normalized_term=norm) except Exception as e: db.rollback() + _append_state_transition( + db, + agent_run_id, + from_node="embedding", + to_node="failed", + reason="knowledge save failed", + metadata={"error_type": type(e).__name__}, + ) _finish_agent_run( db, agent_run_id, @@ -3204,27 +3801,40 @@ def build_rag_context( }, }, ) + _append_state_transition( + db, + master_agent_run_id, + from_node="start", + to_node="retrieval_exact", + reason="master agent initialized", + metadata={"segment_count": len(segments)}, + ) except Exception: db.rollback() master_agent_run_id = None - hits: list[RagHit] = [] - seen_ids: set[str] = set() + retrieval = RetrievalPipeline[RagHit]( + id_getter=lambda hit: hit.id, + trace_recorder=(lambda step: _append_agent_step(db, master_agent_run_id, step)) if master_agent_run_id else None, + error_handler=lambda _e: db.rollback(), + ) gate_terms: list[dict[str, Any]] | None = None gate_norms: set[str] = set() gate_duration_ms: int | None = None existing_gate_terms: list[dict[str, Any]] = [] - for hit in exact_term_hits( - db, - text_value=text_value, - target_lang=target_lang, - domain=rag_settings.domain, - limit=max(rag_settings.top_k * 2, 10), - ): - if hit.id not in seen_ids: - seen_ids.add(hit.id) - hits.append(hit) + exact_hits = retrieval.run_stage( + "retrieval_exact_terms", + text_value[:240], + lambda: exact_term_hits( + db, + text_value=text_value, + target_lang=target_lang, + domain=rag_settings.domain, + limit=max(rag_settings.top_k * 2, 10), + ), + ) + for hit in exact_hits: if hit.item_type == "term": existing_gate_terms.append( { @@ -3247,6 +3857,16 @@ def build_rag_context( "ok": True, }, ) + hits: list[RagHit] = retrieval.hits + seen_ids: set[str] = {hit.id for hit in hits} + _append_state_transition( + db, + master_agent_run_id, + from_node="retrieval_exact", + to_node="rag_gate" if rag_settings.auto_discover_terms else "retrieval_pgvector", + reason="exact term retrieval finished", + metadata={"hit_count": len(hits), "existing_term_count": len(existing_gate_terms)}, + ) if rag_settings.auto_discover_terms: gate_started = time.perf_counter() @@ -3286,27 +3906,35 @@ def build_rag_context( error_type="gate_failed", ) gate_duration_ms = _duration_ms(gate_started) + _append_state_transition( + db, + master_agent_run_id, + from_node="rag_gate", + to_node="retrieval_pgvector", + reason="pre-translation RAG gate finished", + metadata={"candidate_count": len(gate_terms or [])}, + ) - try: - 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(): + 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(): + def _vector_lookup() -> list[RagHit]: 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( + vector_hits = search_knowledge( db, query_embedding=query_embedding, target_lang=target_lang, @@ -3314,14 +3942,26 @@ def build_rag_context( 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 = [] + ) + if gate_terms is None: + return vector_hits + return [ + hit + for hit in vector_hits + if not (hit.item_type == "term" and normalize_term(hit.term) not in gate_norms) + ] + + retrieval.run_stage("retrieval_pgvector", rag_query_text[:240], _vector_lookup) + hits = retrieval.hits + seen_ids = {hit.id for hit in hits} + _append_state_transition( + db, + master_agent_run_id, + from_node="retrieval_pgvector", + to_node="dispatch_subagents" if rag_settings.auto_discover_terms else "build_context", + reason="vector retrieval finished", + metadata={"hit_count": len(hits), "stage_count": len(retrieval.stages)}, + ) context_only_cards: list[dict[str, Any]] = [] subagent_count = 0 @@ -3387,6 +4027,14 @@ def build_rag_context( "parallelism": rag_settings.agent_parallelism, }, ) + _append_state_transition( + db, + master_agent_run_id, + from_node="dispatch_subagents", + to_node="wait_subagents", + reason="research children dispatched", + metadata={"research_count": len(research_items), "parallelism": rag_settings.agent_parallelism}, + ) for result in _run_research_agents( db=db, session_factory=session_factory, @@ -3412,6 +4060,14 @@ def build_rag_context( seen_ids.add(result.hit.id) hits.append(result.hit) discovered = [] + _append_state_transition( + db, + master_agent_run_id, + from_node="wait_subagents", + to_node="build_context", + reason="child agents returned", + metadata={"context_only_cards": len(context_only_cards), "hit_count": len(hits)}, + ) term_cards: list[dict[str, Any]] = list(context_only_cards) knowledge_cards: list[dict[str, Any]] = [] @@ -3449,6 +4105,14 @@ def build_rag_context( context=text_value[:2000], ) + _append_state_transition( + db, + master_agent_run_id, + from_node="build_context", + to_node="succeeded", + reason="RagContext built", + metadata={"term_cards": len(term_cards), "knowledge_cards": len(knowledge_cards), "hits": len(hits)}, + ) _finish_agent_run( db, master_agent_run_id, diff --git a/src/videoroll/apps/subtitle_service/retrieval.py b/src/videoroll/apps/subtitle_service/retrieval.py new file mode 100644 index 0000000..3968e3a --- /dev/null +++ b/src/videoroll/apps/subtitle_service/retrieval.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from typing import Any, Callable, Generic, Iterable, TypeVar + + +THit = TypeVar("THit") + + +@dataclass(frozen=True) +class RetrievalStage: + name: str + query: str = "" + count: int = 0 + duration_ms: int = 0 + error_type: str = "" + error: str = "" + + @property + def ok(self) -> bool: + return not self.error_type + + def to_trace_step(self) -> dict[str, Any]: + step: dict[str, Any] = { + "kind": "retrieval", + "action": self.name, + "status": "ok" if self.ok else "failed", + "query": self.query, + "count": self.count, + "duration_ms": self.duration_ms, + } + if self.error_type: + step["error_type"] = self.error_type + if self.error: + step["error"] = self.error + return step + + +@dataclass +class RetrievalResult(Generic[THit]): + hits: list[THit] = field(default_factory=list) + stages: list[RetrievalStage] = field(default_factory=list) + + +def _duration_ms(started: float) -> int: + return max(0, int((time.perf_counter() - started) * 1000)) + + +def merge_hits_by_id(items: Iterable[THit], *, id_getter: Callable[[THit], str]) -> list[THit]: + out: list[THit] = [] + seen: set[str] = set() + for item in items: + key = id_getter(item) + if not key or key in seen: + continue + seen.add(key) + out.append(item) + return out + + +class RetrievalPipeline(Generic[THit]): + """Small explicit retrieval pipeline inspired by RAG frameworks. + + The pipeline owns stage timing, dedupe, and trace events. The actual database + operations stay injected so the subtitle service can keep its current storage + layer and avoid a heavy framework dependency. + """ + + def __init__( + self, + *, + id_getter: Callable[[THit], str], + trace_recorder: Callable[[dict[str, Any]], None] | None = None, + error_handler: Callable[[Exception], None] | None = None, + ) -> None: + self._id_getter = id_getter + self._trace_recorder = trace_recorder + self._error_handler = error_handler + self._hits: list[THit] = [] + self._seen_ids: set[str] = set() + self.stages: list[RetrievalStage] = [] + + @property + def hits(self) -> list[THit]: + return list(self._hits) + + def add_hits(self, hits: Iterable[THit]) -> int: + added = 0 + for hit in hits: + key = self._id_getter(hit) + if not key or key in self._seen_ids: + continue + self._seen_ids.add(key) + self._hits.append(hit) + added += 1 + return added + + def run_stage(self, name: str, query: str, fn: Callable[[], list[THit]]) -> list[THit]: + started = time.perf_counter() + try: + hits = fn() + added = self.add_hits(hits) + stage = RetrievalStage(name=name, query=query, count=added, duration_ms=_duration_ms(started)) + except Exception as e: + if self._error_handler is not None: + try: + self._error_handler(e) + except Exception: + pass + hits = [] + stage = RetrievalStage( + name=name, + query=query, + count=0, + duration_ms=_duration_ms(started), + error_type=type(e).__name__, + error=str(e)[:500], + ) + self.stages.append(stage) + if self._trace_recorder is not None: + self._trace_recorder(stage.to_trace_step()) + return hits + + def result(self) -> RetrievalResult[THit]: + return RetrievalResult(hits=self.hits, stages=list(self.stages)) diff --git a/src/videoroll/apps/subtitle_service/schemas.py b/src/videoroll/apps/subtitle_service/schemas.py index 0bc79e2..476a4ae 100644 --- a/src/videoroll/apps/subtitle_service/schemas.py +++ b/src/videoroll/apps/subtitle_service/schemas.py @@ -219,6 +219,13 @@ class TranslateSettingsRead(BaseModel): rag_wiki_enabled: bool = False rag_search_enabled: bool = False rag_search_url: str = "" + rag_search_categories: str = "general" + rag_search_engines: str = "" + rag_search_fallback_engines: str = "bing,baidu" + rag_search_language: str = "all" + rag_search_safesearch: int = 0 + rag_search_time_range: str = "" + rag_search_pageno: int = 1 rag_domain: str = "" rag_agent_parallelism: int = 1 rag_agent_timeout_seconds: float = 120.0 @@ -256,6 +263,13 @@ class TranslateSettingsUpdate(BaseModel): rag_wiki_enabled: Optional[bool] = None rag_search_enabled: Optional[bool] = None rag_search_url: Optional[str] = None + rag_search_categories: Optional[str] = None + rag_search_engines: Optional[str] = None + rag_search_fallback_engines: Optional[str] = None + rag_search_language: Optional[str] = None + rag_search_safesearch: Optional[int] = Field(default=None, ge=0, le=2) + rag_search_time_range: Optional[str] = None + rag_search_pageno: Optional[int] = Field(default=None, ge=1, le=100) 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) diff --git a/src/videoroll/apps/subtitle_service/translate_settings_store.py b/src/videoroll/apps/subtitle_service/translate_settings_store.py index 2eeac79..01cd44a 100644 --- a/src/videoroll/apps/subtitle_service/translate_settings_store.py +++ b/src/videoroll/apps/subtitle_service/translate_settings_store.py @@ -11,6 +11,37 @@ from videoroll.utils.openai_compat import normalize_openai_base_url TRANSLATE_SETTINGS_KEY = "subtitle.translate" +_SEARXNG_TIME_RANGES = {"", "day", "month", "year"} + + +def _clean_csv(value: Any, *, default: str = "", limit: int = 20) -> str: + raw_items = str(value or default or "").replace("\n", ",").split(",") + out: list[str] = [] + seen: set[str] = set() + for item in raw_items: + clean = " ".join(str(item or "").strip().split()) + if not clean: + continue + key = clean.lower() + if key in seen: + continue + seen.add(key) + out.append(clean[:80]) + if len(out) >= limit: + break + return ",".join(out) + + +def _clean_search_language(value: Any, *, default: str = "all") -> str: + clean = str(value or default or "all").strip() + if not clean: + return "all" + return clean[:32] + + +def _clean_search_time_range(value: Any) -> str: + clean = str(value or "").strip().lower() + return clean if clean in _SEARXNG_TIME_RANGES else "" def _get_row(db: Session) -> AppSetting: @@ -89,6 +120,26 @@ def get_translate_settings(db: Session, defaults: SubtitleServiceSettings) -> di "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_search_categories": _clean_csv(stored.get("rag_search_categories"), default=defaults.rag_search_categories or "general"), + "rag_search_engines": _clean_csv(stored.get("rag_search_engines"), default=defaults.rag_search_engines or ""), + "rag_search_fallback_engines": _clean_csv( + stored.get("rag_search_fallback_engines"), + default=defaults.rag_search_fallback_engines or "bing,baidu", + ), + "rag_search_language": _clean_search_language(stored.get("rag_search_language"), default=defaults.rag_search_language or "all"), + "rag_search_safesearch": max( + 0, + min( + 2, + int( + stored.get("rag_search_safesearch") + if stored.get("rag_search_safesearch") is not None + else defaults.rag_search_safesearch + ), + ), + ), + "rag_search_time_range": _clean_search_time_range(stored.get("rag_search_time_range") or defaults.rag_search_time_range), + "rag_search_pageno": max(1, min(100, int(stored.get("rag_search_pageno") or defaults.rag_search_pageno or 1))), "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), @@ -120,6 +171,13 @@ def update_translate_settings(db: Session, defaults: SubtitleServiceSettings, up "rag_wiki_enabled", "rag_search_enabled", "rag_search_url", + "rag_search_categories", + "rag_search_engines", + "rag_search_fallback_engines", + "rag_search_language", + "rag_search_safesearch", + "rag_search_time_range", + "rag_search_pageno", "rag_domain", "rag_agent_parallelism", "rag_agent_timeout_seconds", @@ -221,9 +279,33 @@ def update_translate_settings(db: Session, defaults: SubtitleServiceSettings, up stored[bool_key] = bool(stored[bool_key]) provider = str(stored.get("rag_embedding_provider") or defaults.rag_embedding_provider).strip().lower() stored["rag_embedding_provider"] = provider if provider in {"openai", "local"} else defaults.rag_embedding_provider - for str_key in ["rag_embedding_model", "rag_embedding_model_dir", "rag_embedding_device", "rag_search_url", "rag_domain"]: + for str_key in [ + "rag_embedding_model", + "rag_embedding_model_dir", + "rag_embedding_device", + "rag_search_url", + "rag_search_language", + "rag_search_time_range", + "rag_domain", + ]: if str_key in stored and stored[str_key] is not None: stored[str_key] = str(stored[str_key]).strip() + stored["rag_search_categories"] = _clean_csv(stored.get("rag_search_categories"), default=defaults.rag_search_categories or "general") + stored["rag_search_engines"] = _clean_csv(stored.get("rag_search_engines"), default=defaults.rag_search_engines or "") + stored["rag_search_fallback_engines"] = _clean_csv( + stored.get("rag_search_fallback_engines"), + default=defaults.rag_search_fallback_engines or "bing,baidu", + ) + stored["rag_search_language"] = _clean_search_language(stored.get("rag_search_language"), default=defaults.rag_search_language or "all") + try: + stored["rag_search_safesearch"] = max(0, min(2, int(stored.get("rag_search_safesearch") or defaults.rag_search_safesearch or 0))) + except Exception: + stored["rag_search_safesearch"] = 0 + stored["rag_search_time_range"] = _clean_search_time_range(stored.get("rag_search_time_range") or defaults.rag_search_time_range) + try: + stored["rag_search_pageno"] = max(1, min(100, int(stored.get("rag_search_pageno") or defaults.rag_search_pageno or 1))) + except Exception: + stored["rag_search_pageno"] = 1 if "base_url" in embedding_openai: embedding_openai["base_url"] = normalize_openai_base_url(str(embedding_openai.get("base_url") or "")) try: diff --git a/src/videoroll/config.py b/src/videoroll/config.py index d343748..8f928c6 100644 --- a/src/videoroll/config.py +++ b/src/videoroll/config.py @@ -96,6 +96,13 @@ class SubtitleServiceSettings(CommonSettings): rag_auto_learn_terms: bool = Field(False, alias="SUBTITLE_RAG_AUTO_LEARN_TERMS") rag_search_enabled: bool = Field(False, alias="SUBTITLE_RAG_SEARCH_ENABLED") rag_search_url: str = Field("", alias="SUBTITLE_RAG_SEARCH_URL") + rag_search_categories: str = Field("general", alias="SUBTITLE_RAG_SEARCH_CATEGORIES") + rag_search_engines: str = Field("", alias="SUBTITLE_RAG_SEARCH_ENGINES") + rag_search_fallback_engines: str = Field("bing,baidu", alias="SUBTITLE_RAG_SEARCH_FALLBACK_ENGINES") + rag_search_language: str = Field("all", alias="SUBTITLE_RAG_SEARCH_LANGUAGE") + rag_search_safesearch: int = Field(0, alias="SUBTITLE_RAG_SEARCH_SAFESEARCH") + rag_search_time_range: str = Field("", alias="SUBTITLE_RAG_SEARCH_TIME_RANGE") + rag_search_pageno: int = Field(1, alias="SUBTITLE_RAG_SEARCH_PAGENO") rag_domain: str = Field("", alias="SUBTITLE_RAG_DOMAIN") # Orchestrator API (used by subtitle worker for auto pipelines). diff --git a/src/web/src/pages/SettingsTranslatePage.tsx b/src/web/src/pages/SettingsTranslatePage.tsx index 7f3512e..4cc3caf 100644 --- a/src/web/src/pages/SettingsTranslatePage.tsx +++ b/src/web/src/pages/SettingsTranslatePage.tsx @@ -34,6 +34,13 @@ type TranslateSettings = { rag_wiki_enabled: boolean; rag_search_enabled: boolean; rag_search_url: string; + rag_search_categories: string; + rag_search_engines: string; + rag_search_fallback_engines: string; + rag_search_language: string; + rag_search_safesearch: number; + rag_search_time_range: string; + rag_search_pageno: number; rag_domain: string; rag_agent_parallelism: number; rag_agent_timeout_seconds: number; @@ -47,6 +54,51 @@ type EmbeddingModelInfo = { type TranslateSettingsTab = "translation" | "rag" | "embedding" | "test"; +const SEARXNG_CATEGORY_PRESETS = ["general", "it", "science", "news", "videos", "images", "files", "social media"]; +const SEARXNG_ENGINE_PRESETS = [ + "bing", + "baidu", + "brave", + "duckduckgo", + "google", + "startpage", + "qwant", + "mojeek", + "wikipedia", + "wikidata", + "arxiv", + "crossref", + "pubmed", + "semantic scholar", + "github", + "stackoverflow", + "reddit", + "youtube", +]; + +function csvItems(value: string) { + const seen = new Set(); + const out: string[] = []; + value + .split(",") + .map((item) => item.trim()) + .filter(Boolean) + .forEach((item) => { + const key = item.toLowerCase(); + if (seen.has(key)) return; + seen.add(key); + out.push(item); + }); + return out; +} + +function toggleCsvItem(value: string, item: string) { + const clean = item.trim(); + const current = csvItems(value); + const exists = current.some((x) => x.toLowerCase() === clean.toLowerCase()); + return exists ? current.filter((x) => x.toLowerCase() !== clean.toLowerCase()).join(",") : [...current, clean].join(","); +} + function safeEmbeddingModelName(raw: string) { const value = raw.trim().replace(/[\\/]/g, "--").replace(/[^A-Za-z0-9._-]/g, "-"); return value.slice(0, 96) || "embedding-model"; @@ -90,6 +142,13 @@ export default function SettingsTranslatePage() { const [ragWikiEnabled, setRagWikiEnabled] = useState(false); const [ragSearchEnabled, setRagSearchEnabled] = useState(false); const [ragSearchUrl, setRagSearchUrl] = useState(""); + const [ragSearchCategories, setRagSearchCategories] = useState("general"); + const [ragSearchEngines, setRagSearchEngines] = useState(""); + const [ragSearchFallbackEngines, setRagSearchFallbackEngines] = useState("bing,baidu"); + const [ragSearchLanguage, setRagSearchLanguage] = useState("all"); + const [ragSearchSafesearch, setRagSearchSafesearch] = useState(0); + const [ragSearchTimeRange, setRagSearchTimeRange] = useState(""); + const [ragSearchPageno, setRagSearchPageno] = useState(1); const [ragDomain, setRagDomain] = useState(""); const [ragAgentParallelism, setRagAgentParallelism] = useState(1); const [ragAgentTimeoutSeconds, setRagAgentTimeoutSeconds] = useState(120); @@ -140,6 +199,13 @@ export default function SettingsTranslatePage() { setRagWikiEnabled(s.rag_wiki_enabled ?? false); setRagSearchEnabled(s.rag_search_enabled); setRagSearchUrl(s.rag_search_url); + setRagSearchCategories(s.rag_search_categories || "general"); + setRagSearchEngines(s.rag_search_engines || ""); + setRagSearchFallbackEngines(s.rag_search_fallback_engines || "bing,baidu"); + setRagSearchLanguage(s.rag_search_language || "all"); + setRagSearchSafesearch(s.rag_search_safesearch ?? 0); + setRagSearchTimeRange(s.rag_search_time_range || ""); + setRagSearchPageno(s.rag_search_pageno ?? 1); setRagDomain(s.rag_domain); setRagAgentParallelism(s.rag_agent_parallelism ?? 1); setRagAgentTimeoutSeconds(s.rag_agent_timeout_seconds ?? 120); @@ -183,6 +249,13 @@ export default function SettingsTranslatePage() { rag_wiki_enabled: ragWikiEnabled, rag_search_enabled: ragSearchEnabled, rag_search_url: ragSearchUrl, + rag_search_categories: ragSearchCategories, + rag_search_engines: ragSearchEngines, + rag_search_fallback_engines: ragSearchFallbackEngines, + rag_search_language: ragSearchLanguage, + rag_search_safesearch: ragSearchSafesearch, + rag_search_time_range: ragSearchTimeRange, + rag_search_pageno: ragSearchPageno, rag_domain: ragDomain, rag_agent_parallelism: ragAgentParallelism, rag_agent_timeout_seconds: ragAgentTimeoutSeconds, @@ -335,6 +408,7 @@ export default function SettingsTranslatePage() { ["summary", settings.default_enable_summary ? "true" : "false"], ["rag", settings.rag_enabled ? "enabled" : "disabled"], ["wiki", settings.rag_wiki_enabled ? "enabled" : "disabled"], + ["search", settings.rag_search_enabled ? `${settings.rag_search_categories || "general"} / ${settings.rag_search_language || "all"}` : "disabled"], ["agents", `${settings.rag_agent_parallelism} / ${settings.rag_agent_timeout_seconds}s`], ["embedding", `${settings.rag_embedding_provider}:${settings.rag_embedding_model}`], ["embedding key", settings.rag_embedding_api_key_set ? "set" : "unset"], @@ -551,6 +625,80 @@ export default function SettingsTranslatePage() {
SearXNG Base URL
setRagSearchUrl(e.target.value)} /> +
+
SearXNG categories
+ setRagSearchCategories(e.target.value)} /> +
+ {SEARXNG_CATEGORY_PRESETS.map((item) => { + const selected = csvItems(ragSearchCategories).some((x) => x.toLowerCase() === item.toLowerCase()); + return ( + + ); + })} +
+
+
+
SearXNG engines
+ setRagSearchEngines(e.target.value)} /> +
+ {SEARXNG_ENGINE_PRESETS.map((item) => { + const selected = csvItems(ragSearchEngines).some((x) => x.toLowerCase() === item.toLowerCase()); + return ( + + ); + })} +
+
+ + + + +
diff --git a/tests/test_translation_rag.py b/tests/test_translation_rag.py index bbe38b6..85c788e 100644 --- a/tests/test_translation_rag.py +++ b/tests/test_translation_rag.py @@ -6,6 +6,8 @@ import httpx from videoroll.ai.client import OpenAIChatConfig from videoroll.ai.client import request_openai_embedding +from videoroll.apps.subtitle_service.agent_runtime import AgentBudget, AgentBudgetExceeded, AgentRuntime, AgentTraceEvent +from videoroll.apps.subtitle_service.retrieval import RetrievalPipeline from videoroll.apps.subtitle_service.rag import ( _run_research_agents, _fallback_search_queries, @@ -56,6 +58,13 @@ def test_rag_settings_from_translate_settings_clamps_values() -> None: "rag_wiki_enabled": True, "rag_search_enabled": True, "rag_search_url": "https://search.example", + "rag_search_categories": "general,it,general", + "rag_search_engines": "bing, baidu", + "rag_search_fallback_engines": "brave,duckduckgo", + "rag_search_language": "zh-CN", + "rag_search_safesearch": 2, + "rag_search_time_range": "month", + "rag_search_pageno": 3, "rag_domain": "CS2", "rag_agent_parallelism": 99, "rag_agent_timeout_seconds": 9999, @@ -74,11 +83,88 @@ def test_rag_settings_from_translate_settings_clamps_values() -> None: assert cfg.wiki_enabled is True assert cfg.search_enabled is True assert cfg.search_url == "https://search.example" + assert cfg.search_categories == "general,it" + assert cfg.search_engines == "bing,baidu" + assert cfg.search_fallback_engines == "brave,duckduckgo" + assert cfg.search_language == "zh-CN" + assert cfg.search_safesearch == 2 + assert cfg.search_time_range == "month" + assert cfg.search_pageno == 3 assert cfg.domain == "CS2" assert cfg.agent_parallelism == 8 assert cfg.agent_timeout_seconds == 900.0 +def test_agent_runtime_enforces_budget_and_records_normalized_steps() -> None: + steps: list[dict[str, object]] = [] + runtime = AgentRuntime( + agent_name="rag_term_research", + run_id="run-1", + budget=AgentBudget(max_llm_calls=1, max_tool_calls=1, max_fetch_calls=0, timeout_seconds=30), + trace_recorder=steps.append, + ) + + runtime.before_llm() + runtime.record(AgentTraceEvent(kind="agent", action="started", output={"ok": True})) + + try: + runtime.before_llm() + except AgentBudgetExceeded as e: + assert "LLM call budget" in str(e) + else: + raise AssertionError("expected LLM budget to stop the agent") + + assert steps[0]["run_id"] == "run-1" + assert steps[0]["agent_name"] == "rag_term_research" + assert steps[0]["event_id"] + assert steps[0]["span_id"] + assert steps[0]["status"] == "ok" + assert steps[0]["budget"]["llm_calls"] == 1 # type: ignore[index] + + +def test_retrieval_pipeline_dedupes_records_and_rolls_back_on_error() -> None: + events: list[dict[str, object]] = [] + errors: list[str] = [] + + pipeline = RetrievalPipeline[dict[str, str]]( + id_getter=lambda item: item["id"], + trace_recorder=events.append, + error_handler=lambda e: errors.append(type(e).__name__), + ) + + first = pipeline.run_stage("exact", "AWP", lambda: [{"id": "1"}, {"id": "1"}, {"id": "2"}]) + second = pipeline.run_stage("vector", "AWP", lambda: (_ for _ in ()).throw(RuntimeError("db aborted"))) + + assert first == [{"id": "1"}, {"id": "1"}, {"id": "2"}] + assert second == [] + assert pipeline.hits == [{"id": "1"}, {"id": "2"}] + assert events[0]["action"] == "exact" + assert events[0]["count"] == 2 + assert events[1]["status"] == "failed" + assert events[1]["error_type"] == "RuntimeError" + assert errors == ["RuntimeError"] + + +def test_research_tool_registry_exposes_enabled_tool_schemas() -> None: + from videoroll.apps.subtitle_service import rag as rag_module + + cfg = rag_settings_from_translate_settings( + { + "rag_enabled": True, + "rag_wiki_enabled": True, + "rag_search_enabled": True, + } + ) + specs = rag_module._ordered_tool_specs(rag_module._research_tool_registry(cfg)) + names = [spec["name"] for spec in specs] + + assert names == ["rag_lookup", "wiki_search", "search_web", "fetch_url", "finish"] + search_spec = next(spec for spec in specs if spec["name"] == "search_web") + assert "input_schema" in search_spec + assert "query" in search_spec["input_schema"]["properties"] + assert "filter_search_engine_internal_pages" in search_spec["guardrails"] + + def test_build_knowledge_embedding_text_includes_term_fields() -> None: text = build_knowledge_embedding_text( item_type="term", @@ -645,6 +731,22 @@ def test_search_url_accepts_base_url_and_legacy_search_url() -> None: _search_url_with_params("https://search.example", query="truth table", json_format=True, extra_params={"engines": "bing,baidu"}) == "https://search.example/search?engines=bing%2Cbaidu&q=truth+table&format=json" ) + assert ( + _search_url_with_params( + "https://search.example", + query="VGA 红色信号", + json_format=True, + extra_params={ + "categories": "general,it", + "engines": "bing,baidu", + "language": "zh-CN", + "safesearch": "0", + "time_range": "month", + "pageno": "2", + }, + ) + == "https://search.example/search?categories=general%2Cit&engines=bing%2Cbaidu&language=zh-CN&safesearch=0&time_range=month&pageno=2&q=VGA+%E7%BA%A2%E8%89%B2%E4%BF%A1%E5%8F%B7&format=json" + ) def test_parse_search_json_accepts_searxng_content_results() -> None: