From 854de9b43c0891380dc8518259df35090747e0bc Mon Sep 17 00:00:00 2001 From: fengxiangdebaozi <10845145+fengxiangdebaozi@user.noreply.gitee.com> Date: Thu, 25 Jun 2026 17:05:09 +0800 Subject: [PATCH] =?UTF-8?q?fix(ai=E5=86=99=E4=BD=9C):=20=E6=8A=80=E8=83=BD?= =?UTF-8?q?=E6=A3=80=E7=B4=A2=E7=A8=B3=E5=AE=9A=E6=8E=A5=E6=94=B6=E6=95=B0?= =?UTF-8?q?=E6=8D=AE(=E5=A4=8D=E7=94=A8=E5=8F=82=E8=80=83=E6=96=87?= =?UTF-8?q?=E7=8C=AE=E6=8F=90=E5=8F=96+=E6=88=AA=E6=96=AD=E6=8A=A2?= =?UTF-8?q?=E6=95=91+=E4=B8=8A=E9=99=90=E6=97=A9=E5=81=9C)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 复用主智能体「参考文献」模式的提取逻辑(deerflow.runtime.references)解析技能JSON，字段并集兼容 - bash输出中间截断时逐条抢救完整记录对象，不再因大JSON解析失败而全丢 - 所有退出路径(超时/递归上限/异常)都从tool输出收割，不依赖模型最终汇总 - 达到max_materials上限即停止驱动模型，修复步骤条爆100+条/素材超额 - 步骤条实时显示每次检索获得多少条数据 - 内网兜底:与技能并行预跑、响应解析兼容两套接口字段、0结果换简单检索词重试、超时放宽到10分钟 Co-Authored-By: Claude Opus 4.8 (1M context) --- frontend-web/src/core/auth/task-buttons.ts | 26 +- .../pages/RoundtablePlanningPage.tsx | 1 + .../task-components/TaskJumpBar.tsx | 20 +- offline-backend-20260512/backend/CLAUDE.md | 6 +- .../agents/ai_writing/nodes/_skill_agent.py | 293 +++++++++++------- .../agents/ai_writing/nodes/researcher.py | 103 ++++-- .../agents/ai_writing/search/intranet_stub.py | 53 +++- 7 files changed, 340 insertions(+), 162 deletions(-) diff --git a/frontend-web/src/core/auth/task-buttons.ts b/frontend-web/src/core/auth/task-buttons.ts index 07b8d6aa..19857479 100644 --- a/frontend-web/src/core/auth/task-buttons.ts +++ b/frontend-web/src/core/auth/task-buttons.ts @@ -20,7 +20,7 @@ import { readCopTaskDetail, resolveTaskDeeplinkTarget, } from "@/core/auth/task-deeplink"; -import { readUserInfo } from "@/strategy-components/lib/light-app"; +import { readUserInfo, resolveRouteParamValue } from "@/strategy-components/lib/light-app"; /** 「大屏绘制」独立页应用内路由（深链「上一步」默认指向它，跳转追加 ?taskId=）。 */ export const DASHBOARD_AGENT_ROUTE = "/page/strategy/dashboard-agent"; @@ -106,17 +106,20 @@ export interface TaskButton { sortOrder: number; } -/** 运行时解析单个登录参数的值（``other`` 取字面值，否则从 ``userInfo`` 取 ``source`` 字段）。 */ +/** + * 运行时解析单个登录参数的值——**与「应用管理」完全一致**：直接复用 light-app 的 + * ``resolveRouteParamValue``（把本按钮的登录参数适配成 ``login`` 类型的路由参数）， + * 保证取值来源/逻辑跟轻应用一模一样（``other`` 取字面 ``customValue``，否则从 + * ``userInfo`` 取 ``source`` 字段）。 + */ function resolveLoginParamValue( p: TaskButtonLoginParam, userInfo: Record, ): string { - if (p.source === "other") return p.customValue ?? ""; - if (p.source) { - const value = userInfo[p.source]; - return value == null ? "" : String(value); - } - return ""; + return resolveRouteParamValue( + { key: p.key, paramType: "login", source: p.source, customValue: p.customValue }, + userInfo, + ); } /** 内部切换可选的目标业务（``business`` 类型按钮的下拉项）。 */ @@ -311,13 +314,20 @@ export async function runTaskButton(btn: TaskButton, ctx: RunTaskButtonCtx): Pro if (base.startsWith("/") && !base.startsWith("//")) { if (btn.openMode === "blank") { const full = `${window.location.origin}${window.location.pathname}#${url}`; + // 打印实际打开地址，便于排查登录参数/拼接问题。 + console.info(`[task-button] 「${btn.label}」内部路由(新标签) → ${full}`); window.open(full, "_blank", "noopener,noreferrer"); } else { + console.info(`[task-button] 「${btn.label}」内部路由(同窗) → ${url}`); ctx.navigate(url); } return; } // 外链：self=当前窗口（嵌入态即当前 iframe）；否则 blank=新窗口（默认）。 + // 打印最终跳转地址（含追加的 id / 登录参数），便于排查问题。 + console.info( + `[task-button] 「${btn.label}」外链(${btn.openMode === "self" ? "当前窗口" : "新窗口"}) → ${url}`, + ); if (btn.openMode === "self") { window.location.assign(url); } else { diff --git a/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx b/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx index 3404647c..a4e1c536 100644 --- a/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx +++ b/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx @@ -2265,6 +2265,7 @@ export default function RoundtablePlanningPage() { )} diff --git a/frontend-web/src/strategy-components/components/task-components/TaskJumpBar.tsx b/frontend-web/src/strategy-components/components/task-components/TaskJumpBar.tsx index 0003b4f1..58e2f359 100644 --- a/frontend-web/src/strategy-components/components/task-components/TaskJumpBar.tsx +++ b/frontend-web/src/strategy-components/components/task-components/TaskJumpBar.tsx @@ -44,10 +44,19 @@ import { TaskPurposeDialog } from "@/strategy-components/components/task-compone export function TaskJumpBar({ business, taskId, + actionId, roundtable, }: { business: TaskBusiness; taskId: string; + /** + * xdfx 深链原始「行动id」（URL ?taskId= 传入值，与解析后的任务id 不同）。 + * 单智能体问答（``TaskFxWorkspace``）由 ``persistTaskContext`` 写进 ``TaskContext``， + * 故省略时回退 ``readTaskContext()?.actionId``；多智能体会商（``RoundtablePlanningPage``） + * 不写 ``TaskContext``（只在 ``roundtablePrefill`` 里），必须**显式**把 ``prefill.actionId`` + * 传进来，否则「追加行动id」的按钮会拿不到行动id 而回退任务id。 + */ + actionId?: string; /** 多智能体会商深链模式：不展示「编辑」与「弹框查看」（purpose）按钮。 */ roundtable?: boolean; }) { @@ -73,10 +82,13 @@ export function TaskJumpBar({ // 目的树弹框（purpose 类型按钮）由组件层打开。 const [purposeOpen, setPurposeOpen] = useState(false); - // xdfx：深链原始「行动id」（与解析后的任务id 不同才算真行动id），弹框据此换回任务id 再查目的。 - const ctxActionId = readTaskContext()?.actionId; + // xdfx 深链原始「行动id」：会商模式由 prop 显式传入；单智能体回退 TaskContext。 + const effectiveActionId = actionId ?? readTaskContext()?.actionId; + // 与解析后的任务id 不同才算真行动id，弹框据此换回任务id 再查目的。 const purposeActionId = - business === "xdfx" && ctxActionId && ctxActionId !== taskId ? ctxActionId : undefined; + business === "xdfx" && effectiveActionId && effectiveActionId !== taskId + ? effectiveActionId + : undefined; if (jumpButtons.length === 0) return null; // 深链 `hideJump=1`（外部 iframe 套用时）：整条跳转栏隐藏——底部「下一步」等按钮 + @@ -90,7 +102,7 @@ export function TaskJumpBar({ } void runTaskButton(btn, { taskId, - actionId: readTaskContext()?.actionId, + actionId: effectiveActionId, navigate, onError: (m) => toast.error(m), }); diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md index 73c1e57b..dbdda109 100644 --- a/offline-backend-20260512/backend/CLAUDE.md +++ b/offline-backend-20260512/backend/CLAUDE.md @@ -274,9 +274,11 @@ Seeding mirrors the roundtable pattern: verified `config.yaml` + `SOUL.md` ship **Skill-driven retrieval (检索类型).** `material_source` has exactly two values: `general` (通用检索, default) and `notebook` (我的空间); legacy `knowledge_base`/`skill` are normalized to `general` by the researcher node. 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 **final 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 "检不出". +**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 "检不出". -**Partial-result harvest + query simplification (检不出/超时的容错).** When `researcher_skill_agent: true` runs a non-native skill via `run_skill_via_real_agent` (the deployment's real path — the skill, e.g. `knowledge-search-v2`, is invoked by the real agent through `bash python3 scripts/search.py ""`), one writing run issues many queries and **some time out** (the skill script prints `{"success":false,"error":"请求超时…","count":0}`) while others succeed (`{"success":true,"count":30,"results":[…]}`). Two fixes so successful data is never dropped: (1) `_drive()` captures every **ToolMessage** stdout and `_harvest_tool_results()` parses the skill's JSON directly (`_try_load_json` tolerates a trailing `Exit Code: 1`; skips `success:false`; field-name union across `knowledge-search-v2`/intranet-ES shapes) — so even if the model's final `{materials:[]}` summary gives up (confused by the timeouts), the successful queries' real results are still collected; the function merges model materials + harvested results (dedup by url/recUuid/title, capped at `max_results`). (2) The model is instructed (in `_MATERIAL_OUTPUT_CONTRACT`) that a 0-result/timeout query must be **retried with a simpler, core query** (drop topic suffixes like 竞选/诉讼/政策, keep the core entity: 「特朗普竞选」→「特朗普」) and that any data already retrieved must always be kept. The same deterministic simplifier `search/base.py::simplify_query` is wired into `IntranetSearchProvider.search`: an empty first attempt retries once with the simplified query. Tests: `tests/test_ai_writing_skill_real_agent.py`, `tests/test_ai_writing_query_simplify.py`. +**Parallel intranet fallback (技能 0 条 → 直接用并行预跑的内网结果).** When `ai_writing.intranet_search_url` is configured, the researcher node launches the intranet ES search (`IntranetSearchProvider` over the generated keywords) as a **concurrent `asyncio.Task` the moment the skill loop starts** (not sequentially after all skills fail). If the configured + fallback skills collect **nothing**, the node awaits that already-running task and uses its results (`_collect`) — no extra wait. If the skills **did** get materials, the task is cancelled (`_cancel_intranet_task`, suppresses CancelledError) so the intranet call isn't wasted. URL empty → no task, behavior unchanged. Tests: `tests/test_ai_writing_intranet_parallel.py`. + +**Partial-result harvest + query simplification (检不出/超时的容错).** When `researcher_skill_agent: true` runs a non-native skill via `run_skill_via_real_agent` (the deployment's real path — the skill, e.g. `knowledge-search-v2`, is invoked by the real agent through `bash python3 scripts/search.py ""`), one writing run issues many queries and **some time out** (the skill script prints `{"success":false,"error":"请求超时…","count":0}`) while others succeed (`{"success":true,"count":30,"results":[…]}`). Two fixes so successful data is never dropped: (1) `_drive()` captures every **ToolMessage** stdout and `_harvest_tool_results()` parses the skill's JSON **by reusing the lead-agent 参考文献 mode's proven extractor** (`deerflow.runtime.references`: `_parse_json_like` + `_first_list` + `_normalize_reference_item`) — the same logic that already extracts skill JSON into citations 100% reliably on ChatPage/AgentChatPage, with its huge field-name union (title/m_title/name, content/page_content/content_preview/snippet/…, recUuid/uuid/docId/…) and per-skill display-config support — plus skips `success:false` outputs — so even if the model's final `{materials:[]}` summary gives up (confused by the timeouts), the successful queries' real results are still collected; the function merges model materials + harvested results (dedup by url/recUuid/title, capped at `max_results`). **The harvest+merge (`_finalize`) runs on EVERY exit path — normal end, overall timeout, AND `await task` exceptions (e.g. `GraphRecursionError`)** — so when the run is cut off before the model writes its summary (observed: 10 queries × ~2 graph steps > the old `recursion_limit`, so the agent errored out with the real 30+ results already in `tool_outputs`), those results are still salvaged instead of returning `[]`. `recursion_limit` was also raised from `_REAL_AGENT_MAX_TURNS` to `_REAL_AGENT_MAX_TURNS*2+4` (it counts graph super-steps ≈ 2 per model→tool round, not model turns) so the model normally has room to finish its summary. **Early-stop on `max_results`:** `_drive` re-harvests after every tool output and, once the deduped harvested count reaches `max_results` (= `per_skill` = `max_materials`), **breaks the nested `astream`** and emits a 「已获取 N 条素材，达到上限，停止本技能检索」 step — without it a weak model keeps issuing new queries forever (observed: 100+ step-bar entries, materials far over the limit). Because the skill now returns exactly `max_materials`, the researcher loop's own `collect_limit` early-stop fires right after, so later configured skills don't run either. (2) The model is instructed (in `_MATERIAL_OUTPUT_CONTRACT`) that a 0-result/timeout query must be **retried with a simpler, core query** (drop topic suffixes like 竞选/诉讼/政策, keep the core entity: 「特朗普竞选」→「特朗普」) and that any data already retrieved must always be kept. The same deterministic simplifier `search/base.py::simplify_query` is wired into `IntranetSearchProvider.search`: an empty first attempt retries once with the simplified query. **(3) Root cause of the instability — bash output middle-truncation.** The skill prints its results as JSON to stdout, and the `bash` tool **middle-truncates** any output over `sandbox.bash_output_max_chars` (default **20000**, inserting a `... [middle truncated …] ...` marker). A `count:30` skill response (~30 records × ~600 chars) exceeds 20000 → the JSON is cut in the middle → **unparseable as a whole** by both the model and `_try_load_json` → all materials lost (while a small `count:9` response parses fine — hence the flapping). Fix: when whole-JSON parse yields no items, `_harvest_tool_results` falls back to `_scan_balanced_json_objects` — a brace-depth/string-aware scanner that recovers **every individually-complete `{…}` record object** from the truncated text (the records on each side of the truncation marker are still valid JSON), so ~16+16 records survive a 20000-char cut, far above `max_materials`. This is why the lead-agent chat pages (`/page/workspace/chats/new`, agent chat) call the same skill with 100% success — they only need the model to summarize prose from whatever it sees, never to parse a structured list, so truncation doesn't hurt them; ai-writing needs the structured records, so it must salvage them itself. Tests: `tests/test_ai_writing_skill_real_agent.py`, `tests/test_ai_writing_query_simplify.py`. **Agentic skill execution (`ai_writing.researcher_skill_agent`, default false).** By default a configured non-native skill is *not* run agent-style — the node just turns its `description` into a query for the generic search provider. With this flag on, the directed-skill branch instead runs `nodes/_skill_agent.py::run_skill_search_agent`: it injects the skill's **SKILL.md** (read from `skill.skill_file`) alongside the researcher SOUL as the system prompt, binds a `search` tool to the researcher model, and runs a bounded (`_MAX_TURNS=4`) tool-call loop so the model drives the retrieval per the skill's instructions; the `SearchResult`s the tool returns are harvested directly as materials (no fuzzy text extraction). Requires a tool-calling model; if the model can't tool-call (returns no `tool_calls`) or the agent yields nothing, `_execute_skill` falls back to the legacy description→query search so materials are never empty. The built-in `knowledge-base-search` native concurrent path is unchanged. Tests: `tests/test_ai_writing_skill_agent.py`. diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_skill_agent.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_skill_agent.py index 3ceda582..cfc5f316 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_skill_agent.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_skill_agent.py @@ -70,108 +70,152 @@ _MATERIAL_OUTPUT_CONTRACT = ( ) -# 技能 stdout 里「结果列表」常见的字段名（不同技能各异，做并集兼容） -_RESULT_LIST_KEYS = ("results", "data", "items", "materials", "documents", "list", "hits") -_TITLE_KEYS = ("title", "m_title", "name", "doc_title") -_CONTENT_KEYS = ("content", "content_preview", "page_content", "text", "summary", "abstract") -_SOURCE_KEYS = ("source", "source_category", "source1", "source_name", "kb_name") -_DATE_KEYS = ("publish_time", "published_at", "m_publish", "date", "create_time") +def _scan_balanced_json_objects(text: str) -> list: + """从可能被**中间截断**的文本里扫出所有「括号平衡且完整」的 ``{...}`` 子串（含各嵌套层级）。 - -def _try_load_json(text: str): - """尽量从技能 stdout 解析出 JSON：先整体 loads，失败再截取首个平衡的 {…}/[…]。 - - 技能脚本的输出常带尾巴（如 ``...}\n\nExit Code: 1``），整体 loads 会失败，故再用 - 「首个左括号 → 最后一个对应右括号」截取主体重试。 + 根因：bash 工具对超过 ~20000 字的输出做**中间截断**（插入「... [middle truncated …] ...」）， + 使整段大 JSON（如 count=30 的技能返回）无法解析；但被截断点**两侧的单条结果对象**通常仍是 + 完整合法的 JSON，可逐个抢救出来。带字符串状态机（正确处理引号内的 {}/转义），逐个返回闭合的 + 对象子串，调用方再逐个 json.loads + 字段识别。 """ import json as _json - text = (text or "").strip() - if not text: - return None - try: - return _json.loads(text) - except Exception: # noqa: BLE001 - pass - for open_c, close_c in (("{", "}"), ("[", "]")): - start = text.find(open_c) - end = text.rfind(close_c) - if 0 <= start < end: - try: - return _json.loads(text[start : end + 1]) - except Exception: # noqa: BLE001 - continue - return None + out: list = [] + stack: list[int] = [] + in_str = False + esc = False + for i, ch in enumerate(text): + if in_str: + if esc: + esc = False + elif ch == "\\": + esc = True + elif ch == '"': + in_str = False + continue + if ch == '"': + in_str = True + elif ch == "{": + stack.append(i) + elif ch == "}": + if stack: + s = stack.pop() + frag = text[s : i + 1] + try: + parsed = _json.loads(frag) + except Exception: # noqa: BLE001 + continue + if isinstance(parsed, dict): + out.append(parsed) + return out -def _extract_result_items(obj) -> list: - """从解析出的 JSON 里取「结果数组」：顶层就是数组则直接用，否则按常见字段名找。""" - if isinstance(obj, list): - return obj - if isinstance(obj, dict): - for key in _RESULT_LIST_KEYS: - v = obj.get(key) - if isinstance(v, list): - return v - return [] +def _describe_tool_output(raw: str, skill_name: str) -> str: + """从单次技能输出里提炼一句步骤条文案：**检索词 + 本次命中条数**（用户要求实时显示每次拿到多少条）。 + 返回空串 = 这次工具输出不是可识别的检索结果（如 read_file 读 SKILL.md），不发步骤。 + """ + import re as _re -def _pick(d: dict, keys: tuple) -> str: - for k in keys: - v = d.get(k) - if isinstance(v, str) and v.strip(): - return v.strip() - if v not in (None, "") and not isinstance(v, (dict, list)): - return str(v).strip() - return "" + from deerflow.runtime import references as _ref + + if not isinstance(raw, str) or not raw.strip(): + return "" + payload = _ref._parse_json_like(raw) + query = "" + if isinstance(payload, dict): + query = str(payload.get("query") or "").strip() + # 明确失败（超时/报错）：也给一条「0 条 + 原因」，让用户看到这次没出数据 + if payload.get("success") is False: + err = str(payload.get("error") or "").strip() + q = f"「{query}」" if query else "" + return f"技能「{skill_name}」检索{q}未获得数据（0 条）{('：' + err) if err else ''}" + + count: int | None = None + if isinstance(payload, dict) and isinstance(payload.get("count"), int): + count = payload["count"] + items = _ref._first_list(payload) if payload is not None else [] + if count is None and items: + count = len(items) + if count is None: + # 整段没解析出来（最常见：bash 输出被中间截断）→ 从头部正则抢救 query/count， + # 再退化为「抢救出的完整记录对象」计数。 + if not query: + m = _re.search(r'"query"\s*:\s*"([^"]{1,80})"', raw) + query = m.group(1).strip() if m else "" + m2 = _re.search(r'"count"\s*:\s*(\d+)', raw) + if m2: + count = int(m2.group(1)) + else: + salvaged = _scan_balanced_json_objects(raw) + n = sum(1 for o in salvaged if o.get("content") or o.get("content_preview") or o.get("title") or o.get("m_title")) + count = n if n > 0 else None + if count is None: + return "" # 非检索类输出，不发步骤 + q = f"「{query}」" if query else "" + return f"技能「{skill_name}」检索{q}获得 {count} 条数据" def _harvest_tool_results(tool_outputs: list, skill_name: str, limit: int) -> list: """**直接从技能 stdout 收割真实检索结果**（不依赖模型最后的 JSON 汇总）。 - 关键容错：一次写作会对同一技能跑多个检索词，部分检索词可能超时（``success:false``）， - 但成功的那些（``success:true``）拿回的真实数据**必须照单全收**，不能因为别的检索词没出 - 数据就整段丢弃。本函数逐条解析技能输出、跳过 ``success:false`` 的、把结果数组映射成 - SearchResult，按 url/recUuid/title 去重，截断到 limit。字段名做并集兼容（适配 knowledge- - search-v2 的 ``{success,query,count,results:[{recUuid,title,url,content_preview,...}]}`` - 及内网 ES 的 ``{m_title,content,recUuid,...}`` 等多种技能输出）。 + **复用主智能体「参考文献」模式同一套久经考验的提取逻辑**（`deerflow.runtime.references`）： + 那套逻辑在 ChatPage/AgentChatPage 已 100% 稳定地把技能 JSON 解析成参考文献——它对结果列表 + 字段（results/data/items/documents/...）、单条字段（title/m_title、content/page_content/ + content_preview/snippet/...、recUuid/uuid/docId/...）都做了超大并集兼容，并尊重每个技能的 + 展示配置。ai 写作直接复用它，确保和参考文献模式「解析能力完全一致」。 + + 在此基础上再叠加两层本地容错： + - 跳过 ``success:false`` 的输出（某检索词超时不影响其它成功的）； + - 整段 JSON 解析不出结果时（最常见：bash 输出被**中间截断**），退化为「逐个抢救完整的 + 记录对象」（`_scan_balanced_json_objects`），截断点两侧的单条记录仍能救回。 """ from ..search.base import SearchResult + from deerflow.runtime import references as _ref + tool_context = {"skill_name": skill_name, "tool_name": skill_name} harvested: list = [] seen: set[str] = set() for raw in tool_outputs: - obj = _try_load_json(raw) - if obj is None: + if not isinstance(raw, str): continue + payload = _ref._parse_json_like(raw) # 明确标记失败的（超时 / 报错）跳过，但不影响其它成功的输出 - if isinstance(obj, dict) and obj.get("success") is False: + if isinstance(payload, dict) and payload.get("success") is False: continue - for it in _extract_result_items(obj): + items = _ref._first_list(payload) if payload is not None else [] + if not items: + # 整段解析不出结果（bash 中间截断把大 JSON 剪断）→ 逐个抢救完整的记录对象 + items = _scan_balanced_json_objects(raw) + for idx, it in enumerate(items): if not isinstance(it, dict): continue - # 内网 ES 风格：真实字段藏在 metadata 里 - md = it.get("metadata") if isinstance(it.get("metadata"), dict) else {} - merged = {**md, **it} if md else it - content = _pick(merged, _CONTENT_KEYS) - title = _pick(merged, _TITLE_KEYS) + # 用参考文献模式的归一化（含字段并集 + 每技能展示配置），与右侧参考文献完全一致 + norm = _ref._normalize_reference_item( + it, len(harvested) + 1, payload=payload, tool_context=tool_context + ) + if not norm: + continue + content = str(norm.get("content") or norm.get("snippet") or "").strip() + title = str(norm.get("title") or "").strip() if not content and not title: continue - url = _pick(merged, ("url",)) - rec = _pick(merged, ("recUuid", "rec_uuid", "id")) + raw_rec = norm.get("raw") if isinstance(norm.get("raw"), dict) else {} + rec = str(raw_rec.get("recUuid") or raw_rec.get("uuid") or raw_rec.get("docId") or "").strip() + url = str(norm.get("url") or "").strip() key = url or rec or title if key in seen: continue seen.add(key) try: - score = float(merged.get("relevance_score") or 0.0) + score = float(norm.get("score")) except (TypeError, ValueError): - score = 0.0 + score = 0.75 harvested.append(SearchResult( title=title or "（无标题）", content=content[:500], - source=_pick(merged, _SOURCE_KEYS) or skill_name, - published_at=_pick(merged, _DATE_KEYS), + source=str(norm.get("source") or skill_name).strip() or skill_name, + published_at=str(norm.get("time") or "").strip(), url=url, relevance_score=score or 0.75, source_type="skill", @@ -304,7 +348,10 @@ async def run_skill_via_real_agent( logger.warning("skill real-agent[%s]: init failed: %s", skill_name, e) return [], f"运行时初始化失败: {e}" - run_config: dict = {"recursion_limit": _REAL_AGENT_MAX_TURNS} + # recursion_limit 计的是 graph 超级步（每个「模型→工具」回合≈2 步），不能直接等于 max_turns， + # 否则模型做了 N 个检索词后还没来得及写最终汇总就被 GraphRecursionError 截断（实测 10 个 + # 检索词≈20 步 > 16，结果汇总丢失）。给到 ~2×turns + 余量，让模型正常情况下能写完汇总。 + run_config: dict = {"recursion_limit": _REAL_AGENT_MAX_TURNS * 2 + 4} run_context: dict = {} if thread_id: run_config["configurable"] = {"thread_id": thread_id} @@ -330,6 +377,7 @@ async def run_skill_via_real_agent( async def _drive() -> None: nonlocal final_text + reached_limit = False try: async for chunk in agent.astream( init_state, config=run_config, context=run_context, stream_mode="values" @@ -352,6 +400,18 @@ async def run_skill_via_real_agent( out = _content_to_text(m.content) if out.strip(): tool_outputs.append(out) + # 实时步骤条：这次检索拿到多少条数据（含失败/截断的诚实计数） + desc = _describe_tool_output(out, skill_name) + if desc: + step_queue.put_nowait(desc) + # **早停**：已收割素材达上限就停止驱动模型——否则弱模型会无限换检索词 + # （步骤条爆 100+ 条、素材远超上限）。停掉嵌套 astream 即不再检索。 + if max_results and len(_harvest_tool_results(tool_outputs, skill_name, max_results)) >= max_results: + step_queue.put_nowait(f"已获取 {max_results} 条素材，达到上限，停止本技能检索") + reached_limit = True + break + if reached_limit: + break finally: step_queue.put_nowait(None) # sentinel：跑完/出错都让 pump 收尾 @@ -365,10 +425,62 @@ async def run_skill_via_real_agent( if emit_query: emit_query(item) # 本协程上下文调用 writer → custom 事件实时刷给前端 + def _finalize() -> list: + """把「模型最终汇总的 materials」与「技能 stdout 直接收割的结果」合并去重。 + + **关键**：无论本次运行是正常结束、超时、还是抛异常，只要已经收到过技能的工具输出， + 就必须把这些真实结果收割下来——绝不能因为模型还没来得及汇总、或整段被切断，就把已经 + 拿到的数据整段丢掉（用户实测：技能明明检到了 30+ 条，却因没汇总而全丢）。 + """ + # 1) 模型最终汇总的 materials（它可能已按相关度筛选/去重） + model_results: list = [] + parsed = parse_json(final_text or "", {"materials": []}) + raw_materials = parsed.get("materials") + if isinstance(raw_materials, list): + for m in raw_materials: + if not isinstance(m, dict): + continue + content = str(m.get("content") or "").strip() + title = str(m.get("title") or "").strip() + if not content and not title: + continue + model_results.append(SearchResult( + title=title or "（无标题）", + content=content[:500], + source=str(m.get("source") or skill_name).strip() or skill_name, + published_at=str(m.get("published_at") or "").strip(), + url=str(m.get("url") or "").strip(), + relevance_score=0.8, + source_type="skill", + )) + + # 2) **直接从技能 stdout 收割真实检索结果**——哪怕模型最后没汇总、或别的检索词超时， + # 成功检索词拿回的数据也照单全收。 + harvested = _harvest_tool_results(tool_outputs, skill_name, max_results) + + # 3) 合并去重（模型汇总优先、技能原始结果补齐），按 url/recUuid/title 去重，截断到上限。 + merged: list = [] + seen_keys: set[str] = set() + for r in model_results + harvested: + key = (r.url or "").strip() or (r.recUuid or "").strip() or (r.title or "").strip() + if key and key in seen_keys: + continue + if key: + seen_keys.add(key) + merged.append(r) + if len(merged) >= max_results: + break + return merged + try: await asyncio.wait_for(_pump(), timeout=max(timeout_s, _REAL_AGENT_TIMEOUT_SECONDS)) except asyncio.TimeoutError: task.cancel() + # 超时也要收割已拿到的技能数据：有就返回（无错误），实在没有才报超时。 + salvaged = _finalize() + if salvaged: + logger.info("skill real-agent[%s]: timed out but salvaged %d results", skill_name, len(salvaged)) + return salvaged, "" return [], f"技能检索超时（>{max(timeout_s, _REAL_AGENT_TIMEOUT_SECONDS)}s）" # pump 正常结束（收到 sentinel）后等后台任务收尾，拿 final_text / 抛出的异常 @@ -376,47 +488,14 @@ async def run_skill_via_real_agent( await task except Exception as e: # noqa: BLE001 logger.warning("skill real-agent[%s]: astream failed: %s", skill_name, e, exc_info=True) + # 递归上限 / 中途异常同样要收割已拿到的技能数据，不整段丢。 + salvaged = _finalize() + if salvaged: + logger.info("skill real-agent[%s]: errored but salvaged %d results", skill_name, len(salvaged)) + return salvaged, "" return [], f"技能检索失败: {e}" - # 1) 模型最终汇总的 materials（它可能已按相关度筛选/去重） - model_results: list = [] - parsed = parse_json(final_text or "", {"materials": []}) - raw_materials = parsed.get("materials") - if isinstance(raw_materials, list): - for m in raw_materials: - if not isinstance(m, dict): - continue - content = str(m.get("content") or "").strip() - title = str(m.get("title") or "").strip() - if not content and not title: - continue - model_results.append(SearchResult( - title=title or "（无标题）", - content=content[:500], - source=str(m.get("source") or skill_name).strip() or skill_name, - published_at=str(m.get("published_at") or "").strip(), - url=str(m.get("url") or "").strip(), - relevance_score=0.8, - source_type="skill", - )) - - # 2) **直接从技能 stdout 收割真实检索结果**——哪怕模型最后没汇总、或别的检索词超时， - # 成功检索词拿回的数据也照单全收（用户诉求：拿到的数据必须接收，不能整段丢）。 - harvested = _harvest_tool_results(tool_outputs, skill_name, max_results) - - # 3) 合并去重（模型汇总优先、技能原始结果补齐），按 url/recUuid/title 去重，截断到上限。 - merged: list = [] - seen_keys: set[str] = set() - for r in model_results + harvested: - key = (r.url or "").strip() or (r.recUuid or "").strip() or (r.title or "").strip() - if key and key in seen_keys: - continue - if key: - seen_keys.add(key) - merged.append(r) - if len(merged) >= max_results: - break - return merged, "" + return _finalize(), "" def _read_skill_body(skill) -> str: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py index 355e4010..c5790883 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py @@ -767,6 +767,60 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict # 早停上限：通常为 max_materials；用户显式要「再追加 N 条」时放宽到 N，确保凑够 collect_limit = max(max_materials, requested_append_count or 0) + # ── 内网检索接口：与技能检索**并行**预跑 ── + # 用户诉求：调用技能的同时就把内网检索接口跑起来；技能没拿到结果时，直接用这里 + # 预跑好的内网结果兜底（省去"技能全失败后再串行等一遍内网"的时间）。需在 config.yaml + # 的 ai_writing.intranet_search_url 配地址才启用；技能拿到结果则取消这个任务、不浪费。 + intranet_url = (ai_writing_config.get("intranet_search_url") or "").strip() + intranet_task: "asyncio.Task | None" = None + intranet_holder: dict = {} + + async def _run_intranet_search() -> list[SearchResult]: + # 全程吞异常：并行内网任务**绝不**能把主流程带挂（实测内网接口可能瞬间失败）。 + # 失败只记 search_errors + 日志，返回空列表，交由技能结果或空结果文案兜底。 + try: + from ..search.intranet_stub import IntranetSearchProvider + + intranet = IntranetSearchProvider( + base_url=intranet_url, + knowledge_base_name=ai_writing_config.get("intranet_knowledge_base", "samples"), + # 兜底接口超时可在 config.yaml 的 ai_writing.intranet_search_timeout 调大， + # 默认 600s（10 分钟），给内网慢检索留足时间。 + timeout=ai_writing_config.get("intranet_search_timeout", 600), + ) + intranet_holder["provider"] = intranet + fb_keywords = keywords[:keyword_count] or [intent] + gathered = await asyncio.gather( + *[intranet.search(kw, max_results=per_skill) for kw in fb_keywords], + return_exceptions=True, + ) + collected: list[SearchResult] = [] + for res in gathered: + if isinstance(res, Exception): + logger.warning("[intranet-parallel] one keyword failed: %s", res) + search_errors.append(f"{type(res).__name__}: {res}") + continue + collected.extend(res or []) + return collected + except asyncio.CancelledError: + raise # 取消要正常传播（技能已拿到结果时主动取消） + except Exception as e: # noqa: BLE001 + logger.warning("[intranet-parallel] intranet fallback crashed: %s", e, exc_info=True) + search_errors.append(f"内网兜底异常：{type(e).__name__}: {e}") + return [] + + if intranet_url: + intranet_task = asyncio.create_task(_run_intranet_search()) + + async def _cancel_intranet_task() -> None: + """技能已拿到素材：取消并行的内网检索任务，避免无谓占用 / 未取回异常告警。""" + if intranet_task is not None and not intranet_task.done(): + intranet_task.cancel() + try: + await intranet_task + except BaseException: # noqa: BLE001 — CancelledError / 任务内异常都吞掉 + pass + for sk in ordered: added, error_detail = await _execute_skill(sk) tried_names.add(sk.name) @@ -802,45 +856,34 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict # 兜底已拿到素材即止，不再无谓尝试后续技能 break - # ── 最终兜底：技能（含兜底技能）都没检索到 → 直连内网知识库接口（ES）兜一把 ── - # 用户诉求：技能因网络/配置连不上而 0 条时，至少用内网 ES 接口直检，避免素材全空。 - # 需在 config.yaml 的 ai_writing 段配 intranet_search_url 才启用；接口契约见 - # IntranetSearchProvider（POST {query,knowledge_base_name,ts} → [{page_content,metadata}]）。 + # ── 最终兜底：技能（含兜底技能）都没检索到 → 用并行预跑的内网知识库接口结果兜底 ── + # 用户诉求：调用技能时已同时跑了内网检索接口；技能因网络/配置连不上而 0 条时，直接用 + # 那份并行结果，避免素材全空、也省去串行等待。需在 config.yaml 配 intranet_search_url。 if not new_results: - intranet_url = (ai_writing_config.get("intranet_search_url") or "").strip() - if intranet_url: - from ..search.intranet_stub import IntranetSearchProvider - + if intranet_task is not None: _emit_skill_step({ "type": "fallback", - "message": "技能均未检索到素材，改为直连内网知识库接口兜底检索…", + "message": "技能均未检索到素材，使用并行预跑的内网知识库接口结果兜底…", }) - intranet = IntranetSearchProvider( - base_url=intranet_url, - knowledge_base_name=ai_writing_config.get("intranet_knowledge_base", "samples"), - # 兜底接口超时可在 config.yaml 的 ai_writing.intranet_search_timeout 调大， - # 默认 600s（10 分钟），给内网慢检索留足时间。 - timeout=ai_writing_config.get("intranet_search_timeout", 600), - ) - fb_keywords = keywords[:keyword_count] or [intent] - fb_gathered = await asyncio.gather( - *[intranet.search(kw, max_results=per_skill) for kw in fb_keywords], - return_exceptions=True, - ) - fb_added = 0 - for fb_results in fb_gathered: - if isinstance(fb_results, Exception): - search_errors.append(f"{type(fb_results).__name__}: {fb_results}") - continue - fb_added += _collect(fb_results) - search_errors.extend(intranet.errors) - # 让下方 notice 拿到内网接口的原始响应体（解析失败时便于定位） - provider = intranet + try: + fb_results = await intranet_task + except BaseException as e: # noqa: BLE001 — 任务内异常/取消都按"无结果"处理 + fb_results = [] + search_errors.append(f"{type(e).__name__}: {e}") + fb_added = _collect(fb_results) + intranet = intranet_holder.get("provider") + if intranet is not None: + search_errors.extend(getattr(intranet, "errors", [])) + # 让下方 notice 拿到内网接口的原始响应体（解析失败时便于定位） + provider = intranet _emit_skill_step({ "type": "skill_result", "message": f"内网知识库接口兜底检索到 {fb_added} 条新素材", "matched_count": fb_added, }) + else: + # 技能已拿到素材：取消并行的内网检索，不再浪费内网接口调用 + await _cancel_intranet_task() # 合并去重 # - 首次检索 / 普通 re_search（无输入，用户只想重出）：旧素材丢弃，覆盖式替换 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py index a0519518..12b58ff5 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py @@ -15,6 +15,17 @@ _DEFAULT_KB = "samples" _RAW_SNIPPET_LIMIT = 2000 +def _pick(d: dict, keys: tuple) -> str: + """从结果项里按候选字段名取第一个非空值（两套接口字段名不同，做并集兼容）。""" + for k in keys: + v = d.get(k) + if isinstance(v, str) and v.strip(): + return v.strip() + if v not in (None, "") and not isinstance(v, (dict, list)): + return str(v).strip() + return "" + + class IntranetSearchProvider(BaseSearchProvider): """内网知识库搜索接口适配器。 @@ -92,11 +103,17 @@ class IntranetSearchProvider(BaseSearchProvider): self._record_raw(raw_text) return [] - # 正常情况下接口直接返回数组；兼容个别情况下包了一层 data/results 的结构 + # 正常情况下接口直接返回数组；兼容包了一层 data/results 的结构（含工作正常的 + # knowledge-search-v2：{"success":true,"results":[...]}）。 if isinstance(data, list): items = data elif isinstance(data, dict): - items = data.get("data") or data.get("results") or [] + # 接口显式标失败时把整体回传前端便于定位 + if data.get("success") is False: + self.errors.append(f"接口返回 success=false：{str(data.get('error') or '')[:200]}") + self._record_raw(raw_text) + return [] + items = data.get("data") or data.get("results") or data.get("documents") or [] else: # 返回结构既不是数组也不是已知 dict 包装：视为异常，把原始响应回传前端 logger.warning("intranet search unexpected response type for query=%r: %s", query, type(data).__name__) @@ -106,18 +123,32 @@ class IntranetSearchProvider(BaseSearchProvider): results: List[SearchResult] = [] for i, item in enumerate(items[:max_results]): - md = (item or {}).get("metadata") or {} - content = md.get("content") or (item or {}).get("page_content") or "" + if not isinstance(item, dict): + continue + # 字段名做并集兼容两套接口： + # 老 ES： {page_content, metadata:{m_title, content, recUuid, m_publish, source1}} + # knowledge-search-v2：{title, content_preview, url, recUuid, source_category, publish_time, relevance_score} + md = item.get("metadata") if isinstance(item.get("metadata"), dict) else {} + merged = {**md, **item} if md else item + content = _pick(merged, ("content", "content_preview", "page_content", "text", "summary")) + title = _pick(merged, ("title", "m_title", "name", "doc_title")) + if not content and not title: + continue + try: + score = float(merged.get("relevance_score")) + except (TypeError, ValueError): + # 接口未返回相关度分值：按返回顺序赋递减分以保留排序 + score = max(0.5, 0.95 - i * 0.03) results.append(SearchResult( - title=str(md.get("m_title") or "").strip() or "（无标题）", + title=title or "（无标题）", content=str(content)[:500], - source=str(md.get("source1") or "").strip() or "内网知识库", - published_at=str(md.get("m_publish") or "").strip(), - url="", # 内网素材的原文地址由前端依据 recUuid 拼接 - # 接口未返回相关度分值，按 ES 返回顺序赋递减分以保留排序 - relevance_score=max(0.5, 0.95 - i * 0.03), + source=_pick(merged, ("source_category", "source1", "source", "source_name", "kb_name")) or "内网知识库", + published_at=_pick(merged, ("publish_time", "m_publish", "published_at", "date")), + # 工作接口直接给了 url；老 ES 没有 url（由前端依据 recUuid 拼接），保持空串 + url=_pick(merged, ("url",)), + relevance_score=score, source_type="intranet", - recUuid=str(md.get("recUuid") or "").strip(), + recUuid=_pick(merged, ("recUuid", "rec_uuid", "id")), )) return results