fix: disable ssl verification for intranet ai writing search

This commit is contained in:
董宏凯 committed 2026-07-01 10:21:52 +08:00
1 parent 8fbfd35296
commit 4bfaeb728b
6 files changed
+117 -18

No files matched your search

+1 -1
View File
@@ -273,7 +273,7 @@ The AI-writing graph (`packages/harness/deerflow/agents/ai_writing/`) has four s
Seeding mirrors the roundtable pattern: verified `config.yaml` + `SOUL.md` ship in `app/gateway/routers/_ai_writing_seed_assets/<id>/`; `_ai_writing_seed.py::ensure_ai_writing_functional_agents()` copies them to `.deer-flow/agents/` when missing (never overwrites local edits), called from the startup lifespan (before `_sync_legacy_agents`, so the agents appear in `/api/agents` on first boot) and self-heals from `GET /api/ai-writing/sessions` + `GET /api/ai-writing/article-types`. Tests: `tests/test_ai_writing_seed.py`, `tests/test_ai_writing_agent_runner.py`.
**Skill-driven retrieval (检索类型).** `material_source` has three values: `general` (通用检索, default), `knowledge_base` (知识库检索), and `notebook` (我的空间); legacy `skill` is normalized to `general` by the researcher node. **`knowledge_base` 直连内网 ES 接口(= 旧版「内网直连检索」)** — it bypasses skill orchestration entirely and calls `IntranetSearchProvider` over `ai_writing.intranet_search_url` (concurrent over the generated keywords; `intranet_knowledge_base`/`intranet_search_timeout` apply). **An empty/unconfigured `intranet_search_url` does NOT disable it** — the empty string is passed straight to `IntranetSearchProvider`, which falls back to its built-in default endpoint (`intranet_stub._DEFAULT_ENDPOINT`), exactly like the legacy `get_search_provider(search_provider=intranet)` path. So 内网直连检索 works out of the box without configuring an address (configure `intranet_search_url` only to point at a different endpoint). This is the "弱模型识别不了技能" escape hatch — the old no-skill retrieval path restored as an explicit option (weak models like deepseek-chat often fail to invoke configured skills). Tests: `tests/test_ai_writing_knowledge_base.py`. General retrieval is orchestrated from the **skills configured on the `ai-writing-researcher` agent** (`config.yaml → skills`, editable from the agent management page): the built-in `knowledge-base-search` skill (知识库检索) runs the native keyword fan-out web/intranet search (the old default path), while any other configured skill searches with a `description`-targeted query; skills run in configured order with early-stop at `max_materials`, and an empty/unresolvable skills config falls back to the built-in knowledge search so retrieval always works. The skill itself is seeded from `_ai_writing_seed_assets/skills/knowledge-base-search/SKILL.md` into `skills/public/` (never overwrites), and a pre-existing researcher `config.yaml` lacking the `skills` key is patched in place with the default (explicit `skills: []` is respected). **PAUSE-1 input append flow**: when the user types natural language into the material-confirm input (`re_search`/legacy `enable_skill_search` + `user_query`), the node matches the most relevant configured skill via `SkillPicker(candidates=…)` (skipped when only one skill is configured), searches with the user's words, and **appends** the new materials to the existing package (URL-dedup, original order preserved, no LLM re-rank/truncation, 100-item hard cap). **Skill-failure fault tolerance**: a configured skill that times out / raises / has all its keyword searches fail (vs. legitimately returning 0 results) no longer just silently moves on — the researcher node emits a `skill_search_step` of type `skill_error` (carrying the error `reason`), and if the configured skills collected **no** new materials at all it runs a `fallback` pass over other available skills (`_fallback_candidate_skills`: the built-in `knowledge-base-search` native path first, then other enabled catalog skills not yet tried) instead of erroring out. The frontend `SkillStepsCard` renders `skill_error` (rose) and `fallback` (amber) so the timeline shows "技能报错 → 改用其他技能搜集". Tests: `tests/test_ai_writing_material_source.py`.
**Skill-driven retrieval (检索类型).** `material_source` has three values: `general` (通用检索, default), `knowledge_base` (知识库检索), and `notebook` (我的空间); legacy `skill` is normalized to `general` by the researcher node. **`knowledge_base` 直连内网 ES 接口(= 旧版「内网直连检索」)** — it bypasses skill orchestration entirely and calls `IntranetSearchProvider` over `ai_writing.intranet_search_url` (concurrent over the generated keywords; `intranet_knowledge_base`/`intranet_search_timeout` apply). The client uses `trust_env=False` and defaults `intranet_verify_ssl=false`, so internal HTTPS endpoints with self-signed/internal-CA certificates do not fail on certificate verification; set `ai_writing.intranet_verify_ssl: true` only for environments that require strict certificate checks. **An empty/unconfigured `intranet_search_url` does NOT disable it** — the empty string is passed straight to `IntranetSearchProvider`, which falls back to its built-in default endpoint (`intranet_stub._DEFAULT_ENDPOINT`), exactly like the legacy `get_search_provider(search_provider=intranet)` path. So 内网直连检索 works out of the box without configuring an address (configure `intranet_search_url` only to point at a different endpoint). This is the "弱模型识别不了技能" escape hatch — the old no-skill retrieval path restored as an explicit option (weak models like deepseek-chat often fail to invoke configured skills). Tests: `tests/test_ai_writing_knowledge_base.py`. General retrieval is orchestrated from the **skills configured on the `ai-writing-researcher` agent** (`config.yaml → skills`, editable from the agent management page): the built-in `knowledge-base-search` skill (知识库检索) runs the native keyword fan-out web/intranet search (the old default path), while any other configured skill searches with a `description`-targeted query; skills run in configured order with early-stop at `max_materials`, and an empty/unresolvable skills config falls back to the built-in knowledge search so retrieval always works. The skill itself is seeded from `_ai_writing_seed_assets/skills/knowledge-base-search/SKILL.md` into `skills/public/` (never overwrites), and a pre-existing researcher `config.yaml` lacking the `skills` key is patched in place with the default (explicit `skills: []` is respected). **PAUSE-1 input append flow**: when the user types natural language into the material-confirm input (`re_search`/legacy `enable_skill_search` + `user_query`), the node matches the most relevant configured skill via `SkillPicker(candidates=…)` (skipped when only one skill is configured), searches with the user's words, and **appends** the new materials to the existing package (URL-dedup, original order preserved, no LLM re-rank/truncation, 100-item hard cap). **Skill-failure fault tolerance**: a configured skill that times out / raises / has all its keyword searches fail (vs. legitimately returning 0 results) no longer just silently moves on — the researcher node emits a `skill_search_step` of type `skill_error` (carrying the error `reason`), and if the configured skills collected **no** new materials at all it runs a `fallback` pass over other available skills (`_fallback_candidate_skills`: the built-in `knowledge-base-search` native path first, then other enabled catalog skills not yet tried) instead of erroring out. The frontend `SkillStepsCard` renders `skill_error` (rose) and `fallback` (amber) so the timeline shows "技能报错 → 改用其他技能搜集". Tests: `tests/test_ai_writing_material_source.py`.
**Skill-call timeout** is hardcoded **300s (5 min)** — `researcher.py` `_SKILL_SEARCH_TIMEOUT_SECONDS` (per-skill wrapper) and `_skill_agent.py` `_REAL_AGENT_TIMEOUT_SECONDS` (whole real-agent skill run). The intranet ES fallback (`IntranetSearchProvider`) gets a longer **600s (10 min)** default (`intranet_stub.py`; overridable via `ai_writing.intranet_search_timeout`) — that interface is the slow one, so our own fallback timeout is generous to avoid being blamed for "检不出".
@@ -135,6 +135,10 @@ FastAPI application providing REST endpoints for frontend integration:
Custom agents are indexed in the `agents` database table. The `id` column is the stable runtime identifier and filesystem directory name (`.deer-flow/agents/{id}/SOUL.md`); `name` is display-only, may be Chinese, and may be duplicated. Rows with `user_id = NULL` are built-in agents, user rows are private unless `published = true`, and users can only update or delete agents they created.
### AI Writing Intranet Retrieval
AI 写作的「知识库检索」会直连 `ai_writing.intranet_search_url` 内网 ES/知识库接口,并默认跳过代理环境变量和 HTTPS 证书校验,适配自签名或内部 CA 证书的内网服务。需要强校验时可在 `config.yaml` 里设置 `ai_writing.intranet_verify_ssl: true`。
### IM Channels
The IM bridge supports Feishu, Slack, and Telegram. Slack and Telegram still use the final `runs.wait()` response path, while Feishu now streams through `runs.stream(["messages-tuple", "values"])` and updates a single in-thread card in place.
@@ -27,13 +27,15 @@ from types import SimpleNamespace
from langchain_core.runnables import RunnableConfig
from ..search import get_search_provider, SearchResult
from ..prompts.researcher_prompts import (
EXTRACT_KEYWORDS_SYSTEM,
EXTRACT_KEYWORDS_USER,
RANK_MATERIALS_SYSTEM,
RANK_MATERIALS_USER,
)
from ..search import SearchResult, get_search_provider
from ..search.skill_search import SkillPicker
from ..state import AIWritingState, Material, MaterialPackage
from ..prompts.researcher_prompts import (
EXTRACT_KEYWORDS_SYSTEM, EXTRACT_KEYWORDS_USER,
RANK_MATERIALS_SYSTEM, RANK_MATERIALS_USER,
)
from ._agent_runner import (
KEYWORDS_READY_MARKER,
MATERIALS_RANKED_MARKER,
@@ -620,6 +622,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
# 内网慢检索超时可在 config.yaml 的 ai_writing.intranet_search_timeout 调大,
# 默认 600s,且最低 300s。
timeout=_intranet_timeout_seconds(ai_writing_config),
verify_ssl=ai_writing_config.get("intranet_verify_ssl", False),
)
endpoint_label = intranet_url or getattr(provider, "endpoint", "内置默认内网端点")
intranet_debug = {
@@ -628,6 +631,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
"endpoint": endpoint_label,
"knowledge_base": ai_writing_config.get("intranet_knowledge_base", "samples"),
"timeout": _intranet_timeout_seconds(ai_writing_config),
"verify_ssl": getattr(provider, "verify_ssl", False),
"status": "running",
"queries": [],
"calls": [],
@@ -780,7 +784,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
timeout=_SKILL_SEARCH_TIMEOUT_SECONDS,
)
return (results or []), ""
except asyncio.TimeoutError:
except TimeoutError:
msg = f"检索超时(>{_SKILL_SEARCH_TIMEOUT_SECONDS}s)"
logger.warning("skill[%s] search timed out after %ss", skill_name, _SKILL_SEARCH_TIMEOUT_SECONDS)
search_errors.append(f"skill[{skill_name}]: {msg}")
@@ -923,13 +927,14 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
intranet_url = (ai_writing_config.get("intranet_search_url") or "").strip()
# 诊断开关:开启后即使技能成功也照样跑完内网并记录完整结果(用于验证内网接口是否真能通)。
intranet_debug_force = bool(ai_writing_config.get("intranet_search_debug", False))
intranet_task: "asyncio.Task | None" = None
intranet_task: asyncio.Task | None = None
intranet_holder: dict = {}
intranet_debug = {
"configured": bool(intranet_url),
"endpoint": intranet_url or "(未配置 ai_writing.intranet_search_url)",
"knowledge_base": ai_writing_config.get("intranet_knowledge_base", "samples"),
"timeout": _intranet_timeout_seconds(ai_writing_config),
"verify_ssl": bool(ai_writing_config.get("intranet_verify_ssl", False)),
"debug_force": intranet_debug_force,
"status": "pending" if intranet_url else "not_configured",
"queries": [], # 每个检索词:{query, count, error?}
@@ -952,6 +957,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
# 兜底接口超时可在 config.yaml 的 ai_writing.intranet_search_timeout 调大,
# 默认 600s(10 分钟),且最低 300s。
timeout=_intranet_timeout_seconds(ai_writing_config),
verify_ssl=ai_writing_config.get("intranet_verify_ssl", False),
)
intranet_holder["provider"] = intranet
fb_keywords = keywords[:keyword_count] or [intent]
@@ -1,8 +1,8 @@
from .base import BaseSearchProvider, SearchResult
from .internet_search import InternetSearchProvider
from .intranet_stub import IntranetSearchProvider, normalize_intranet_timeout
from .demo_search import DemoSearchProvider
from .duckduckgo_search import DuckDuckGoSearchProvider
from .internet_search import InternetSearchProvider
from .intranet_stub import IntranetSearchProvider, normalize_intranet_timeout, normalize_intranet_verify_ssl
from .skill_search import SkillSearchProvider
@@ -14,6 +14,7 @@ def get_search_provider(ai_writing_config: dict) -> BaseSearchProvider:
intranet_search_url: http://10.106.57.180:7863/knowledge_base/search_docs_es
intranet_knowledge_base: samples
intranet_search_timeout: 300 # 内网检索超时秒数,最少 300 秒
intranet_verify_ssl: false # 内网 HTTPS 自签名证书场景默认关闭证书校验
ddg_region: "cn-zh" # DuckDuckGo 区域,默认 cn-zh
internet 模式:DuckDuckGo 为主搜索(免 API Key),失败时 fallback 到 demo 演示素材。
@@ -26,6 +27,7 @@ def get_search_provider(ai_writing_config: dict) -> BaseSearchProvider:
base_url=cfg.get("intranet_search_url", ""),
knowledge_base_name=cfg.get("intranet_knowledge_base", "samples"),
timeout=normalize_intranet_timeout(cfg.get("intranet_search_timeout")),
verify_ssl=normalize_intranet_verify_ssl(cfg.get("intranet_verify_ssl")),
)
if provider == "tavily":
return _FallbackSearchProvider(InternetSearchProvider(), DemoSearchProvider())
@@ -3,7 +3,6 @@ from __future__ import annotations
import logging
import time
from typing import List
from .base import BaseSearchProvider, SearchResult, simplify_query
@@ -15,6 +14,8 @@ _DEFAULT_KB = "samples"
# 内网检索最少等待 5 分钟;该接口经常慢返回,低于这个值会造成素材收集误判为空。
_MIN_TIMEOUT_SECONDS = 300
_DEFAULT_TIMEOUT_SECONDS = 600
# 内网 HTTPS 检索服务常使用自签名/内部 CA 证书,默认跳过证书校验避免 SSL 证书链错误。
_DEFAULT_VERIFY_SSL = False
# 单条原始响应体最大保留长度,避免把超大响应整段灌进前端时间线
_RAW_SNIPPET_LIMIT = 2000
@@ -28,6 +29,23 @@ def normalize_intranet_timeout(timeout: object, default: int = _DEFAULT_TIMEOUT_
return max(value, _MIN_TIMEOUT_SECONDS)
def normalize_intranet_verify_ssl(value: object, default: bool = _DEFAULT_VERIFY_SSL) -> bool:
"""返回内网检索是否校验 SSL 证书,兼容 config.yaml 里的 bool / 字符串写法。"""
if isinstance(value, bool):
return value
if value is None:
return default
if isinstance(value, (int, float)):
return bool(value)
if isinstance(value, str):
normalized = value.strip().lower()
if normalized in {"1", "true", "yes", "on"}:
return True
if normalized in {"0", "false", "no", "off"}:
return False
return default
def _pick(d: dict, keys: tuple) -> str:
"""从结果项里按候选字段名取第一个非空值(两套接口字段名不同,做并集兼容)。"""
for k in keys:
@@ -62,12 +80,14 @@ class IntranetSearchProvider(BaseSearchProvider):
base_url: str = "",
knowledge_base_name: str = _DEFAULT_KB,
timeout: int = _DEFAULT_TIMEOUT_SECONDS,
verify_ssl: bool = _DEFAULT_VERIFY_SSL,
**_ignored,
):
self.endpoint = base_url or _DEFAULT_ENDPOINT
self.knowledge_base_name = knowledge_base_name or _DEFAULT_KB
# 内网 ES 接口很慢,兜底检索超时默认 10 分钟,且无论配置如何至少 5 分钟。
self.timeout = normalize_intranet_timeout(timeout)
self.verify_ssl = normalize_intranet_verify_ssl(verify_ssl)
# 收集本次运行的检索异常,供 researcher 节点取出并提示用户(ES 波动等)
self.errors: list[str] = []
# 解析失败时收集的原始响应体,供前端「错误返回结果」折叠展示
@@ -82,7 +102,7 @@ class IntranetSearchProvider(BaseSearchProvider):
if snippet:
self.raw_responses.append(snippet)
async def search(self, query: str, max_results: int = 5, **kwargs) -> List[SearchResult]:
async def search(self, query: str, max_results: int = 5, **kwargs) -> list[SearchResult]:
results = await self._search_once(query, max_results)
# 0 结果兜底:复合检索词常检不出(如「特朗普竞选」),换成更核心的检索词重试一次
# (「特朗普竞选」→「特朗普」)。仅在首检为空、且能简化出不同的词时才重试。
@@ -93,7 +113,7 @@ class IntranetSearchProvider(BaseSearchProvider):
results = await self._search_once(simpler, max_results)
return results
async def _search_once(self, query: str, max_results: int = 5) -> List[SearchResult]:
async def _search_once(self, query: str, max_results: int = 5) -> list[SearchResult]:
import httpx
payload = {
@@ -112,12 +132,13 @@ class IntranetSearchProvider(BaseSearchProvider):
"status": "pending",
"ok": False,
"count": 0,
"verify_ssl": self.verify_ssl,
}
# 排查用:每次真实调用都打一行请求明细(端点 / 知识库 / 检索词 / 超时),
# 内网复现时直接看后端日志即可知道"到底请求了哪个地址、用什么参数"。
logger.info(
"[intranet-search] POST endpoint=%s kb=%s timeout=%ss query=%r",
self.endpoint, self.knowledge_base_name, self.timeout, query,
"[intranet-search] POST endpoint=%s kb=%s timeout=%ss verify_ssl=%s query=%r",
self.endpoint, self.knowledge_base_name, self.timeout, self.verify_ssl, query,
)
try:
# trust_env=False:**关键**。内网 ES 是内网地址,但本部署的模型走公网,
@@ -125,7 +146,8 @@ class IntranetSearchProvider(BaseSearchProvider):
# 会把这个内网请求也套用公网代理 → 代理到不了内网 IP → 连接被重置/空回复,
# 表现为「接口异常」。强制忽略代理 env、直连内网(与 multi_agent.py 的回环
# 调用同款修法)。
async with httpx.AsyncClient(timeout=self.timeout, trust_env=False) as client:
# verify=False:内网 HTTPS 常见自签名/内部 CA 证书;这里仅跳过本内网检索端点的证书校验。
async with httpx.AsyncClient(timeout=self.timeout, trust_env=False, verify=self.verify_ssl) as client:
resp = await client.post(self.endpoint, json=payload)
call["http_status"] = getattr(resp, "status_code", None)
resp.raise_for_status()
@@ -210,7 +232,7 @@ class IntranetSearchProvider(BaseSearchProvider):
"[intranet-search] OK endpoint=%s query=%r http_status=%s raw_items=%s",
self.endpoint, query, call.get("http_status"), len(items) if isinstance(items, list) else 0,
)
results: List[SearchResult] = []
results: list[SearchResult] = []
for i, item in enumerate(items[:max_results]):
if not isinstance(item, dict):
continue
@@ -10,7 +10,9 @@ from types import SimpleNamespace
import pytest
from deerflow.agents.ai_writing.nodes import researcher as R
from deerflow.agents.ai_writing.search import get_search_provider
from deerflow.agents.ai_writing.search.base import SearchResult
from deerflow.agents.ai_writing.search.intranet_stub import IntranetSearchProvider
class _FakeIntranet:
@@ -21,6 +23,7 @@ class _FakeIntranet:
self.raw_responses = []
self.calls = []
self.kw = kw
self.verify_ssl = kw.get("verify_ssl", False)
_FakeIntranet.instances.append(self)
async def search(self, query, max_results=5, **_):
@@ -55,12 +58,24 @@ class _SkillProviderShouldNotRun:
raise AssertionError("knowledge_base 模式不应调用通用 provider / 技能检索")
def _patch(monkeypatch, intranet_url="http://es/api", intranet_cls=_FakeIntranet):
class _FakeResponse:
status_code = 200
text = '[{"title":"内网标题","content":"内网正文","url":"https://intranet/doc"}]'
def raise_for_status(self) -> None:
return None
def json(self):
return [{"title": "内网标题", "content": "内网正文", "url": "https://intranet/doc"}]
def _patch(monkeypatch, intranet_url="http://es/api", intranet_cls=_FakeIntranet, intranet_verify_ssl=False):
_FakeIntranet.instances = []
cfg = {
"max_materials": 20,
"intranet_search_url": intranet_url,
"intranet_knowledge_base": "samples",
"intranet_verify_ssl": intranet_verify_ssl,
}
monkeypatch.setattr(R, "get_ai_writing_config", lambda: cfg)
monkeypatch.setattr(R, "use_builtin_agent", lambda *_: False)
@@ -125,6 +140,15 @@ async def test_knowledge_base_no_url_falls_back_to_default_endpoint(monkeypatch)
assert _FakeIntranet.instances[0].kw.get("base_url") == "", "空地址应原样下传,由 provider 回退默认端点"
@pytest.mark.asyncio
async def test_knowledge_base_passes_verify_ssl_config(monkeypatch):
_patch(monkeypatch, intranet_verify_ssl=True)
out = await R.researcher_node(_state(), {"configurable": {}})
assert _FakeIntranet.instances[0].kw.get("verify_ssl") is True
assert out["material_package"]["intranet_debug"]["verify_ssl"] is True
@pytest.mark.asyncio
async def test_knowledge_base_does_not_orchestrate_skills(monkeypatch):
"""knowledge_base 不编排技能:通用 provider 一旦被调用就会 raise。"""
@@ -163,3 +187,44 @@ async def test_knowledge_base_failure_is_visible(monkeypatch):
steps = pkg["skill_search_steps"]
assert any(step["type"] == "skill_error" for step in steps)
assert any(event["type"] == "notice" and event.get("raw") for event in out["progress_events"])
@pytest.mark.asyncio
async def test_intranet_search_disables_ssl_verification_for_internal_https(monkeypatch):
import httpx
seen: dict = {}
class _FakeAsyncClient:
def __init__(self, **kwargs):
seen.update(kwargs)
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return None
async def post(self, endpoint, json):
seen["endpoint"] = endpoint
seen["payload"] = json
return _FakeResponse()
monkeypatch.setattr(httpx, "AsyncClient", _FakeAsyncClient)
provider = IntranetSearchProvider(base_url="https://intranet.example/search", knowledge_base_name="kb", timeout=300)
results = await provider.search("测试", max_results=1)
assert results
assert seen["verify"] is False
assert seen["trust_env"] is False
assert seen["endpoint"] == "https://intranet.example/search"
assert seen["payload"]["knowledge_base_name"] == "kb"
assert provider.calls[0]["verify_ssl"] is False
def test_intranet_verify_ssl_can_be_reenabled_from_config():
provider = get_search_provider({"search_provider": "intranet", "intranet_verify_ssl": "true"})
assert isinstance(provider, IntranetSearchProvider)
assert provider.verify_ssl is True