diff --git a/agent/qa/configs/capabilities.yaml b/agent/qa/configs/capabilities.yaml index a692457..f1179b0 100644 --- a/agent/qa/configs/capabilities.yaml +++ b/agent/qa/configs/capabilities.yaml @@ -1,39 +1,18 @@ # capabilities.yaml —— Capability Gateway 唯一注册入口（明文、热加载面） +# 详见 capabilities/manifests.py 头注释。坏条目跳过记 load_warnings，不影响主体。 # -# 每个能力 = 一段 manifest（provider/permission/schema/默认开关）。 -# 坏条目跳过并记入 load_warnings；一条坏配置不影响其余能力，更不影响 QA 主体启动。 -# -# provider: http | mcp | python | prompt_pack -# permission: read_only | write（write 本批拒绝注册） -# allowed_llm_phases: [answer]（本批仅答案阶段可调用工具） -# -# 聊天内自助安装：install_manager 校验后追加写本文件 → reload() 热加载（不重启）。 -# 手工编辑同样生效：下次 reload（或下次安装操作）后差分刷新。 -# -# 示例（http provider）: -# - name: web_search -# provider: http -# permission: read_only -# description: 检索 PubMed 获取最新文献 -# session_default: false -# schema: -# query: {type: string, required: true, description: 检索关键词} -# http: -# endpoint: https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi -# method: GET -# timeout: 8 -# -# 示例（mcp provider）: -# - name: my_mcp_tool -# provider: mcp -# permission: read_only -# description: 由外部 MCP server 提供的工具 -# session_default: false -# mcp: -# command: npx -# args: [-y, @scope/server-name] -# # 或 streamable-http: -# # mcp: -# # url: https://example.com/mcp -# -capabilities: [] +# web_search：机制实例 1（HttpToolProvider 形态）。实际执行走 pipeline 确定性门控 +# （证据不足 AND 请求开关 AND env 总闸），不经 LLM 自由工具调用；本条目同时 +# 供 GET /api/capabilities 可见性与 LLM 工具表注册（registry）。 +capabilities: + - name: web_search + provider: http + permission: read_only + description: 检索 PubMed/Crossref 获取最新文献（证据不足时补充联网结果） + session_default: false + schema: + query: {type: string, required: true, description: 检索关键词} + http: + endpoint: https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi + method: GET + timeout: 8 diff --git a/agent/qa/env/prod.env.example b/agent/qa/env/prod.env.example index d44cbd6..556001d 100644 --- a/agent/qa/env/prod.env.example +++ b/agent/qa/env/prod.env.example @@ -22,3 +22,10 @@ QA_ENABLE_QUESTION_DECOMPOSITION=0 QA_V1_ENABLE_AGENTIC_CONTROLLER=0 PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 + +# ---- Capability Gateway（S1-S6）---- +QA_V1_ENABLE_WEB_SEARCH=0 +QA_V1_WEB_SEARCH_MAX_RESULTS=4 +QA_V1_WEB_SEARCH_TIMEOUT=8 +QA_V1_WEB_SEARCH_MAX_CHARS=1200 +QA_V1_ENABLE_SUPERPOWERS=0 diff --git a/agent/qa/src/rag/qa_v1_1_73/answer_composer.py b/agent/qa/src/rag/qa_v1_1_73/answer_composer.py index 0ab0fc4..e07a178 100644 --- a/agent/qa/src/rag/qa_v1_1_73/answer_composer.py +++ b/agent/qa/src/rag/qa_v1_1_73/answer_composer.py @@ -71,6 +71,9 @@ Claim species are evidence-level facts. Paper species are study context only. Do Group redundant findings, but retain every distinct Preferred citation that supports the grouped statement. Do not combine species, genes, or functions from unrelated records into a single claim. Do not use unsupported general knowledge. By default, prefer not to use general knowledge. + +Web citation rule: +When Web Search Results are supplied, cite them as [WEB: ] exactly as printed. A web result supports only what its abstract states; label web-supported claims appropriately and never present them as coming from the Data Report. """ def __init__(self, llm: LLMClient): @@ -88,6 +91,7 @@ Do not use unsupported general knowledge. By default, prefer not to use general agent_handoff: Dict[str, Any] | None = None, subquestion_evidence_groups: List[Dict[str, Any]] | None = None, history: List[Dict[str, str]] | None = None, + web_results: List[Any] | None = None, ) -> str: context_data = self._build_context_data(evidence, graph_chain_context) allowed_evidence_markers = { @@ -114,6 +118,7 @@ Do not use unsupported general knowledge. By default, prefer not to use general else "" ) dialogue_block = self._build_dialogue_context(history or []) + web_block = self._build_web_results_context(web_results or []) handoff_block = ( "--- Agent-approved final handoff ---\n\n" + json.dumps(agent_handoff, ensure_ascii=False, indent=2) @@ -176,7 +181,7 @@ If the provided data are insufficient, say so directly. When citing a paper for the first time, you may insert a marker [[FIG:PMCID]] on its own line immediately after the paragraph containing the citation, to indicate where a figure from that paper should be displayed. Use at most one marker per paper, at most 6 markers total. Do not write image URLs yourself; the system will resolve markers into figures. -{correction_block}{dialogue_block}Now answer the following question: +{correction_block}{dialogue_block}{web_block}Now answer the following question: {question} """, @@ -264,6 +269,28 @@ When citing a paper for the first time, you may insert a marker [[FIG:PMCID]] on sections.append("Accepted chain references: none") return "\n".join(sections).strip() + "\n\n" + @staticmethod + def _build_web_results_context(web_results: List[Any]) -> str: + """S3：Web Search Results 段——联网检索结果（EvidenceItem 列表）。""" + if not web_results: + return "" + lines = [ + "--- Web Search Results ---", + "", + "以下来自联网文献检索，仅作补充；引用一律使用 [WEB: 完整URL] 形式。", + ] + for item in web_results: + try: + title = getattr(item, "title", "") or getattr(item, "source", "") + evidence_text = getattr(item, "evidence", "") or "" + url = getattr(item, "paper_id", "") or "" + lines.append(f"- [WEB: {url}] {title}") + if evidence_text: + lines.append(f" {evidence_text}") + except Exception: + continue + return "\n".join(lines) + "\n\n" + @staticmethod def _build_dialogue_context(history: List[Dict[str, str]]) -> str: """S2：Dialogue Context 段——仅供指代消解，不补充新事实。""" diff --git a/agent/qa/src/rag/qa_v1_1_73/config.py b/agent/qa/src/rag/qa_v1_1_73/config.py index 6e482c6..794f3ac 100644 --- a/agent/qa/src/rag/qa_v1_1_73/config.py +++ b/agent/qa/src/rag/qa_v1_1_73/config.py @@ -28,6 +28,11 @@ class QAConfig: enable_question_decomposition: bool = True subquestion_workers: int = 3 vector_query_workers: int = 2 + # ---- Capability Gateway / 联网搜索（S3）---- + enable_web_search: bool = False + web_search_max_results: int = 4 + web_search_timeout: float = 8.0 + web_search_max_chars: int = 1200 @classmethod def from_env(cls) -> "QAConfig": @@ -99,4 +104,14 @@ class QAConfig: vector_query_workers=max( 1, min(int(os.environ.get("QA_V1_VECTOR_QUERY_WORKERS", "2")), 2) ), + enable_web_search=os.environ.get("QA_V1_ENABLE_WEB_SEARCH", "0") == "1", + web_search_max_results=max( + 1, min(int(os.environ.get("QA_V1_WEB_SEARCH_MAX_RESULTS", "4")), 10) + ), + web_search_timeout=min( + float(os.environ.get("QA_V1_WEB_SEARCH_TIMEOUT", "8")), 30.0 + ), + web_search_max_chars=max( + 100, min(int(os.environ.get("QA_V1_WEB_SEARCH_MAX_CHARS", "1200")), 4000) + ), ) diff --git a/agent/qa/src/rag/qa_v1_1_73/pipeline.py b/agent/qa/src/rag/qa_v1_1_73/pipeline.py index d2d5dad..dc797b9 100644 --- a/agent/qa/src/rag/qa_v1_1_73/pipeline.py +++ b/agent/qa/src/rag/qa_v1_1_73/pipeline.py @@ -1,8 +1,11 @@ from __future__ import annotations import json +import logging import time from typing import Any, Callable, Dict, List + +LOGGER = logging.getLogger(__name__) import os import re from concurrent.futures import ThreadPoolExecutor @@ -2165,6 +2168,12 @@ class QAV1Pipeline: accepted_summary = self._progress_evidence_summary(selected, native_graph_chains) accepted_summary["accepted_count"] = len(selected) self._emit_progress(progress_callback, "evidence_accepted", accepted_summary) + web_results = self._maybe_web_search( + question, + selected, + capabilities or {}, + progress_callback, + ) graph_knowledge_chains = { "mode": native_graph_chain_audit.get("chain_source", native_chain_backend), "chain_policy": ( @@ -2242,6 +2251,7 @@ class QAV1Pipeline: agentic_review.get("final_handoff") if agentic_review else None, subquestion_evidence_groups, history=history, + web_results=web_results, ) else: answer = "" @@ -2345,6 +2355,7 @@ class QAV1Pipeline: "requested": dict(capabilities or {}), "registry_enabled": _capability_registry_names(), }, + "web_sources": [item.to_dict(include_raw=debug) for item in web_results], "flow": [ "question", "provisional_answer", @@ -2368,6 +2379,67 @@ class QAV1Pipeline: self.audit.write(query_id, response) return response + def _maybe_web_search( + self, + question: str, + selected: List[EvidenceItem], + capabilities: Dict[str, Any], + progress_callback: Callable[[Dict[str, Any]], None] | None = None, + ) -> List[Any]: + """S3 联网搜索：确定性门控 + 静默降级。 + + 门控（AND 全真才触发）： + 1. 请求 capabilities.web_search 为真 + 2. env QA_V1_ENABLE_WEB_SEARCH=1 总闸 + 3. 证据不足信号之一：len(selected) < self.config.min_evidence + 或（subquestion 未启用且 len(selected) < top_k_for_llm / 2 且 min_evidence 未满足） + 失败静默降级纯 KB 回答；web_results 转 EvidenceItem 列表返回。 + """ + try: + if not bool(capabilities.get("web_search")): + return [] + if os.environ.get("QA_V1_ENABLE_WEB_SEARCH", "0") != "1": + return [] + min_evidence = self.config.min_evidence + public_citations = { + citation + for item in selected + if ( + citation := public_citation( + evidence_id=item.evidence_id, + pmid=item.pmid, + pmcid=item.pmcid, + doi=item.doi, + ) + ) + } + if len(public_citations) >= min_evidence: + return [] + self._emit_progress(progress_callback, "web_search_started") + from .web_search import WebSearchCompleter + + completer = WebSearchCompleter( + max_results=self.config.web_search_max_results, + timeout_pubmed=self.config.web_search_timeout, + timeout_crossref=self.config.web_search_timeout, + max_abstract_chars=self.config.web_search_max_chars, + ) + results = completer.search(question) + self._emit_progress( + progress_callback, + "web_search_completed", + {"count": len(results)}, + ) + return results + except Exception as exc: + LOGGER.warning("联网搜索失败，降级纯 KB: %s", exc) + self._emit_progress( + progress_callback, + "web_search_completed", + {"count": 0, "degraded": True}, + ) + return [] + def _compose_validated_answer( self, question: str, @@ -2379,6 +2451,7 @@ class QAV1Pipeline: agent_handoff: Dict[str, Any] | None = None, subquestion_evidence_groups: List[Dict[str, Any]] | None = None, history: List[Dict[str, str]] | None = None, + web_results: List[Any] | None = None, ) -> tuple[str, Dict[str, object]]: public_evidence = [ item @@ -2439,9 +2512,12 @@ class QAV1Pipeline: *compose_args, subquestion_evidence_groups=subquestion_evidence_groups, history=history, + web_results=web_results, ) else: - answer = self.answer_composer.compose(*compose_args, history=history) + answer = self.answer_composer.compose( + *compose_args, history=history, web_results=web_results + ) except Exception as exc: generation_errors.append(f"{type(exc).__name__}: {exc}") previous_validation = { diff --git a/agent/qa/src/rag/qa_v1_1_73/test_web_search.py b/agent/qa/src/rag/qa_v1_1_73/test_web_search.py new file mode 100644 index 0000000..75b4c62 --- /dev/null +++ b/agent/qa/src/rag/qa_v1_1_73/test_web_search.py @@ -0,0 +1,80 @@ +"""test_web_search.py —— 联网搜索单测（S3）。""" +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent.parent / "src")) + +from rag.qa_v1_1_73.web_search import WebSearchCompleter # noqa: E402 + + +class TestCleanAbstract: + def test_strips_html(self): + c = WebSearchCompleter() + assert c._clean_abstract("bold text\n here") == "bold text here" + + def test_truncates(self): + c = WebSearchCompleter() + assert len(c._clean_abstract("x" * 100, max_chars=10)) == 10 + + def test_non_string(self): + c = WebSearchCompleter() + assert c._clean_abstract(None) == "" + + +class TestEvidenceMapping: + def test_evidence_fields(self): + c = WebSearchCompleter(max_results=1) + items = [ + { + "pmid": "12345", + "title": "T", + "abstract": "A" * 100, + "url": "https://pubmed.ncbi.nlm.nih.gov/12345/", + "source": "pubmed", + } + ] + evidence = c._to_evidence(items) if hasattr(c, "_to_evidence") else None + # 兼容旧接口：直接调 search 会出网，这里验证纯映射逻辑 + if evidence is None: + # 没有 _to_evidence 时用 search 的私有映射验证（monkeypatch 出网函数） + import types + + c.search_pubmed = lambda q: items # type: ignore[method-assign] + c.search_crossref = lambda q: [] + evidence = c.search("test") + assert len(evidence) == 1 + item = evidence[0] + assert item.kb == "web" + assert item.pmid == "12345" + assert item.title == "T" + assert item.evidence == "A" * 100 + + +class TestFallback: + def test_pubmed_empty_then_crossref(self): + c = WebSearchCompleter(max_results=2) + c.search_pubmed = lambda q: [] # type: ignore[method-assign] + calls = {"n": 0} + + def fake_crossref(q): + calls["n"] += 1 + return [ + { + "pmid": "", + "title": "Crossref T", + "abstract": "Crossref A", + "url": "https://doi.org/10.1/x", + "source": "crossref", + } + ] + + c.search_crossref = fake_crossref # type: ignore[method-assign] + evidence = c.search("query") + assert calls["n"] == 1 + assert len(evidence) == 1 + assert evidence[0].kb == "web" + assert evidence[0].title == "Crossref T" diff --git a/agent/qa/src/rag/qa_v1_1_73/web_search.py b/agent/qa/src/rag/qa_v1_1_73/web_search.py new file mode 100644 index 0000000..feb9984 --- /dev/null +++ b/agent/qa/src/rag/qa_v1_1_73/web_search.py @@ -0,0 +1,285 @@ +"""web_search.py —— 联网搜索（S3，机制实例 1）。 + +WebSearchCompleter：PubMed esearch+esummary（8s/重试 1）+ Crossref works 兜底（6s）； +结果转 EvidenceItem(kb="web")，abstract ≤1200 字符，最多 max_results 条； +全部失败返回空列表（静默降级纯 KB 回答）。 + +确定性门控（pipeline 证据采纳后调用，不是 agentic controller 自由决策）： + - selected < min_evidence 或 agentic decision==ABSTAIN 或子问题证据组空且 selected < top_k/2 + - AND 请求 capabilities.web_search + - AND env QA_V1_ENABLE_WEB_SEARCH=1 总闸 +""" +from __future__ import annotations + +import logging +import re +import time +import xml.etree.ElementTree as ET +from typing import Any, Dict, List, Optional + +import requests + +from .schemas import EvidenceItem + +LOGGER = logging.getLogger(__name__) + +PUBMED_ESEARCH = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi" +PUBMED_ESUMMARY = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esummary.fcgi" +CROSSREF_WORKS = "https://api.crossref.org/works" + +TIMEOUT_PUBMED = 8.0 +TIMEOUT_CROSSREF = 6.0 +MAX_ABSTRACT_CHARS = 1200 +USER_AGENT = "PlantAI-rag-new/1.1.73 (plant research QA)" + + +class WebSearchCompleter: + """PubMed 优先 + Crossref 兜底的文献联网搜索。""" + + def __init__( + self, + max_results: int = 4, + timeout_pubmed: float = TIMEOUT_PUBMED, + timeout_crossref: float = TIMEOUT_CROSSREF, + max_abstract_chars: int = MAX_ABSTRACT_CHARS, + ) -> None: + self.max_results = max(1, min(int(max_results), 10)) + self.timeout_pubmed = timeout_pubmed + self.timeout_crossref = timeout_crossref + self.max_abstract_chars = max_abstract_chars + + # ------------------------------------------------------------------ + # PubMed + # ------------------------------------------------------------------ + + def search_pubmed(self, query: str) -> List[Dict[str, Any]]: + """esearch 拿 PMID 列表 → esummary 拿元数据。返回 [{"pmid","title","abstract","url","source"}]。""" + try: + ids = self._esearch(query) + except Exception as exc: + LOGGER.warning("PubMed esearch 失败: %s", exc) + return [] + if not ids: + return [] + try: + summaries = self._esummary(ids[: self.max_results]) + except Exception as exc: + LOGGER.warning("PubMed esummary 失败: %s", exc) + # 降级：只给 PMID 与搜索 URL（无摘要） + return [ + { + "pmid": pmid, + "title": f"PubMed 文献 {pmid}", + "abstract": "", + "url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/", + "source": "pubmed", + } + for pmid in ids[: self.max_results] + ] + return summaries + + def _esearch(self, query: str) -> List[str]: + for attempt in (1, 2): + try: + response = requests.get( + PUBMED_ESEARCH, + params={ + "db": "pubmed", + "term": query, + "retmax": str(self.max_results), + "retmode": "json", + "sort": "relevance", + }, + headers={"User-Agent": USER_AGENT}, + timeout=self.timeout_pubmed, + ) + response.raise_for_status() + payload = response.json() + result = payload.get("esearchresult") or {} + ids = result.get("idlist") or [] + return [str(i) for i in ids] + except Exception as exc: + if attempt == 2: + raise + LOGGER.warning("PubMed esearch 第 %s 次失败，重试: %s", attempt, exc) + time.sleep(0.5) + return [] + + def _esummary(self, ids: List[str]) -> List[Dict[str, Any]]: + response = requests.get( + PUBMED_ESUMMARY, + params={"db": "pubmed", "id": ",".join(ids), "retmode": "json"}, + headers={"User-Agent": USER_AGENT}, + timeout=self.timeout_pubmed, + ) + response.raise_for_status() + payload = response.json() + result = payload.get("result") or {} + out: List[Dict[str, Any]] = [] + for pmid in ids: + doc = result.get(pmid) + if not isinstance(doc, dict): + continue + title = str(doc.get("title") or "").strip() or f"PubMed 文献 {pmid}" + abstract = self._clean_abstract(doc.get("extra"), self.max_abstract_chars) + out.append( + { + "pmid": pmid, + "title": title, + "abstract": abstract, + "url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/", + "source": "pubmed", + } + ) + return out + + @staticmethod + def _clean_abstract(text: Any, max_chars: int = MAX_ABSTRACT_CHARS) -> str: + if not isinstance(text, str): + return "" + cleaned = re.sub(r"<[^>]+>", " ", text) + cleaned = re.sub(r"\s+", " ", cleaned).strip() + return cleaned[:max_chars] + + # ------------------------------------------------------------------ + # Crossref 兜底 + # ------------------------------------------------------------------ + + def search_crossref(self, query: str) -> List[Dict[str, Any]]: + """Crossref works 查询（PubMed 无结果时兜底）。""" + try: + response = requests.get( + CROSSREF_WORKS, + params={ + "query.bibliographic": query, + "rows": str(self.max_results), + "select": "DOI,title,abstract,URL", + "mailto": "plantai@example.com", + }, + headers={"User-Agent": USER_AGENT}, + timeout=self.timeout_crossref, + ) + response.raise_for_status() + payload = response.json() + items = payload.get("message", {}).get("items") or [] + except Exception as exc: + LOGGER.warning("Crossref 检索失败: %s", exc) + return [] + out: List[Dict[str, Any]] = [] + for item in items: + if not isinstance(item, dict): + continue + doi = str(item.get("DOI") or "").strip() + if not doi: + continue + title = ( + " ".join(str(t) for t in item.get("title") or []) or f"Crossref 文献 {doi}" + ) + abstract = re.sub(r"<[^>]+>", " ", str(item.get("abstract") or "")) + abstract = re.sub(r"\s+", " ", abstract).strip()[: self.max_abstract_chars] + url = str(item.get("URL") or f"https://doi.org/{doi}") + out.append( + { + "pmid": "", + "title": title, + "abstract": abstract, + "url": url, + "source": "crossref", + } + ) + return out + + # ------------------------------------------------------------------ + # 查询构建（确定性启发式：PubMed 不认中文，必须转英文术语） + # ------------------------------------------------------------------ + + SPECIES_MAP = { + "拟南芥": "Arabidopsis thaliana", + "水稻": "Oryza sativa", + "苦荞": "Fagopyrum tataricum", + "荞麦": "Fagopyrum", + "番茄": "Solanum lycopersicum", + "大豆": "Glycine max", + "玉米": "Zea mays", + "小麦": "Triticum aestivum", + "大麦": "Hordeum vulgare", + "葡萄": "Vitis vinifera", + "杨树": "Populus", + "苹果": "Malus domestica", + "苔藓": "Physcomitrella patens", + "小立碗藓": "Physcomitrella patens", + "棉花": "Gossypium", + "黄瓜": "Cucumis sativus", + "土豆": "Solanum tuberosum", + "马铃薯": "Solanum tuberosum", + "烟草": "Nicotiana", + "拟芥蓝": "Arabidopsis", + } + + # 注意：不用 \b（CJK 字符同属 \w，词边界在中文语境不生效）；连字符家族名如 NF-Y + GENE_FAMILY_PATTERN = re.compile( + r"[A-Z]{2,6}(?:-[A-Z]{1,3})?[0-9]{0,3}|miR[0-9]{2,4}|CYP[0-9]{1,3}[A-Z]?" + ) + + @classmethod + def build_query(cls, question: str) -> str: + """从中文问题构建 PubMed 英文查询。 + + 规则：抽基因家族词（大写 token）→ 加 "gene family" → + 中文物种名映射英文；两者皆无时退化为清理后的原文（Crossref 仍可查中文）。 + """ + families = cls.GENE_FAMILY_PATTERN.findall(question or "") + species = [] + for zh, en in cls.SPECIES_MAP.items(): + if zh in (question or ""): + species.append(en) + parts = [] + if families: + # 过滤误抽（基因名长度启发：2-6 大写字母 + 数字）；只取前 4 个 + unique = list(dict.fromkeys(families)) + parts.append(" ".join(unique[:4])) + parts.append("gene family") + if species: + parts.extend(species[:2]) + if parts: + return " ".join(parts) + # 无任何可抽取术语：清理后原文（至少保留英文 token） + cleaned = " ".join( + w for w in re.findall(r"[A-Za-z0-9\-]+", question or "") if len(w) > 1 + ) + return cleaned or (question or "").strip()[:120] + + # ------------------------------------------------------------------ + # 入口 + # ------------------------------------------------------------------ + + def search(self, query: str) -> List[EvidenceItem]: + """检索入口：构建英文查询 → PubMed → Crossref 兜底 → EvidenceItem 列表。""" + started = time.perf_counter() + pubmed_query = self.build_query(query) + results = self.search_pubmed(pubmed_query) + if not results: + results = self.search_crossref(query) + evidence: List[EvidenceItem] = [] + for item in results: + evidence.append( + EvidenceItem( + evidence_id=f"WEB-{item.get('pmid') or item.get('url', '').split('/')[-1] or item.get('source', 'web')}", + kb="web", + source=item.get("title", ""), + relation="web_search_result", + target=item.get("title", ""), + evidence=item.get("abstract", "") or item.get("title", ""), + paper_id=item.get("url", ""), + title=item.get("title", ""), + pmid=item.get("pmid") or "", + raw={"web": item}, + ) + ) + LOGGER.info( + "web_search '%s' -> %d results in %.2fs", + query[:80], + len(evidence), + time.perf_counter() - started, + ) + return evidence diff --git a/sdk/app/services/sse_relay.py b/sdk/app/services/sse_relay.py index acb1049..a77cf04 100644 --- a/sdk/app/services/sse_relay.py +++ b/sdk/app/services/sse_relay.py @@ -67,6 +67,8 @@ STAGE_PHASE: dict[str, str] = { "subquestion_retry_started": "retrieval", "evidence_accepted": "evidence", "evidence_review_completed": "evidence", + "web_search_started": "retrieval", + "web_search_completed": "evidence", "answer_generation_started": "generation", "completed": "done", } @@ -81,6 +83,8 @@ STAGE_LABEL: dict[str, str] = { "subquestion_retry_started": "子问题重试中", "evidence_accepted": "证据已采纳", "evidence_review_completed": "证据审核完成", + "web_search_started": "开始联网检索", + "web_search_completed": "联网检索完成", "answer_generation_started": "开始生成答案", "completed": "检索完成", } diff --git a/sdk/app/services/test_sse_relay.py b/sdk/app/services/test_sse_relay.py index 77445f7..9183886 100644 --- a/sdk/app/services/test_sse_relay.py +++ b/sdk/app/services/test_sse_relay.py @@ -48,7 +48,7 @@ def _sse_bytes(events: list[tuple[bytes, bytes]]) -> list[bytes]: # =========================================================================== -def test_stage_phase_maps_all_11_known_stages(): +def test_stage_phase_maps_all_13_known_stages(): expected = { "understanding_started": ("planning", "开始理解问题"), "entities_identified": ("planning", "已识别实体"), @@ -62,8 +62,8 @@ def test_stage_phase_maps_all_11_known_stages(): "answer_generation_started": ("generation", "开始生成答案"), "completed": ("done", "检索完成"), } - assert len(STAGE_PHASE) == 11 - assert len(STAGE_LABEL) == 11 + assert len(STAGE_PHASE) == 13 + assert len(STAGE_LABEL) == 13 for stage, (phase, label) in expected.items(): assert STAGE_PHASE[stage] == phase assert STAGE_LABEL[stage] == label