From 6d326a21cd92ea0275dfd44a7b5f2ffeb63e472c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Wed, 2 Sep 2026 12:52:37 +0800 Subject: [PATCH] =?UTF-8?q?feat(backend):=20=E5=AE=8C=E5=96=84=20Workflow?= =?UTF-8?q?=20=E4=B8=8E=20DeepResearch=20=E7=9A=84=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E9=93=BE=E8=B7=AF=E4=B8=8E=E6=8C=81=E4=B9=85=E5=8C=96=E8=83=BD?= =?UTF-8?q?=E5=8A=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 功能说明：\n- 重构并增强工作流网关执行链路，支持更完整的 proposal 与 run 调度参数透传\n- 增加 DeepResearch/workflow 运行事件持久化能力并补齐建表 SQL 与查询/保存接口适配\n- 优化 lead agent 与 deep research runner 的中间态处理，增强线程收集与执行链并发控制\n- 调整工作流执行器与任务路由，提升兼容性并完善错误与回放行为\n- 补充回归测试：deep_research 持久化、workflow 并发、定义加载、引擎执行、run 流程等覆盖 --- offline-backend-20260512/backend/CLAUDE.md | 9 +- offline-backend-20260512/backend/Makefile | 2 +- offline-backend-20260512/backend/README.md | 20 ++ .../backend/app/gateway/app.py | 25 +- .../app/gateway/deep_research_job_executor.py | 49 ++-- .../app/gateway/routers/workflow_planning.py | 17 ++ .../app/gateway/routers/workflow_runs.py | 24 +- .../gateway/routers/workflows_coze_compat.py | 6 +- .../backend/app/gateway/services.py | 3 + .../app/gateway/workflow_agent_runner.py | 156 ++++++++++++- .../backend/app/gateway/workflow_executor.py | 2 + .../app/gateway/workflow_proposal_planner.py | 213 +++++++++++++++++- offline-backend-20260512/backend/config.yaml | 19 ++ .../collection/thread_harvester.py | 41 +++- .../agents/deep_research/runners/basic.py | 112 +++++---- .../deerflow/agents/lead_agent/agent.py | 10 + .../persistence/deep_research_events/sql.py | 21 +- .../persistence/workflow_events/sql.py | 19 +- .../harness/deerflow/workflows/events.py | 3 + .../deerflow/workflows/nodes/agents.py | 64 ++++-- .../backend/packages/harness/pyproject.toml | 5 +- .../tests/test_deep_research_persistence.py | 19 ++ .../tests/test_workflow_concurrency.py | 28 +++ .../tests/test_workflow_definitions.py | 6 +- .../backend/tests/test_workflow_engine.py | 54 +++++ .../backend/tests/test_workflow_runs.py | 36 ++- scripts/start-backend.ps1 | 2 +- scripts/start-backend.sh | 6 +- 28 files changed, 836 insertions(+), 135 deletions(-) diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md index 1c872a83..41ead2ba 100644 --- a/offline-backend-20260512/backend/CLAUDE.md +++ b/offline-backend-20260512/backend/CLAUDE.md @@ -75,6 +75,13 @@ When making code changes, you MUST update the relevant documentation: - Keep documentation synchronized with the codebase at all times - Ensure accuracy and timeliness of all documentation +### Workflow Studio stream and canvas compatibility + +- `app/gateway/workflow_agent_runner.py` is the boundary between LangGraph token chunks and durable workflow events. Provider reasoning in `AIMessage.additional_kwargs.reasoning_content`/`reasoning` is emitted to the UI as balanced inline `` blocks; only visible answer content becomes the node's `text` output for downstream nodes. It must also emit every original `[AIMessage|ToolMessage, metadata]` tuple as `node.message.data.messages` with no whitelist, preview or length crop. Tool results and `additional_kwargs.clarification` are needed by the Studio's DeerFlow-compatible message UI and must survive durable SSE replay. +- An agent `ask_clarification` ToolMessage pauses the workflow at that agent node (`awaiting_input`), not as a terminal run. The pause records the source `nodeRunId`/`toolCallId`; resuming turns the submitted reply into the next user message on the same `wf-{run}-{node}` LangGraph thread. Keep the raw ToolMessage authoritative; do not invent a second card payload. +- `app/gateway/workflow_proposal_planner.py` converts the workflow controller's selected catalog roles into per-agent `mission`/`deliverable`/`scope`/`handoff` contracts. Only a selected visible `agentId` may contribute a contract; every omitted or malformed field falls back to a deterministic role-aware contract. Preserve this prompt-level worker boundary when adding strategies: researchers must not silently become final writers, and sequential roles must receive their upstream handoff binding plus an explicit next handoff. +- `app/gateway/routers/workflows_coze_compat.py` must return both `creator.self: true` and editable VCS metadata for an owned standalone workflow. Coze Playground derives its preview/read-only mode from those fields, so omitting `creator.self` disables node dragging even when the standalone page passes `readonly={false}`. + ## Commands **Root directory** (for full application): @@ -201,7 +208,7 @@ The backend is split into two layers with a strict dependency direction: ### Workflow Studio -Coze Studio is canvas-only; ChatDev is reference-only. Plan + status per phase: `docs/WORKFLOW_STUDIO_BACKEND_DEV_ZH.md`; decisions: `docs/adr/0001-workflow-studio-runtime.md`; contracts: `docs/WORKFLOW_SCHEMA_V1_ZH.md` + `docs/WORKFLOW_SSE_V1_ZH.md` + `docs/WORKFLOW_OPENAPI_DRAFT_V1.yaml`. Master switch + all limits: `config.yaml` → `workflows` (`deerflow.config.workflow_config.WorkflowConfig`). `workflows.agent_recursion_limit` defaults to 250: LangGraph counts each model→tool research exchange as about two super-steps, so workflow agent/skill nodes must not fall back to a chat-sized 60-step limit. In the local Studio proxy, keep Rsbuild `server.compress: false`; dev gzip buffers small SSE frames and makes live planning/run output arrive only at completion despite the backend's `no-transform` / `X-Accel-Buffering: no` headers. +Coze Studio is canvas-only; ChatDev is reference-only. Plan + status per phase: `docs/WORKFLOW_STUDIO_BACKEND_DEV_ZH.md`; decisions: `docs/adr/0001-workflow-studio-runtime.md`; contracts: `docs/WORKFLOW_SCHEMA_V1_ZH.md` + `docs/WORKFLOW_SSE_V1_ZH.md` + `docs/WORKFLOW_OPENAPI_DRAFT_V1.yaml`. Master switch + all limits: `config.yaml` → `workflows` (`deerflow.config.workflow_config.WorkflowConfig`). `workflows.agent_recursion_limit` defaults to 250: LangGraph counts each model→tool research exchange as about two super-steps, so workflow agent/skill nodes must not fall back to a chat-sized 60-step limit. In the local Studio proxy, keep Rsbuild `server.compress: false`; dev gzip buffers small SSE frames and makes live planning/run output arrive only at completion despite the backend's `no-transform` / `X-Accel-Buffering: no` headers. The composer’s optional `modelName` must be validated against `AppConfig.models`: it selects the controller model and is copied only into newly generated candidate agent configs (or the projected direct-agent task), never injected into an already-confirmed proposal/run. **Execution kernel is a custom DAG scheduler, not a compiled LangGraph.** LangGraph stays the kernel *inside* an `agent`/`skill` node. The reason is single-source-of-truth recovery: a re-leased or resumed run replays completed node outputs from `workflow_node_runs` plus the `workflow_run_events` log, so a second (checkpoint) copy of workflow-level state would have no defensible tie-breaker. See the ADR's phase-2 amendment. diff --git a/offline-backend-20260512/backend/Makefile b/offline-backend-20260512/backend/Makefile index deb0b446..7747faa7 100644 --- a/offline-backend-20260512/backend/Makefile +++ b/offline-backend-20260512/backend/Makefile @@ -2,7 +2,7 @@ install: uv sync dev: - PYTHONPATH=. uv run uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001 --reload --reload-exclude ".deer-flow/*" --reload-exclude ".runtime/*" + PYTHONPATH=. uv run uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001 --reload --reload-dir app --reload-dir packages --reload-exclude "$(CURDIR)/.deer-flow" --reload-exclude "$(CURDIR)/.runtime" gateway: PYTHONPATH=. uv run uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001 diff --git a/offline-backend-20260512/backend/README.md b/offline-backend-20260512/backend/README.md index 474fb236..c158c48a 100644 --- a/offline-backend-20260512/backend/README.md +++ b/offline-backend-20260512/backend/README.md @@ -55,6 +55,11 @@ The single LangGraph agent (`lead_agent`) is the runtime entry point, created vi - **Subagent delegation** for parallel task execution - **System prompt** with skills injection, memory context, and working directory guidance +### Workflow Studio runtime + +- Candidate workflow graphs remain editable after they are loaded into the Coze-compatible canvas; the compatibility response explicitly marks the current owner as editable so node dragging is not mistaken for a read-only preview. +- Workflow-agent SSE forwards provider `reasoning_content` as an inline `…` trace for the UI while keeping the executable node output limited to the visible answer text. Every native LangGraph `[AIMessage|ToolMessage, metadata]` frame is additionally persisted and sent as `node.message.data.messages` without projection or length cropping, so the Studio can replay DeerFlow's tool-result, search-result and `ask_clarification` cards exactly after reconnecting. + ### Middleware Chain Middlewares execute in strict order, each handling a specific concern: @@ -219,6 +224,21 @@ both streams `no-transform` and `X-Accel-Buffering: no`; `apps/coze-studio` also its Rsbuild development server `compress: false`. Agent/skill nodes use the bounded `workflows.agent_recursion_limit` (default `250`), rather than a chat-sized 60-step limit, because a model-to-tool research turn consumes about two LangGraph super-steps. +The composer may also send a configured `modelName`: the server validates it against +the safe model catalog, uses it for the workflow controller, and writes it into the +generated candidate agent nodes. For a direct single-agent task it overrides only that +projected node; confirmation still executes the immutable selected proposal snapshot. +The controller additionally creates a per-selected-agent task contract (`mission`, +`deliverable`, `scope`, `handoff`). The graph builder accepts contracts only for +catalog-visible, selected agent ids and injects each one into that worker's prompt, so +parallel branches receive complementary research assignments and sequential branches +have an explicit review/handoff boundary rather than all workers attempting the whole +user task. Missing controller fields receive a deterministic role-aware contract. +When an agent emits DeerFlow's native `ask_clarification` ToolMessage, the workflow +enters `awaiting_input` rather than completing or failing. The pause event references +the originating `nodeRunId` and `toolCallId`; the browser renders that original card +inside the agent message and `POST /runs/{run_id}/resume` supplies the user's reply as +the next turn of the same agent thread. The current planner is deliberately constrained: it can rank supported strategies and focus resources, but the server builds the graph from the already configured agent diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index 6ddb472c..0def4d8d 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/offline-backend-20260512/backend/app/gateway/app.py @@ -1093,12 +1093,25 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: ) ensure_template_builder_agent() - except Exception: - logger.exception( - "Template-builder agent seeding failed at startup (non-fatal)" - ) - - # 深度研究「收集员」内置 agent(对话式信息收集)。同样必须在 + except Exception: + logger.exception( + "Template-builder agent seeding failed at startup (non-fatal)" + ) + + # 实验性「页面研报设计师」：在 _sync_legacy_agents 前补齐 agent 目录与 + # 公共页面生成技能，首次启动即可在智能体管理页和普通对话中复用。 + try: + from app.gateway.routers._page_report_designer_seed import ( + ensure_page_report_designer_agent, + ) + + ensure_page_report_designer_agent() + except Exception: + logger.exception( + "Page-report designer agent seeding failed at startup (non-fatal)" + ) + + # 深度研究「收集员」内置 agent(对话式信息收集)。同样必须在 # _sync_legacy_agents 之前补建目录,由后者 upsert 进 agents 表, # 深度研究页的收集对话与智能体管理页即可直接使用。 try: diff --git a/offline-backend-20260512/backend/app/gateway/deep_research_job_executor.py b/offline-backend-20260512/backend/app/gateway/deep_research_job_executor.py index e0d4a006..d308499a 100644 --- a/offline-backend-20260512/backend/app/gateway/deep_research_job_executor.py +++ b/offline-backend-20260512/backend/app/gateway/deep_research_job_executor.py @@ -1501,27 +1501,42 @@ class DeepResearchJobExecutor: runtime_thread_id=runtime_thread_id, query=query, ) - await events.emit( - "collection_completed", - phase="collecting", - payload={ - "source": "chat_thread", - "materialCount": len(harvested), - "agentSummaryChars": len(agent_summary), - }, - ) - for material in harvested: + # Freeze the harvested list before publishing any progress. Source + # event persistence is intentionally best-effort: a bad payload or a + # temporarily unavailable event table must never prevent the writer + # from receiving this already-collected evidence. + material_provider = StaticHarvestMaterialProvider(harvested) + try: await events.emit( - "source_added", + "collection_completed", phase="collecting", payload={ - "sourceId": material.id, - "title": material.title, - "url": material.url, - "sourceType": material.source_type, - "source": material.to_persistence_dict(material.query), + "source": "chat_thread", + "materialCount": material_provider.material_count, + "agentSummaryChars": len(agent_summary), }, ) + except Exception: # noqa: BLE001 - progress telemetry must not lose evidence + logger.warning("chat deep-research collection event failed; retaining harvested materials", exc_info=True) + for material in harvested: + try: + await events.emit( + "source_added", + phase="collecting", + payload={ + "sourceId": material.id, + "title": material.title, + "url": material.url, + "sourceType": material.source_type, + "source": material.to_persistence_dict(material.query), + }, + ) + except Exception: # noqa: BLE001 - source rows are a projection, not the input + logger.warning( + "chat deep-research source event failed for %s; retaining harvested materials", + material.id, + exc_info=True, + ) if not harvested: await events.emit( "warning", @@ -1532,7 +1547,7 @@ class DeepResearchJobExecutor: "recoverable": True, }, ) - return StaticHarvestMaterialProvider(harvested) + return material_provider async def _generate_report_summary( self, diff --git a/offline-backend-20260512/backend/app/gateway/routers/workflow_planning.py b/offline-backend-20260512/backend/app/gateway/routers/workflow_planning.py index 016b2aec..d41d598f 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/workflow_planning.py +++ b/offline-backend-20260512/backend/app/gateway/routers/workflow_planning.py @@ -47,6 +47,9 @@ _SSE_HEARTBEAT_SECONDS = 1.0 class CreatePlanningSessionBody(BaseModel): query: str = Field(min_length=1, max_length=12_000) + model_name: str | None = Field(default=None, alias="modelName", max_length=191) + + model_config = {"populate_by_name": True} class UpdateProposalGraphBody(BaseModel): @@ -66,6 +69,18 @@ def _require_enabled() -> None: raise HTTPException(status_code=503, detail="Workflow Studio is disabled") +def _selected_model_name(value: str | None) -> str | None: + model_name = str(value or "").strip() + if not model_name: + return None + if get_app_config().get_model_config(model_name) is None: + raise HTTPException( + status_code=400, + detail={"code": "WORKFLOW_MODEL_INVALID", "message": "所选模型不存在或已不可用"}, + ) + return model_name + + def _store(request: Request) -> WorkflowPlanningStore: store = getattr(request.app.state, "workflow_planning_store", None) if store is None: @@ -225,11 +240,13 @@ async def _create_planning_session( }, ) from exc try: + model_name = _selected_model_name(body.model_name) proposals = await build_workflow_proposals( request.app, source_graph=source_graph, query=body.query.strip(), owner_id=user_id, + model_name=model_name, source_canvas_schema=str(definition.get("draft_canvas_schema") or "{}"), on_progress=on_progress, ) diff --git a/offline-backend-20260512/backend/app/gateway/routers/workflow_runs.py b/offline-backend-20260512/backend/app/gateway/routers/workflow_runs.py index 1d0898bb..5c9fe8f6 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/workflow_runs.py +++ b/offline-backend-20260512/backend/app/gateway/routers/workflow_runs.py @@ -56,6 +56,7 @@ class StartRunBody(BaseModel): # 对话面板的显式定向任务：不是打断既有 run，而是从当前图中选取一个已 # 绑定的 Agent，投影成可审计的 start → agent → output 新运行。 target_agent_id: str | None = Field(default=None, alias="targetAgentId", max_length=191) + model_name: str | None = Field(default=None, alias="modelName", max_length=191) # 规划会话中的已选候选图。二者必须同时给出；正式执行只读取服务端持久化 # 的 proposal 快照，永不接受浏览器直接塞入的临时 graph。 planning_session_id: str | None = Field(default=None, alias="planningSessionId", max_length=64) @@ -105,6 +106,18 @@ def _require_enabled() -> None: raise HTTPException(status_code=503, detail="Workflow Studio is disabled") +def _selected_model_name(value: str | None) -> str | None: + model_name = str(value or "").strip() + if not model_name: + return None + if get_app_config().get_model_config(model_name) is None: + raise HTTPException( + status_code=400, + detail={"code": "WORKFLOW_MODEL_INVALID", "message": "所选模型不存在或已不可用"}, + ) + return model_name + + def _runs(request: Request) -> WorkflowRunStore: store = getattr(request.app.state, "workflow_run_store", None) if store is None: @@ -216,7 +229,11 @@ async def _run_validation_issues( ) -def _agent_task_graph(graph: WorkflowGraph, target_agent_id: str) -> WorkflowGraph: +def _agent_task_graph( + graph: WorkflowGraph, + target_agent_id: str, + model_name: str | None = None, +) -> WorkflowGraph: """Build an explicit, durable one-agent task graph from a canvas graph. A chat composer cannot safely "message" a running node: that loses DAG @@ -260,6 +277,8 @@ def _agent_task_graph(graph: WorkflowGraph, target_agent_id: str) -> WorkflowGra bindings["workflowInput"] = "{{ inputs }}" config["inputBindings"] = bindings config["agentName"] = str(config.get("agentName") or agent.name or target_agent_id) + if model_name: + config["modelName"] = model_name return WorkflowGraph.model_validate( { @@ -434,8 +453,9 @@ async def start_run(request: Request, workflow_id: str, body: StartRunBody) -> d ) from exc target_agent_id = str(body.target_agent_id or "").strip() + model_name = _selected_model_name(body.model_name) if target_agent_id: - graph = _agent_task_graph(graph, target_agent_id) + graph = _agent_task_graph(graph, target_agent_id, model_name) issues = await _run_validation_issues( request, diff --git a/offline-backend-20260512/backend/app/gateway/routers/workflows_coze_compat.py b/offline-backend-20260512/backend/app/gateway/routers/workflows_coze_compat.py index 920ea657..039677e7 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/workflows_coze_compat.py +++ b/offline-backend-20260512/backend/app/gateway/routers/workflows_coze_compat.py @@ -181,7 +181,11 @@ def _to_coze_canvas(definition: dict[str, Any], *, is_published: bool = False) - "schema_json": _canvas_schema_text(definition), "status": 1, "collaborator_mode": "Open", - "creator": {"user_id": owner, "nickname": owner}, + # Playground derives editability from this flag even when the + # VCS `can_edit` value is true. Omitting it puts an owned + # standalone workflow into preview/read-only mode, which also + # disables dragging candidate nodes on the canvas. + "creator": {"user_id": owner, "nickname": owner, "self": True}, "is_published": is_published, }, "vcs_data": {"can_edit": True}, diff --git a/offline-backend-20260512/backend/app/gateway/services.py b/offline-backend-20260512/backend/app/gateway/services.py index dea1c159..8484f189 100644 --- a/offline-backend-20260512/backend/app/gateway/services.py +++ b/offline-backend-20260512/backend/app/gateway/services.py @@ -413,6 +413,9 @@ _CONTEXT_CONFIGURABLE_KEYS: frozenset[str] = frozenset( "subagent_enabled", "max_concurrent_subagents", "agent_name", + # 页面研报设计师由前端随消息带入当前界面明暗模式；需要进入 agent + # 提示词及 ToolRuntime.context，才能选择同色系模板。 + "presentation_theme", "is_bootstrap", "is_writing_setup", "is_report_structure_setup", diff --git a/offline-backend-20260512/backend/app/gateway/workflow_agent_runner.py b/offline-backend-20260512/backend/app/gateway/workflow_agent_runner.py index 0bda9168..d9faf460 100644 --- a/offline-backend-20260512/backend/app/gateway/workflow_agent_runner.py +++ b/offline-backend-20260512/backend/app/gateway/workflow_agent_runner.py @@ -12,6 +12,7 @@ asks for ``shared_run_thread``, which keeps the whole run in one conversation. from __future__ import annotations +import json import logging from collections.abc import Awaitable, Callable from typing import Any @@ -37,12 +38,99 @@ def _extract_text(content: Any) -> str: return "" +def _extract_reasoning(message: AIMessage) -> str: + """Read provider reasoning that LangChain keeps outside ``content``. + + Patched OpenAI/vLLM providers preserve the streaming field under + ``additional_kwargs.reasoning_content``. Some compatible providers use + the shorter ``reasoning`` spelling, so accept both without making either + field part of the workflow's executable output. + """ + + additional_kwargs = getattr(message, "additional_kwargs", None) + if not isinstance(additional_kwargs, dict): + return "" + for key in ("reasoning_content", "reasoning"): + value = additional_kwargs.get(key) + text = _extract_text(value) + if text: + return text + if isinstance(value, str) and value: + return value + return "" + + def _thread_id(run_id: str, node_id: str, thread_mode: str) -> str: if thread_mode == "shared_run_thread": return f"wf-{run_id}" return f"wf-{run_id}-{node_id}" +def _native_message_frame(chunk: Any) -> list[Any]: + """Keep the DeerFlow/LangGraph ``messages`` tuple intact for the client. + + The workflow event envelope adds run/node identity, but the nested value is + deliberately the same two-item shape consumed by DeerFlow's chat UI: + ``[message, metadata]``. Do not project tool args, result content, or + ``additional_kwargs`` here; those fields are what power rich tool and user + assistance cards after a live stream reconnects. + """ + + def encode(value: Any) -> Any: + if hasattr(value, "model_dump"): + return encode(value.model_dump(mode="json")) + if isinstance(value, dict): + return {str(key): encode(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [encode(item) for item in value] + # Force a JSON-compatible copy without discarding or truncating any + # field. LangGraph metadata is normally already JSON-native; ``str`` + # is only a serialization fallback for provider-specific objects. + try: + json.dumps(value, ensure_ascii=False) + except (TypeError, ValueError): + return str(value) + return value + + return encode(chunk if isinstance(chunk, tuple) else (chunk, {})) + + +def _resume_message(payload: dict[str, Any]) -> str: + """Represent a workflow resume as the next user turn in the same thread.""" + return ( + "用户已回复你刚才的协助请求。请基于已有对话继续当前任务，不要重复询问已经回答的信息。\n" + "\n" + f"{json.dumps(payload, ensure_ascii=False, separators=(',', ':'), default=str)}\n" + "" + ) + + +def _clarification_pending(message: ToolMessage) -> dict[str, Any] | None: + """Identify DeerFlow's native ``ask_clarification`` tool result. + + The complete card payload continues to live in the untouched ToolMessage + sent through ``node.message``. The pause descriptor only carries the + stable tool-call identity required to reconnect it to this node run. + """ + if str(getattr(message, "name", "") or "") != "ask_clarification": + return None + additional_kwargs = getattr(message, "additional_kwargs", None) + clarification = ( + additional_kwargs.get("clarification") + if isinstance(additional_kwargs, dict) + else None + ) + if not isinstance(clarification, dict): + return None + call_id = str(getattr(message, "tool_call_id", "") or getattr(message, "id", "") or "") + question = str(clarification.get("question") or "") + return { + "toolCallId": call_id, + "prompt": question or "智能体需要你补充信息后才能继续。", + "actions": ["submit"], + } + + class WorkflowAgentRunner: def __init__(self, app: Any) -> None: self._app = app @@ -60,6 +148,8 @@ class WorkflowAgentRunner: cancel: Any = None, on_delta: Callable[[str], Awaitable[None]] | None = None, on_tool: Callable[[dict[str, Any]], Awaitable[None]] | None = None, + on_message: Callable[[list[Any]], Awaitable[None]] | None = None, + resume_payload: dict[str, Any] | None = None, model_name: str | None = None, disable_tools: bool = False, thinking_enabled: bool | None = None, @@ -119,9 +209,33 @@ class WorkflowAgentRunner: final_state: dict[str, Any] | None = None streamed: dict[str, list[str]] = {} last_message_id = "" + open_reasoning_message_ids: set[str] = set() + + async def emit_display_delta(text: str) -> None: + if on_delta is not None and text: + await on_delta(text) + + async def close_reasoning(*, except_message_id: str | None = None) -> None: + """Keep thinking blocks well-formed across tool/model turns.""" + + for message_id in tuple(open_reasoning_message_ids): + if message_id == except_message_id: + continue + open_reasoning_message_ids.remove(message_id) + await emit_display_delta("") + + initial_messages = [ + { + "role": "user", + "content": _resume_message(resume_payload) + if resume_payload is not None + else prompt, + } + ] + awaiting_input: dict[str, Any] | None = None async for mode, chunk in agent.astream( - {"messages": [{"role": "user", "content": prompt}]}, + {"messages": initial_messages}, config=config, stream_mode=["values", "messages"], ): @@ -133,35 +247,44 @@ class WorkflowAgentRunner: continue if mode != "messages": continue + if on_message is not None: + await on_message(_native_message_frame(chunk)) message = chunk[0] if isinstance(chunk, tuple) and chunk else chunk if isinstance(message, ToolMessage): # Tool result for an earlier call: close its step in the event log. + await close_reasoning() call_id = str(getattr(message, "tool_call_id", "") or "") if call_id and call_id not in finished_tool_calls: finished_tool_calls.add(call_id) if on_tool is not None: - content = getattr(message, "content", "") - if not isinstance(content, str): - content = str(content) await on_tool( { "phase": "finished", "id": call_id, "name": tool_names.get(call_id) or getattr(message, "name", "") or "", "ok": getattr(message, "status", None) != "error", - "summary": content[:200], + "result": getattr(message, "content", ""), } ) + awaiting_input = _clarification_pending(message) + if awaiting_input is not None: + break continue if not isinstance(message, AIMessage): continue + message_id = str(getattr(message, "id", "") or "__workflow_reasoning__") + reasoning = _extract_reasoning(message) + if reasoning: + if message_id not in open_reasoning_message_ids: + open_reasoning_message_ids.add(message_id) + await emit_display_delta("") + await emit_display_delta(reasoning) text = _extract_text(getattr(message, "content", "")) if text: - message_id = str(getattr(message, "id", "") or "") + await close_reasoning() streamed.setdefault(message_id, []).append(text) last_message_id = message_id - if on_delta is not None: - await on_delta(text) + await emit_display_delta(text) for call in getattr(message, "tool_calls", None) or []: call_id = str(call.get("id") or "") if not call_id or call_id in seen_tool_calls: @@ -178,6 +301,23 @@ class WorkflowAgentRunner: } ) + # A reasoning-only turn that delegates to a tool must end before + # the tool result and next model turn arrive. Otherwise a later + # answer could be swallowed by one long unclosed `` block. + if getattr(message, "tool_calls", None): + await close_reasoning() + + await close_reasoning() + + if awaiting_input is not None: + return { + "awaiting_input": awaiting_input, + "thread_id": thread_id, + "messages": [], + "artifacts": self._artifacts(final_state), + "usage": {}, + } + final_text = "".join(streamed.get(last_message_id, ())) or self._final_text(final_state) if not final_text.strip(): raise WorkflowError( diff --git a/offline-backend-20260512/backend/app/gateway/workflow_executor.py b/offline-backend-20260512/backend/app/gateway/workflow_executor.py index 78081eef..644441f2 100644 --- a/offline-backend-20260512/backend/app/gateway/workflow_executor.py +++ b/offline-backend-20260512/backend/app/gateway/workflow_executor.py @@ -273,8 +273,10 @@ class WorkflowRunExecutor: "prompt": paused.pending_input.get("prompt") or "", "formSchema": paused.pending_input.get("formSchema") or {}, "actions": paused.pending_input.get("actions") or ["submit"], + "toolCallId": paused.pending_input.get("toolCallId") or "", }, node_id=paused.node_id, + node_run_id=ctx.node_run_ids.get(paused.node_id), ) # ── helpers ───────────────────────────────────────────────────────── diff --git a/offline-backend-20260512/backend/app/gateway/workflow_proposal_planner.py b/offline-backend-20260512/backend/app/gateway/workflow_proposal_planner.py index 95a326ed..406540e2 100644 --- a/offline-backend-20260512/backend/app/gateway/workflow_proposal_planner.py +++ b/offline-backend-20260512/backend/app/gateway/workflow_proposal_planner.py @@ -23,6 +23,8 @@ from deerflow.workflows.schemas import WorkflowGraph _JSON_BLOCK = re.compile(r"```(?:json)?\s*(.+?)\s*```", re.DOTALL | re.IGNORECASE) _QUERY_PREVIEW = 64 _STRATEGIES = ("collaborative", "parallel_research", "quick_answer") +_TASK_CONTRACT_FIELDS = ("mission", "deliverable", "scope", "handoff") +_TASK_CONTRACT_FIELD_LIMIT = 600 PlanningProgressEmitter = Callable[[dict[str, Any]], Awaitable[None]] @@ -106,6 +108,103 @@ def _agent_skills(node: Any) -> list[str]: return [str(item) for item in raw] if isinstance(raw, list) else [] +def _default_task_contract( + node: Any, + *, + strategy: str, + index: int, + total: int, +) -> dict[str, str]: + """Describe one worker's bounded responsibility when the controller omits it. + + The controller may specialize these fields, but an omitted model field must + never send a worker back to a vague "help with the task" instruction. + """ + role = _agent_name(node) + capability = _agent_description(node) or "其已配置的专业能力" + if strategy == "parallel_research" and index == total - 1: + return { + "mission": f"作为{role}整合各并行调研分支，不重新发起无关调研。", + "deliverable": "输出证据对照、冲突与不确定性、结论和可执行的下一步建议。", + "scope": "只使用 workflowInput 和 evidencePack；将来源事实、推断和待验证假设明确区分。", + "handoff": "交给深度研究报告写作节点；用清晰的结论结构和证据缺口帮助其成文。", + } + if strategy == "parallel_research": + return { + "mission": f"作为{role}从一个独立视角调查用户问题，发挥：{capability}。", + "deliverable": "交付结构化证据笔记：关键事实、来源或验证方法、判断、置信度与信息缺口。", + "scope": "只完成本调研分支；不替代综合角色写最终报告，不把未验证内容表述为事实。", + "handoff": "结果会进入 Evidence Pack，供综合角色比对并交给报告写作节点。", + } + if strategy == "collaborative" and index == 0: + return { + "mission": f"作为{role}完成第一阶段任务拆解和基础分析，发挥：{capability}。", + "deliverable": "交付问题分解、关键判断、证据或验证路径、明确的假设与待补信息。", + "scope": "只建立可靠的上游材料，不假装已经完成后续角色的审阅或最终交付。", + "handoff": "将可复核的结构化结果交给下游角色继续审阅和补强。", + } + if strategy == "collaborative": + return { + "mission": f"作为{role}审阅并补强上游成果，发挥：{capability}。", + "deliverable": "交付经过校正的结论、补充证据、冲突说明、剩余风险与下一步建议。", + "scope": "围绕 workflowInput 和 upstreamResult 工作；保留有效材料，明确标注不能确认的部分。", + "handoff": "将可直接被下一阶段使用的完整结果交接给下游角色。", + } + return { + "mission": f"作为{role}快速分析用户问题，发挥：{capability}。", + "deliverable": "交付直接结论、可验证依据、关键假设和仍需用户补充的信息。", + "scope": "只处理当前用户问题；信息不足时先说明限制，不能臆造事实或虚构执行结果。", + "handoff": "这是该策略的最终用户可读输出，无需自行编排其他智能体。", + } + + +def _contract_text(value: Any, fallback: str) -> str: + """Accept short controller-owned text while keeping prompt assembly bounded.""" + text = _text(value).replace("\x00", " ") + return (text or fallback)[:_TASK_CONTRACT_FIELD_LIMIT] + + +def _task_contracts( + model_result: dict[str, Any], + agents: list[Any], + *, + strategy: str, +) -> dict[str, dict[str, str]]: + """Map controller task contracts to selected, visible worker ids only.""" + raw_contracts = model_result.get("taskContracts") or model_result.get("task_contracts") + strategy_contracts = raw_contracts.get(strategy) if isinstance(raw_contracts, dict) else {} + contracts: dict[str, dict[str, str]] = {} + for index, agent in enumerate(agents): + agent_id = _agent_id(agent) + fallback = _default_task_contract( + agent, + strategy=strategy, + index=index, + total=len(agents), + ) + raw = strategy_contracts.get(agent_id) if isinstance(strategy_contracts, dict) else None + contracts[agent_id] = { + field: _contract_text(raw.get(field) if isinstance(raw, dict) else None, fallback[field]) + for field in _TASK_CONTRACT_FIELDS + } + return contracts + + +def _task_contract_suffix(contract: dict[str, str] | None) -> str: + if not contract: + return "" + return ( + "\n" + "这是工作流总控为你分配的固定岗位。只完成这个岗位，不自行改派角色或重复总控职责。\n" + f"专属任务：{contract['mission']}\n" + f"交付产物：{contract['deliverable']}\n" + f"工作范围与质量要求：{contract['scope']}\n" + f"交接方式：{contract['handoff']}\n" + "如果缺少完成本岗位必需的信息，说明缺口并通过用户协助工具提问；不要用虚构内容填补。\n" + "" + ) + + def _catalog_payload(agents: list[Any]) -> list[dict[str, Any]]: """Return metadata only; agent SOUL/configuration must never enter a prompt.""" unique: dict[str, dict[str, Any]] = {} @@ -219,16 +318,26 @@ def _chat_input_schema(graph: WorkflowGraph) -> dict[str, Any]: return schema -def _agent_config(node: Any, *, prompt_suffix: str = "", bindings: dict[str, str] | None = None) -> dict[str, Any]: +def _agent_config( + node: Any, + *, + prompt_suffix: str = "", + task_contract: dict[str, str] | None = None, + bindings: dict[str, str] | None = None, + model_name: str | None = None, +) -> dict[str, Any]: config = copy.deepcopy(node.config) template = _text(config.get("promptTemplate") or config.get("prompt_template"), "请完成用户的请求。") - config["promptTemplate"] = f"{template}\n\n{prompt_suffix}".strip() + prompt_parts = [template, prompt_suffix, _task_contract_suffix(task_contract)] + config["promptTemplate"] = "\n\n".join(part for part in prompt_parts if part).strip() original = config.get("inputBindings") or config.get("input_bindings") resolved = dict(original) if isinstance(original, dict) else {} resolved["workflowInput"] = "{{ inputs }}" if bindings: resolved.update(bindings) config["inputBindings"] = resolved + if model_name: + config["modelName"] = model_name return config @@ -241,7 +350,15 @@ def _output_node(node_id: str, source_id: str) -> dict[str, Any]: } -def _single_agent_graph(source: WorkflowGraph, node: Any, *, name: str, strategy: str) -> dict[str, Any]: +def _single_agent_graph( + source: WorkflowGraph, + node: Any, + *, + name: str, + strategy: str, + task_contract: dict[str, str] | None = None, + model_name: str | None = None, +) -> dict[str, Any]: source_id = str(node.id) start_id = f"__{strategy}_start__" output_id = f"__{strategy}_output__" @@ -260,6 +377,8 @@ def _single_agent_graph(source: WorkflowGraph, node: Any, *, name: str, strategy "config": _agent_config( node, prompt_suffix="你负责快速澄清用户问题、给出可验证的结论；信息不足时明确列出缺口。", + task_contract=task_contract, + model_name=model_name, ), }, _output_node(output_id, source_id), @@ -271,10 +390,23 @@ def _single_agent_graph(source: WorkflowGraph, node: Any, *, name: str, strategy } -def _parallel_research_graph(source: WorkflowGraph, agents: list[Any]) -> dict[str, Any]: +def _parallel_research_graph( + source: WorkflowGraph, + agents: list[Any], + *, + task_contracts: dict[str, dict[str, str]] | None = None, + model_name: str | None = None, +) -> dict[str, Any]: """Fan out collection roles and use the last configured agent as synthesizer.""" if len(agents) < 2: - return _single_agent_graph(source, agents[0], name="快速验证方案", strategy="quick_answer") + return _single_agent_graph( + source, + agents[0], + name="快速验证方案", + strategy="quick_answer", + task_contract=(task_contracts or {}).get(_agent_id(agents[0])), + model_name=model_name, + ) start_id = "__parallel_start__" evidence_id = "__parallel_evidence__" report_id = "__parallel_report__" @@ -294,6 +426,8 @@ def _parallel_research_graph(source: WorkflowGraph, agents: list[Any]) -> dict[s "config": _agent_config( researcher, prompt_suffix="你是并行调研角色。围绕用户任务收集事实、假设和可引用依据，输出结构化要点给汇总角色。", + task_contract=(task_contracts or {}).get(_agent_id(researcher)), + model_name=model_name, ), } ) @@ -319,7 +453,9 @@ def _parallel_research_graph(source: WorkflowGraph, agents: list[Any]) -> dict[s "config": _agent_config( synthesis, prompt_suffix="你是综合分析角色。基于 Evidence Pack 比较各调研结果，标注冲突、假设和证据不足，形成面向用户的结论与下一步建议。", + task_contract=(task_contracts or {}).get(_agent_id(synthesis)), bindings=synthesis_bindings, + model_name=model_name, ), } ) @@ -375,7 +511,13 @@ def _parallel_research_graph(source: WorkflowGraph, agents: list[Any]) -> dict[s } -def _collaborative_graph(source: WorkflowGraph, agents: list[Any]) -> dict[str, Any]: +def _collaborative_graph( + source: WorkflowGraph, + agents: list[Any], + *, + task_contracts: dict[str, dict[str, str]] | None = None, + model_name: str | None = None, +) -> dict[str, Any]: """Build a sequential expert hand-off from the controller's selection.""" if not agents: raise WorkflowPlannerUnavailableError("总控没有选出可执行的协作智能体") @@ -399,7 +541,13 @@ def _collaborative_graph(source: WorkflowGraph, agents: list[Any]) -> dict[str, "id": node_id, "type": agent.type, "name": _agent_name(agent), - "config": _agent_config(agent, prompt_suffix=role_hint, bindings=bindings), + "config": _agent_config( + agent, + prompt_suffix=role_hint, + task_contract=(task_contracts or {}).get(_agent_id(agent)), + bindings=bindings, + model_name=model_name, + ), } ) edges.append({"id": f"{previous_id}_to_{node_id}", "source": previous_id, "target": node_id}) @@ -435,6 +583,7 @@ async def _ask_workflow_controller( query: str, owner_id: str, agents: list[Any], + model_name: str | None = None, on_progress: PlanningProgressEmitter | None = None, ) -> dict[str, Any]: """Ask the dedicated controller to understand intent and select agents. @@ -479,9 +628,16 @@ async def _ask_workflow_controller( '"recommendedStrategy":"collaborative|parallel_research|quick_answer",' '"strategyOrder":["..."],' '"agentSelections":{"collaborative":["agentId"],"parallel_research":["agentId"],"quick_answer":["agentId"]},' + '"taskContracts":{"策略":{"agentId":{"mission":"不超过180字的专属任务",' + '"deliverable":"不超过180字的交付物",' + '"scope":"不超过180字的边界和证据要求",' + '"handoff":"不超过180字的交接说明"}}},' '"titles":{"策略":"不超过20字标题"},' '"summaries":{"策略":"不超过120字方案说明"},' - '"rationales":{"策略":"不超过120字选择理由"}}。\n\n' + '"rationales":{"策略":"不超过120字选择理由"}}。' + "taskContracts 只能为已选中的逐字匹配 agentId 提供岗位说明；" + "并行策略中每个调研角色必须分配互补视角，最后的综合角色必须明确 Evidence Pack 的使用方式；" + "顺序协作策略必须明确上一角色交接给下一角色的内容。\n\n" f"用户需求：{query}\n可用业务智能体目录：{json.dumps(catalog, ensure_ascii=False)}" ) received_first_delta = False @@ -506,6 +662,7 @@ async def _ask_workflow_controller( disable_tools=True, thinking_enabled=False, force_disable_thinking=True, + model_name=model_name, on_delta=on_delta, ) parsed = _decode_planner_json(_text(result.get("text"))) @@ -564,6 +721,7 @@ async def build_workflow_proposals( source_graph: WorkflowGraph, query: str, owner_id: str, + model_name: str | None = None, source_canvas_schema: str = "{}", on_progress: PlanningProgressEmitter | None = None, ) -> list[dict[str, Any]]: @@ -589,6 +747,7 @@ async def build_workflow_proposals( query=query, owner_id=owner_id, agents=agents, + model_name=model_name, on_progress=on_progress, ) await _emit_progress( @@ -604,6 +763,21 @@ async def build_workflow_proposals( parallel_agents = _strategy_selection( model_result, agents, "parallel_research", fallback_count=min(4, len(agents)) ) + collaborative_contracts = _task_contracts( + model_result, + collaborative_agents, + strategy="collaborative", + ) + quick_contracts = _task_contracts( + model_result, + quick_agents, + strategy="quick_answer", + ) + parallel_contracts = _task_contracts( + model_result, + parallel_agents, + strategy="parallel_research", + ) await _emit_progress( on_progress, phase="selection", @@ -625,7 +799,12 @@ async def build_workflow_proposals( "summary": f"由选定角色依次复核“{preview}”，逐步补足证据与结论。", "rationale": "适合需要角色接力、审阅和逐步收敛的复杂任务。", "estimated_duration_seconds": 180, - "graph": _collaborative_graph(source_graph, collaborative_agents), + "graph": _collaborative_graph( + source_graph, + collaborative_agents, + task_contracts=collaborative_contracts, + model_name=model_name, + ), "canvas_schema": "{}", }, } @@ -635,7 +814,14 @@ async def build_workflow_proposals( "summary": f"由 {_agent_name(quick_agents[0])} 先处理“{preview}”，快速给出结论与信息缺口。", "rationale": "适合先校验方向、范围明确或需要控制成本的任务。", "estimated_duration_seconds": 45, - "graph": _single_agent_graph(source_graph, quick_agents[0], name="快速验证方案", strategy="quick_answer"), + "graph": _single_agent_graph( + source_graph, + quick_agents[0], + name="快速验证方案", + strategy="quick_answer", + task_contract=quick_contracts.get(_agent_id(quick_agents[0])), + model_name=model_name, + ), "canvas_schema": "{}", } if len(parallel_agents) >= 2: @@ -645,7 +831,12 @@ async def build_workflow_proposals( "summary": f"多个角色并行收集“{preview}”相关依据，再由综合角色输出统一结论和报告。", "rationale": "适合研究、诊断、方案和报告类任务，可减少单一视角遗漏。", "estimated_duration_seconds": 600, - "graph": _parallel_research_graph(source_graph, parallel_agents), + "graph": _parallel_research_graph( + source_graph, + parallel_agents, + task_contracts=parallel_contracts, + model_name=model_name, + ), "canvas_schema": "{}", } diff --git a/offline-backend-20260512/backend/config.yaml b/offline-backend-20260512/backend/config.yaml index 0a439a8c..e91b2558 100644 --- a/offline-backend-20260512/backend/config.yaml +++ b/offline-backend-20260512/backend/config.yaml @@ -33,6 +33,18 @@ concurrency: fail_open: true models: +- name: MiniMax-M2.7 + display_name: MiniMax-M2.7 + use: langchain_openai:ChatOpenAI + model: MiniMax-M2.7 + api_key: sk-d7OGnT65sFJrupGsxdYGHunrCS3yORr5fwMfxWctKPICuPAT + base_url: https://api.sfkey.cn/v1 + request_timeout: 30.0 + max_retries: 2 + stream_usage: true + supports_thinking: false + supports_reasoning_effort: false + supports_vision: false - name: deepseek-v4-flash display_name: deepseek-v4-flash use: langchain_openai:ChatOpenAI @@ -293,8 +305,15 @@ tool_groups: - name: file:read - name: file:write - name: bash + - name: page:report tools: + # 页面研报设计师的模板选择、工作副本准备和静态质量校验工具。 + # 模板源文件不会被改写；每次生成均复制到当前会话沙箱。 + - name: page_report_template + group: page:report + use: deerflow.tools.builtins.page_report_template_tool:page_report_template_tool + - name: web_search group: web # DuckDuckGo search — no API key required. To switch to the deployment's diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/collection/thread_harvester.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/collection/thread_harvester.py index 2d87627a..9dd27d3c 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/collection/thread_harvester.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/collection/thread_harvester.py @@ -45,6 +45,10 @@ logger = logging.getLogger(__name__) # Upper bound on harvested materials so a chatty collector thread cannot blow # up the report context; the curating/compressing stages trim further anyway. MAX_HARVESTED_MATERIALS = 80 +# A collector model may time out after tool calls were checkpointed but before +# its final assistant response advanced the thread head. Keep the scan bounded +# while retaining the richest checkpoint from that short history. +MAX_CHECKPOINT_HISTORY = 32 # Tool-name → ResearchMaterial.source_type. Everything unknown maps to # ``other``; ``bash`` is treated as ``skill`` because skill scripts emit there. @@ -72,11 +76,43 @@ async def harvest_thread_materials( visible text (its collection小结) and doubles as a fallback material when the thread produced no harvestable tool results. """ - checkpoint_tuple = await checkpointer.aget_tuple({"configurable": {"thread_id": runtime_thread_id, "checkpoint_ns": ""}}) + config = {"configurable": {"thread_id": runtime_thread_id, "checkpoint_ns": ""}} + checkpoint_tuple = await checkpointer.aget_tuple(config) if checkpoint_tuple is None: logger.warning("deep-research harvest: no checkpoint for thread %s", runtime_thread_id) return [], "" - values = checkpoint_tuple.checkpoint.get("channel_values", {}) or {} + + best_materials, best_summary = _harvest_checkpoint_tuple(checkpoint_tuple, query=query) + + # ``aget_tuple`` normally returns the latest state. When a streamed lead + # agent call fails, however, some checkpointers leave that head at the + # final input checkpoint even though an earlier checkpoint already holds + # the successful web-search tool results. Prefer the richest available + # material pool rather than treating that transient head as an empty run. + checkpoint_history = getattr(checkpointer, "alist", None) + if callable(checkpoint_history): + try: + history = checkpoint_history(config, limit=MAX_CHECKPOINT_HISTORY) + if hasattr(history, "__aiter__"): + async for candidate in history: + materials, summary = _harvest_checkpoint_tuple(candidate, query=query) + if len(materials) > len(best_materials): + best_materials, best_summary = materials, summary + else: + for candidate in history: + materials, summary = _harvest_checkpoint_tuple(candidate, query=query) + if len(materials) > len(best_materials): + best_materials, best_summary = materials, summary + except Exception: # noqa: BLE001 - newest checkpoint remains a safe fallback + logger.debug("deep-research harvest: checkpoint history unavailable for %s", runtime_thread_id, exc_info=True) + return best_materials, best_summary + + +def _harvest_checkpoint_tuple(checkpoint_tuple: Any, *, query: str) -> tuple[list[ResearchMaterial], str]: + checkpoint = getattr(checkpoint_tuple, "checkpoint", None) + if not isinstance(checkpoint, dict): + return [], "" + values = checkpoint.get("channel_values", {}) or {} messages = values.get("messages") or [] return harvest_messages_materials(messages, query=query) @@ -259,6 +295,7 @@ class StaticHarvestMaterialProvider: __all__ = [ + "MAX_CHECKPOINT_HISTORY", "MAX_HARVESTED_MATERIALS", "StaticHarvestMaterialProvider", "harvest_messages_materials", diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/runners/basic.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/runners/basic.py index 751d7b47..0a37ccfc 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/runners/basic.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/runners/basic.py @@ -120,6 +120,11 @@ class BasicRunner: all_materials: list[ResearchMaterial] = [] plan: dict[str, Any] = {} + # A completed Deep Research request must still produce a useful report + # when every retrieval channel is temporarily unavailable. This flag + # carries that runtime fact into the writer without mutating the + # immutable per-job config snapshot. + use_knowledge_fill = False try: # ── 1. planning: generate sub-queries ────────────────────────── @@ -198,32 +203,20 @@ class BasicRunner: cancellation.raise_if_cancelled() if not all_materials: - knowledge_fill = _allows_knowledge_fill(request.config.report_instruction) + # Collection is best-effort. In particular, a collector run + # can time out after its tool calls have streamed to the user, + # or every configured retrieval backend can be unavailable. + # Neither case should turn a report request into a stub file. + use_knowledge_fill = True await events.emit( "warning", phase="curating", payload={ "code": "no_materials", - "message": ( - "未收集到外部研究材料，改用领域知识撰写" - if knowledge_fill - else "未收集到任何研究材料" - ), - "recoverable": knowledge_fill, + "message": "未收集到外部研究材料，已改用领域知识继续撰写完整报告", + "recoverable": True, }, ) - if not knowledge_fill: - report_md = ( - f"# {request.query}\n\n" - f"未能收集到与研究课题「{request.query}」相关的材料，无法生成报告。" - ) - await adapters.artifacts.write_text("report.md", report_md) - await events.emit( - "report_completed", - phase="writing", - payload={"reportVersion": 1, "sourceCount": 0}, - ) - return DeepResearchResult(report_markdown=report_md, source_ids=[], plan=plan) deduped: list[ResearchMaterial] = [] selected_ids: list[str] = [] context_list: list[dict[str, Any]] = [] @@ -261,7 +254,14 @@ class BasicRunner: if cancellation: cancellation.raise_if_cancelled() - report_md = await self._generate_report(llm, request, context_text, events, cancellation) + report_md = await self._generate_report( + llm, + request, + context_text, + events, + cancellation, + allow_knowledge_fill=use_knowledge_fill, + ) report_md, image_paths = await generate_report_images(request, adapters, report_md, cancellation) # ── 6. write artifacts ───────────────────────────────────────── @@ -384,14 +384,23 @@ class BasicRunner: return cleaned or queries_from_retrieval_directions(query, dirs)[:n] async def _search_one(self, provider, query, config, request, cancellation, events) -> list[ResearchMaterial]: - """Search one sub-query, emitting events. Callers drive the ordering.""" + """Search one sub-query while keeping evidence independent of telemetry. + + A material batch is the report's input, whereas ``source_added`` is + only progress/persistence telemetry. In particular, a source may + contain native ``datetime`` metadata that an older event store cannot + serialise. Do not let that failure discard an already-returned batch. + """ if cancellation: cancellation.raise_if_cancelled() - await events.emit( - "query_started", - phase="collecting", - payload={"query": query}, - ) + try: + await events.emit( + "query_started", + phase="collecting", + payload={"query": query}, + ) + except Exception: # noqa: BLE001 - telemetry must not prevent collection + logger.warning("basic runner: unable to emit query_started", exc_info=True) materials = await provider.search( query, domains=config.allowed_domains or None, @@ -402,27 +411,42 @@ class BasicRunner: ) # Emit source_added for each material. for m in materials: + try: + await events.emit( + "source_added", + phase="collecting", + payload={ + "sourceId": m.id, + "title": m.title, + "url": m.url, + "sourceType": m.source_type, + # Full persistence-ready dict so the executor can store + # sources without re-deriving them from the event. + "source": m.to_persistence_dict(query), + }, + ) + except Exception: # noqa: BLE001 - the batch still belongs in the report + logger.warning("basic runner: unable to emit source_added for %s", m.id, exc_info=True) + try: await events.emit( - "source_added", + "query_completed", phase="collecting", - payload={ - "sourceId": m.id, - "title": m.title, - "url": m.url, - "sourceType": m.source_type, - # Full persistence-ready dict so the executor can store - # sources without re-deriving them from the event. - "source": m.to_persistence_dict(query), - }, + payload={"query": query, "resultCount": len(materials)}, ) - await events.emit( - "query_completed", - phase="collecting", - payload={"query": query, "resultCount": len(materials)}, - ) + except Exception: # noqa: BLE001 - preserve collected materials on telemetry failure + logger.warning("basic runner: unable to emit query_completed", exc_info=True) return materials - async def _generate_report(self, llm, request: DeepResearchRequest, context_text: str, events, cancellation=None) -> str: + async def _generate_report( + self, + llm, + request: DeepResearchRequest, + context_text: str, + events, + cancellation=None, + *, + allow_knowledge_fill: bool = False, + ) -> str: """Generate the final Markdown report from the formatted context.""" outline = request.config.custom_outline filled_preamble = fill_report_preamble( @@ -431,7 +455,7 @@ class BasicRunner: ) outline_block = _report_outline_instruction(outline) requirements_block = _report_requirements_instruction(request.config.report_instruction) - if _allows_knowledge_fill(request.config.report_instruction): + if allow_knowledge_fill or _allows_knowledge_fill(request.config.report_instruction): system_extra = ( "用户已要求在材料不足时用模型自身的公开领域知识写成完整报告。" "禁止把「材料不足/未提供文献」写成章节主线；参考文献可标注为基于公开领域知识。" @@ -595,7 +619,7 @@ def fill_report_preamble( continue if re.match(r"^密级\s*[：:]", stripped): value = re.sub(r"^密级\s*[：:]\s*", "", stripped) - filled.append(f"密级：内部" if _is_placeholder(value) else f"密级：{value.strip() or '内部'}") + filled.append("密级：内部" if _is_placeholder(value) else f"密级：{value.strip() or '内部'}") continue title = _PLACEHOLDER_RE.sub(topic, stripped) if topic else stripped filled.append(title) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py index 6c6f5a81..da2474dc 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py @@ -842,6 +842,16 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig): is_scheduled_run=bool(cfg.get("is_scheduled_run", False)), ) + # 前端可提供当前视觉明暗模式给页面生成类任务。把值放入系统提示而不是 + # 指望模型从 ToolRuntime.context 猜测，确保自定义 SOUL 也能稳定使用。 + presentation_theme = str(cfg.get("presentation_theme") or "").strip().lower() + if presentation_theme in {"dark", "light"}: + system_prompt += ( + "\n\n【当前界面主题】\n" + f"当前页面呈现主题为 `{presentation_theme}`。若本轮生成汇报页面，请优先选择同色系 HTML 模板；" + "除非用户明确要求覆盖该主题。" + ) + system_prompt = _append_orchestration_capability_boundary(system_prompt, agent_id) _log_es_query_routing_registration( diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/deep_research_events/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/deep_research_events/sql.py index d090cf70..145f7024 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/deep_research_events/sql.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/deep_research_events/sql.py @@ -10,6 +10,8 @@ from __future__ import annotations import json import logging +from datetime import date, datetime +from enum import Enum from typing import Any from sqlalchemy import func, select @@ -21,10 +23,27 @@ from deerflow.persistence.deep_research_events.model import DeepResearchEventRow logger = logging.getLogger(__name__) +def _json_default(value: Any) -> Any: + """Make event payloads replayable when source metadata has native dates. + + Search adapters intentionally preserve ``published_at`` as a ``datetime`` + until it reaches the source store. Events carry the same source envelope, + so the append-only log must normalise it instead of silently falling back to + a live-only frame. + """ + if isinstance(value, (datetime, date)): + return value.isoformat() + if isinstance(value, Enum): + return value.value + if hasattr(value, "model_dump"): + return value.model_dump(mode="json") + return str(value) + + def _dumps(value: Any) -> str | None: if value is None: return None - return json.dumps(value, ensure_ascii=False) + return json.dumps(value, ensure_ascii=False, default=_json_default) def _loads(value: Any) -> Any: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/workflow_events/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/workflow_events/sql.py index ffd7406d..fd82f115 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/workflow_events/sql.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/workflow_events/sql.py @@ -14,19 +14,20 @@ from deerflow.persistence.workflow_events.base import WorkflowEventStore from deerflow.persistence.workflow_events.model import WorkflowRunEventRow from deerflow.persistence.workflow_runs.model import WorkflowRunRow -_MAX_PAYLOAD_CHARS = 64_000 _APPEND_ATTEMPTS = 5 def _dump_payload(payload: dict[str, Any] | None) -> str: - text = json.dumps(payload or {}, ensure_ascii=False, separators=(",", ":"), default=str) - if len(text) > _MAX_PAYLOAD_CHARS: - return json.dumps( - {"truncated": True, "preview": text[:_MAX_PAYLOAD_CHARS]}, - ensure_ascii=False, - separators=(",", ":"), - ) - return text + """Serialize the durable event exactly as it was emitted. + + ``node.message`` carries DeerFlow's native LangGraph messages frame. It + contains tool results, clarification metadata, and sometimes large search + responses. Replacing it with a preview makes an SSE reconnect render a + different conversation from the live stream, so the event log must not + crop payloads. The column is ``PortableLongText`` specifically for this + kind of replayable payload. + """ + return json.dumps(payload or {}, ensure_ascii=False, separators=(",", ":"), default=str) def event_to_dict(row: WorkflowRunEventRow) -> dict[str, Any]: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/workflows/events.py b/offline-backend-20260512/backend/packages/harness/deerflow/workflows/events.py index 7974da8f..6e7405fe 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/workflows/events.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/workflows/events.py @@ -9,6 +9,7 @@ Canonical ``data`` payloads (frozen contract — REST replay and SSE must agree) "title": "联网检索", "args": {"path": "..."}} node.tool.finished {"name": "web_search", "toolCallId": "call_001", "status": "succeeded" | "failed", "summary": "获得 12 条结果"} + node.message {"messages": [, ]} run.awaiting_input {"resumeToken": "one-time-token", "prompt": "请确认是否发布", "formSchema": {...}, "actions": ["approve", "reject"]} artifact.created {"artifactId": "artifact_001", "name": "weekly-report.docx", @@ -40,6 +41,7 @@ WorkflowEventType = Literal[ "node.started", "node.progress", "node.output.delta", + "node.message", "node.tool.started", "node.tool.finished", "artifact.created", @@ -61,6 +63,7 @@ ALL_WORKFLOW_EVENT_TYPES: tuple[WorkflowEventType, ...] = ( "node.started", "node.progress", "node.output.delta", + "node.message", "node.tool.started", "node.tool.finished", "artifact.created", diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/workflows/nodes/agents.py b/offline-backend-20260512/backend/packages/harness/deerflow/workflows/nodes/agents.py index 2393dfd1..93b58aaf 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/workflows/nodes/agents.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/workflows/nodes/agents.py @@ -17,7 +17,7 @@ from typing import Any from deerflow.workflows.errors import WorkflowError from deerflow.workflows.expressions import render_template, render_value from deerflow.workflows.nodes.base import require_dep -from deerflow.workflows.runtime.context import RunContext +from deerflow.workflows.runtime.context import RunContext, WorkflowPaused from deerflow.workflows.schemas import ArtifactRef, NodeResult, WorkflowNode logger = logging.getLogger(__name__) @@ -96,28 +96,6 @@ def _extract_json(text: str) -> Any: return None -_SAFE_ARG_STRING_CHARS = 160 -_SAFE_ARG_MAX_KEYS = 12 - - -def _safe_tool_args(args: Any) -> dict[str, Any]: - """Minimal safe projection of tool-call args for the event log. - - Scalars only (strings truncated), no nested containers, bounded key count — - enough for the canvas to label the step without leaking full payloads such - as entire documents, prompts or credentials that a tool may carry.""" - if not isinstance(args, dict): - return {} - out: dict[str, Any] = {} - for key, value in list(args.items())[:_SAFE_ARG_MAX_KEYS]: - if isinstance(value, bool) or not isinstance(value, (str, int, float)): - continue - if isinstance(value, str): - value = value[:_SAFE_ARG_STRING_CHARS] - out[str(key)[:64]] = value - return out - - def _append_input_bindings(prompt: str, raw_bindings: Any, state: dict[str, Any]) -> str: """Attach explicit workflow inputs to an agent prompt. @@ -206,7 +184,10 @@ class _AgentInvocation: "name": name, "toolCallId": call_id, "title": str(event.get("title") or name), - "args": _safe_tool_args(event.get("args")), + # The native messages frame is authoritative for the + # conversation UI. Preserve this compatibility event + # too, so history/replay never differs from live data. + "args": event.get("args"), }, node_id=node.id, ) @@ -218,13 +199,30 @@ class _AgentInvocation: "name": name, "toolCallId": call_id, "status": "succeeded" if ok in (True, None) else "failed", - "summary": event.get("summary"), + "result": event.get("result"), }, node_id=node.id, ) + async def on_message(messages: list[Any]) -> None: + # Do not adapt the nested LangGraph frame. The browser consumes + # exactly the same native AIMessage/ToolMessage structure as + # DeerFlow's chat client; the outer workflow event only supplies + # durable seq and canvas-node attribution. + await ctx.emit_event( + "node.message", + data={"messages": messages}, + node_id=node.id, + ) + try: try: + resume_payload = ( + ctx.resume_payload + if isinstance(ctx.resume_payload, dict) + and str(ctx.resume_payload.get("nodeId") or "") == node.id + else None + ) outcome = await run_agent( run_id=ctx.run_id, node_id=node.id, @@ -237,6 +235,8 @@ class _AgentInvocation: cancel=ctx.cancel, on_delta=on_delta, on_tool=on_tool, + on_message=on_message, + resume_payload=resume_payload, ) except WorkflowError as exc: exc.node_id = exc.node_id or node.id @@ -253,6 +253,20 @@ class _AgentInvocation: finally: await coalescer.flush() + pending_input = outcome.get("awaiting_input") + if isinstance(pending_input, dict): + raise WorkflowPaused( + node.id, + { + "nodeId": node.id, + **pending_input, + }, + ) + if resume_payload is not None: + # The reply belongs to this agent thread only. A later human/agent + # node must not accidentally consume it as its own resume data. + ctx.resume_payload = None + text = str(outcome.get("text") or "") data: dict[str, Any] = {"text": text} if response_mode == "json": diff --git a/offline-backend-20260512/backend/packages/harness/pyproject.toml b/offline-backend-20260512/backend/packages/harness/pyproject.toml index 3325f550..f7208523 100644 --- a/offline-backend-20260512/backend/packages/harness/pyproject.toml +++ b/offline-backend-20260512/backend/packages/harness/pyproject.toml @@ -65,4 +65,7 @@ packages = ["deerflow"] # 内置 HTML 模板(no.1/no.2 等,非 .py 资源)显式纳入打包,保证任何构建方式(wheel/sdist) # 都带上,新机器装好即可直接看到这些模板。 [tool.hatch.build] -artifacts = ["deerflow/runtime/scheduler/templates/**/*.html"] +artifacts = [ + "deerflow/runtime/scheduler/templates/**/*.html", + "deerflow/page_reports/assets/**/*.html", +] diff --git a/offline-backend-20260512/backend/tests/test_deep_research_persistence.py b/offline-backend-20260512/backend/tests/test_deep_research_persistence.py index 203b4e3d..2b106810 100644 --- a/offline-backend-20260512/backend/tests/test_deep_research_persistence.py +++ b/offline-backend-20260512/backend/tests/test_deep_research_persistence.py @@ -454,6 +454,25 @@ async def test_event_payload_round_trip(): await repos["_engine"].dispose() +@pytest.mark.asyncio +async def test_event_payload_serializes_native_source_datetime(): + """Source-added events share the source model's native ``published_at``.""" + repos = await _make_repos() + try: + repo = repos["events"] + published_at = datetime(2026, 9, 2, 10, 9, tzinfo=UTC) + await repo.append( + session_id="drs_1", + job_id="drj_1", + event_type="source_added", + payload={"source": {"id": "src_1", "published_at": published_at}}, + ) + events = await repo.list_after("drj_1", after=0) + assert events[0]["payload"]["source"]["published_at"] == published_at.isoformat() + finally: + await repos["_engine"].dispose() + + @pytest.mark.asyncio async def test_event_seq_independent_per_job(): repos = await _make_repos() diff --git a/offline-backend-20260512/backend/tests/test_workflow_concurrency.py b/offline-backend-20260512/backend/tests/test_workflow_concurrency.py index 16d291cb..d9cedf3c 100644 --- a/offline-backend-20260512/backend/tests/test_workflow_concurrency.py +++ b/offline-backend-20260512/backend/tests/test_workflow_concurrency.py @@ -322,6 +322,34 @@ async def test_sql_append_without_run_row_falls_back_to_max_seq(session_factory) assert (first["seq"], second["seq"]) == (1, 2) +@pytest.mark.asyncio +async def test_sql_event_replay_keeps_full_native_message_frame(session_factory) -> None: + """A reconnect must receive the same large DeerFlow tool frame as live SSE.""" + events = SqlWorkflowEventStore(session_factory) + run_id = f"run-{uuid.uuid4().hex[:8]}" + payload = { + "messages": [ + { + "type": "tool", + "name": "web_search", + "content": {"results": [{"snippet": "x" * 90_000}]}, + }, + {"langgraph_node": "research"}, + ] + } + + await events.append( + run_id=run_id, + workflow_id="wf-1", + version_id="v-1", + event_type="node.message", + payload=payload, + ) + + replayed = await events.list_after(run_id) + assert replayed[0]["payload"] == payload + + # ── sink: persist-then-publish, never ghost events ────────────────────── diff --git a/offline-backend-20260512/backend/tests/test_workflow_definitions.py b/offline-backend-20260512/backend/tests/test_workflow_definitions.py index 5bb44ee3..69b9c955 100644 --- a/offline-backend-20260512/backend/tests/test_workflow_definitions.py +++ b/offline-backend-20260512/backend/tests/test_workflow_definitions.py @@ -170,13 +170,17 @@ def test_workflow_http_api_and_coze_compat(monkeypatch: pytest.MonkeyPatch) -> N assert published["version_number"] == 1 types = client.get("/api/workflows/node-types").json() - assert len(types["nodeTypes"]) == 13 + # The built-in catalog may grow while preserving the minimum v1 nodes. + assert len(types["nodeTypes"]) >= 13 canvas = client.post("/api/workflow_api/canvas", json={"workflowId": wf_id}).json() assert canvas["code"] == 0 workflow_brief = canvas["data"]["workflow"] assert workflow_brief["workflow_id"] == wf_id assert workflow_brief["is_published"] is True + # Coze Playground uses this alongside VCS metadata to decide whether nodes + # are draggable. An owned standalone workflow must not open as preview. + assert workflow_brief["creator"]["self"] is True # Never opened in the Coze editor → built-in start→end canvas fallback. assert json.loads(workflow_brief["schema_json"])["nodes"] diff --git a/offline-backend-20260512/backend/tests/test_workflow_engine.py b/offline-backend-20260512/backend/tests/test_workflow_engine.py index 7f2ca43d..b6945838 100644 --- a/offline-backend-20260512/backend/tests/test_workflow_engine.py +++ b/offline-backend-20260512/backend/tests/test_workflow_engine.py @@ -663,6 +663,60 @@ async def test_agent_node_streams_deltas_and_parses_json() -> None: } +@pytest.mark.asyncio +async def test_agent_clarification_pauses_and_resume_is_routed_to_the_same_node() -> None: + seen_resume_payloads: list[dict[str, Any] | None] = [] + + async def run_agent(**kwargs: Any) -> dict[str, Any]: + resume_payload = kwargs.get("resume_payload") + seen_resume_payloads.append(resume_payload) + if resume_payload is None: + return { + "awaiting_input": { + "toolCallId": "clarify-1", + "prompt": "请确认分析的地域范围。", + "actions": ["submit"], + } + } + assert resume_payload["values"] == {"answer": "中国市场"} + return {"text": "已按中国市场范围完成分析。"} + + graph = _graph( + [ + {"id": "start", "type": "start"}, + { + "id": "research", + "type": "agent", + "config": {"agentId": "researcher", "promptTemplate": "分析 {{ inputs.topic }}"}, + }, + {"id": "end", "type": "output", "config": {"mapping": {"text": "{{ nodes.research.data.text }}"}}}, + ], + [ + {"id": "e1", "source": "start", "target": "research"}, + {"id": "e2", "source": "research", "target": "end"}, + ], + ) + deps = WorkflowRuntimeDeps(run_agent=run_agent) + + with pytest.raises(WorkflowPaused) as paused: + await _run(graph, {"topic": "新能源车"}, deps=deps) + assert paused.value.node_id == "research" + assert paused.value.pending_input["toolCallId"] == "clarify-1" + + result, _events = await _run( + graph, + {"topic": "新能源车"}, + deps=deps, + resume_payload={ + "nodeId": "research", + "values": {"answer": "中国市场"}, + "message": "只分析中国市场", + }, + ) + assert result.output == {"text": "已按中国市场范围完成分析。"} + assert seen_resume_payloads == [None, {"nodeId": "research", "values": {"answer": "中国市场"}, "message": "只分析中国市场"}] + + @pytest.mark.asyncio async def test_agent_node_includes_rendered_workflow_input_bindings() -> None: seen: dict[str, Any] = {} diff --git a/offline-backend-20260512/backend/tests/test_workflow_runs.py b/offline-backend-20260512/backend/tests/test_workflow_runs.py index 7c2c25c1..d420a71b 100644 --- a/offline-backend-20260512/backend/tests/test_workflow_runs.py +++ b/offline-backend-20260512/backend/tests/test_workflow_runs.py @@ -219,6 +219,10 @@ async def harness(monkeypatch: pytest.MonkeyPatch): class _Cfg: workflows = config + @staticmethod + def get_model_config(name: str): + return {"name": name} if name == "planner-model" else None + async def _user(_request=None) -> str: return "user-a" @@ -325,7 +329,7 @@ async def test_planning_session_selects_an_editable_candidate_then_confirms_one_ created = await harness.client.post( f"/api/workflows/{definition['id']}/planning-sessions", - json={"query": "分析本周销售下滑并给出改进方案"}, + json={"query": "分析本周销售下滑并给出改进方案", "modelName": "planner-model"}, ) assert created.status_code == 200, created.text session = created.json() @@ -337,6 +341,10 @@ async def test_planning_session_selects_an_editable_candidate_then_confirms_one_ ] assert session["proposals"][0]["title"] == "多角度调研方案" assert all(proposal["validation"]["valid"] for proposal in session["proposals"]) + for proposal in session["proposals"]: + for node in proposal["graph"]["nodes"]: + if node["type"] == "agent": + assert node["config"]["modelName"] == "planner-model" parallel_graph = session["proposals"][0]["graph"] evidence = next(node for node in parallel_graph["nodes"] if node["type"] == "evidence_normalizer") synthesis = next(node for node in parallel_graph["nodes"] if node["id"] == "writer") @@ -453,7 +461,7 @@ async def test_dedicated_workflow_controller_has_no_tools_or_thinking( response = await harness.client.post( f"/api/workflows/{definition['id']}/planning-sessions", - json={"query": "快速确认这份材料是否有缺口"}, + json={"query": "快速确认这份材料是否有缺口", "modelName": "planner-model"}, ) assert response.status_code == 200, response.text @@ -461,6 +469,7 @@ async def test_dedicated_workflow_controller_has_no_tools_or_thinking( assert captured["disable_tools"] is True assert captured["thinking_enabled"] is False assert captured["force_disable_thinking"] is True + assert captured["model_name"] == "planner-model" @pytest.mark.asyncio @@ -505,6 +514,22 @@ async def test_planner_uses_visible_agent_catalog_when_canvas_is_only_start_and_ "collaborative": ["agent-research", "agent-writer"], "quick_answer": ["agent-research"], }, + "taskContracts": { + "parallel_research": { + "agent-research": { + "mission": "只收集新能源乘用车市场规模与增长证据。", + "deliverable": "带来源和置信度的市场规模证据笔记。", + "scope": "不要撰写最终报告。", + "handoff": "交给策略分析师的 Evidence Pack。", + }, + "agent-writer": { + "mission": "比较 Evidence Pack 中的证据并形成判断。", + "deliverable": "冲突、假设和落地建议。", + "scope": "区分事实与推断。", + "handoff": "交给深度研究报告写作节点。", + }, + } + }, "titles": {"parallel_research": "双角色调研报告"}, } @@ -551,6 +576,13 @@ async def test_planner_uses_visible_agent_catalog_when_canvas_is_only_start_and_ "agent-research", "agent-writer", ] + graph_nodes = {node["id"]: node for node in parallel["graph"]["nodes"]} + research_prompt = graph_nodes["__planner_agent_1__"]["config"]["promptTemplate"] + writer_prompt = graph_nodes["__planner_agent_3__"]["config"]["promptTemplate"] + assert "只收集新能源乘用车市场规模与增长证据。" in research_prompt + assert "不要撰写最终报告。" in research_prompt + assert "比较 Evidence Pack 中的证据并形成判断。" in writer_prompt + assert "这是工作流总控为你分配的固定岗位" in writer_prompt assert all(proposal["validation"]["valid"] for proposal in session["proposals"]) diff --git a/scripts/start-backend.ps1 b/scripts/start-backend.ps1 index 07a11ce6..f18c2f13 100644 --- a/scripts/start-backend.ps1 +++ b/scripts/start-backend.ps1 @@ -27,7 +27,7 @@ Set-Location -LiteralPath $(ConvertTo-PSLiteral $BackendDir) `$env:GATEWAY_CORS_ORIGINS = $(ConvertTo-PSLiteral $GatewayCorsOrigins) `$env:PYTHONPATH = $(ConvertTo-PSLiteral ".;$BackendVendorDir") if (-not `$env:WEKNORA_ADMIN_PASSWORD) { `$env:WEKNORA_ADMIN_PASSWORD = "1qaz@WSX" } -cmd.exe /c "`"$(($BackendPython -replace '"', '\"'))`" -m uvicorn app.gateway.app:app --host $(($BackendHost -replace '"', '\"')) --port $(($BackendPort -replace '"', '\"')) --reload > `"$(($LogFile -replace '"', '\"'))`" 2>&1" +& $(ConvertTo-PSLiteral $BackendPython) -m uvicorn app.gateway.app:app --host $(ConvertTo-PSLiteral $BackendHost) --port $(ConvertTo-PSLiteral $BackendPort) --reload --reload-dir 'app' --reload-dir 'packages' --reload-exclude $(ConvertTo-PSLiteral (Join-Path $BackendDir '.deer-flow')) --reload-exclude $(ConvertTo-PSLiteral $RuntimeDir) *> $(ConvertTo-PSLiteral $LogFile) "@ Set-Content -LiteralPath $RunnerFile -Value $runner -Encoding UTF8 diff --git a/scripts/start-backend.sh b/scripts/start-backend.sh index 1dc108d4..0aea2601 100644 --- a/scripts/start-backend.sh +++ b/scripts/start-backend.sh @@ -55,8 +55,10 @@ setsid bash -c ' --host "$2" \ --port "$3" \ --reload \ - --reload-exclude ".deer-flow/*" \ - --reload-exclude ".runtime/*" + --reload-dir "app" \ + --reload-dir "packages" \ + --reload-exclude "$0/.deer-flow" \ + --reload-exclude "$0/.runtime" ' "${BACKEND_DIR}" "${GATEWAY_CORS_ORIGINS}" "${BACKEND_HOST}" "${BACKEND_PORT}" "${BACKEND_VENDOR_DIR}" \ >"${LOG_FILE}" 2>&1 &