diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md index e47e3ca8..6fa38cdd 100644 --- a/offline-backend-20260512/backend/CLAUDE.md +++ b/offline-backend-20260512/backend/CLAUDE.md @@ -243,7 +243,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA | **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/messages` - paginated messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /../messages` - thread messages with feedback; `GET /../token-usage` - aggregate tokens | | **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific | | **Runs** (`/api/runs`) | `POST /stream` - stateless run + SSE; `POST /wait` - stateless run + block; `GET /{rid}/messages` - paginated messages by run_id `{data, has_more}` (cursor: `after_seq`/`before_seq`); `GET /{rid}/feedback` - list feedback by run_id | -| **Open Chat** (`/api/open/chat`) | **Unauthenticated** open Q&A entry for external systems (the `/api/open/` prefix is whitelisted in `auth_middleware._PUBLIC_PATH_PREFIXES` + CSRF-exempt in `csrf_middleware.should_check_csrf`). `POST /chat` `{message \| messages[], model_name?, agent_name?, agent_id?, thinking_enabled=false, strip_think=true, thread_id?, recursion_limit=100}` → resolves the agent by **Chinese display name** (`agent_name`, matched against each `.deer-flow/agents/*/config.yaml` `name` via `_resolve_agent_id`; explicit `agent_id` wins if given), builds the lead agent via `make_lead_agent` (so the chosen **agent's** SOUL/tools/skills all apply), `ainvoke`s it once (stateless — no checkpointer, no thread persistence; runs under the `"default"` user bucket), and returns **only the final** `{answer, model_name, agent_name, agent_id, thinking_enabled, thread_id}` (no intermediate tool-call/process). **`POST /chat/stream`** is the **SSE streaming variant** (same request body) for **long tasks** — the blocking `/chat` produces no downstream bytes for the whole run, so a front nginx `proxy_read_timeout` (default 60s) cuts it off with a **504**; the stream keeps the connection warm (a `: keepalive` SSE comment every `_HEARTBEAT_SECONDS=15`s via a background producer-task + queue, plus `event: delta` AI-text increments and `event: progress`), and ends with **`event: result`** carrying the same final `{answer, …}` JSON (or `event: error`). Clients only need to read `event: result`; intermediate events are ignorable. Both endpoints share `_build_run_context` (config/agent/thread). `GET /agents` lists `{name, id}` so callers know what `agent_name` to send. **Thinking off is reliable**: when `thinking_enabled=false` the route also sets `thinking_force_disabled=true` so the lead agent calls `create_chat_model(force_disable_thinking=True)`, which injects **three** OpenAI-compatible "thinking off" shapes into `extra_body` — top-level `enable_thinking=false` (**SiliconFlow / DashScope hybrid models** like `deepseek-ai/DeepSeek-V4-*` on `api.siliconflow.cn` — the knob the nested `chat_template_kwargs` shape does NOT reach), `chat_template_kwargs.enable_thinking=false` (vLLM/Qwen native), and `thinking.type="disabled"` (Zhipu GLM) — even for models that only declare `supports_thinking: true`. For the **web-UI path** (which sends `thinking_enabled=false` but NOT `force_disable`), SiliconFlow models need `when_thinking_disabled: {extra_body: {enable_thinking: false}}` in their `config.yaml` entry (added to `deepseek-ai/DeepSeek-V4-Flash`) for thinking-off to take effect. Sets `is_scheduled_run=true` internally to skip the auto-title LLM call. Router `app/gateway/routers/open_chat.py`; standalone caller `test_open_chat_api.py` (repo root, streaming by default); tests `tests/test_open_chat.py`. | +| **Open Chat** (`/api/open/chat`) | **Unauthenticated** open Q&A entry for external systems (the `/api/open/` prefix is whitelisted in `auth_middleware._PUBLIC_PATH_PREFIXES` + CSRF-exempt in `csrf_middleware.should_check_csrf`). `POST /chat` `{message \| messages[], model_name?, agent_name?, agent_id?, thinking_enabled=false, strip_think=true, thread_id?, recursion_limit=100}` → resolves the agent by **Chinese display name** (`agent_name`, matched against each `.deer-flow/agents/*/config.yaml` `name` via `_resolve_agent_id`; explicit `agent_id` wins if given), builds the lead agent via `make_lead_agent` (so the chosen **agent's** SOUL/tools/skills all apply), `ainvoke`s it once (stateless — no checkpointer, no thread persistence; runs under the `"default"` user bucket), and returns **only the final** `{answer, model_name, agent_name, agent_id, thinking_enabled, thread_id}` (no intermediate tool-call/process). **`POST /chat/stream`** is the **SSE streaming variant** (same request body) for **long tasks** — the blocking `/chat` produces no downstream bytes for the whole run, so a front nginx `proxy_read_timeout` (default 60s) cuts it off with a **504**; the stream keeps the connection warm (a `: keepalive` SSE comment every `_HEARTBEAT_SECONDS=15`s via a background producer-task + queue, plus `event: delta` AI-text increments and `event: progress`), and ends with **`event: result`** carrying the same final `{answer, …}` JSON (or `event: error`). Clients only need to read `event: result`; intermediate events are ignorable. Both endpoints share `_build_run_context` (config/agent/thread). `GET /agents` lists `{name, id}` so callers know what `agent_name` to send. **Thinking off is reliable**: when `thinking_enabled=false` the route also sets `thinking_force_disabled=true` so the lead agent calls `create_chat_model(force_disable_thinking=True)`, which injects **three** OpenAI-compatible "thinking off" shapes into `extra_body` — top-level `enable_thinking=false` (**SiliconFlow / DashScope hybrid models** like `deepseek-ai/DeepSeek-V4-*` on `api.siliconflow.cn` — the knob the nested `chat_template_kwargs` shape does NOT reach), `chat_template_kwargs.enable_thinking=false` (vLLM/Qwen native), and `thinking.type="disabled"` (Zhipu GLM) — even for models that only declare `supports_thinking: true`. For the **web-UI path** (which sends `thinking_enabled=false` but NOT `force_disable`), SiliconFlow models need `when_thinking_disabled: {extra_body: {enable_thinking: false}}` in their `config.yaml` entry (added to `deepseek-ai/DeepSeek-V4-Flash`) for thinking-off to take effect. Sets `is_scheduled_run=true` internally to skip the auto-title LLM call. **Generated-file return (`include_files=true`, default on)**: an agent that produces its report as a file (writes to `/mnt/user-data/outputs/` + `present_files`) returns only a short summary in `answer`; to recover the real deliverable both endpoints now also return `files: [{name, virtual_path, encoding(text|base64), content}]` — `_collect_output_files` scans the run's `state['thread_data']['outputs_path']` (the open thread is stateless so artifacts can't be fetched by thread, but the endpoint runs server-side and the sandbox already wrote the files to disk), text decoded as utf-8 else base64, capped by `_MAX_OUTPUT_FILES=20` / `_MAX_OUTPUT_FILE_BYTES=5MB`. `/chat` puts `files` on the response; `/chat/stream` puts it in the `event: result` payload. A run with no answer **and** no files now 502s ("…文本答案或文件"). Router `app/gateway/routers/open_chat.py`; standalone caller `test_open_chat_api.py` (repo root, streaming by default); tests `tests/test_open_chat.py`. | | **Thread Shares** (`/api/thread-shares`) | Per-conversation sharing with two modes (`thread_shares.mode`). **`import`**: `POST /` `{thread_id, mode}` - create (or reuse) a revocable share code; `GET /{code}/preview` + `POST /{code}/import` - copy that conversation into the caller's account (fresh `thread_id` via `app/gateway/thread_copy.py` + new `threads_meta` row + `user-data` copy; dedup on `metadata.shared_origin`). **`view`**: a public, read-only static page ? see Public Shares below. Both modes: `GET /` - list my codes; `DELETE /{code}` - revoke. | | **Public Shares** (`/api/public/shares`) | **Unauthenticated** (the `/api/public/` prefix is whitelisted in `auth_middleware._PUBLIC_PATH_PREFIXES`). `GET /{code}` - read-only conversation snapshot (title + serialized messages + artifacts) for a `view`-mode share; messages are read straight from the latest checkpoint. `GET /{code}/files/{path}` - serves a file referenced by that conversation (images, Markdown, text, PDF, ?) so the shared page's links are openable; path resolution is confined to the owner's thread directory. Active content (`text/html`, `application/xhtml+xml`, `image/svg+xml`) is forced to download (`_public_share_disposition`); everything else is served inline. Missing/revoked/non-`view` codes all return 404. | | **Roundtable Drafts** (`/api/roundtable-drafts`) | **Strictly per-user** persistence for the roundtable-planning page (replaces the old browser `localStorage`). `GET /` - list drafts (lightweight meta); `POST /` - create; `GET/PUT/DELETE /{id}` - full read / partial update / delete (all owner-scoped). A draft = one full session (Step 1 intent + Step 2 roundtable), step snapshots stored as JSON in `PortableLongText`. Recommendation history bound to a draft: `GET /{id}/recommendations` - list newest-first; `POST /{id}/recommendations` - append one run (objective/status/rationale/picks/candidates). taskId 深链会商记录 are **not** here — they live in the separate `/api/roundtable-task-drafts` store (see below) so personal history never mixes with task records. Backed by `deerflow.persistence.roundtable_drafts` (tables `roundtable_drafts`, `roundtable_recommend_history`); store wired in `deps.py` as `app.state.roundtable_draft_store`. Tests: `tests/test_roundtable_task_drafts.py`. | diff --git a/offline-backend-20260512/backend/app/gateway/routers/open_chat.py b/offline-backend-20260512/backend/app/gateway/routers/open_chat.py index 849e0085..3dad51d2 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/open_chat.py +++ b/offline-backend-20260512/backend/app/gateway/routers/open_chat.py @@ -34,11 +34,14 @@ from __future__ import annotations import asyncio +import base64 import json import logging import re +import shutil import uuid from collections.abc import AsyncIterator +from pathlib import Path from typing import Any, Literal from fastapi import APIRouter, HTTPException @@ -79,6 +82,16 @@ class OpenChatRequest(BaseModel): strip_think: bool = Field(default=True, description="关闭思考时,从答案里剥离残留的 … 块") thread_id: str | None = Field(default=None, description="可选;仅用于文件隔离,留空自动生成") recursion_limit: int = Field(default=100, ge=1, le=300, description="智能体单次问答的最大步数") + include_files: bool = Field(default=True, description="是否随结果回传智能体在 outputs 目录生成的文件(名称+内容)") + + +class OpenChatFile(BaseModel): + """智能体本次运行在 /mnt/user-data/outputs 下生成的一个文件。""" + + name: str # outputs/ 下的相对路径,如 "报告.md" + virtual_path: str # 沙箱虚拟路径,如 "/mnt/user-data/outputs/报告.md" + encoding: Literal["text", "base64"] = "text" # text=可直接用的文本;base64=二进制需解码 + content: str class OpenChatResponse(BaseModel): @@ -88,6 +101,7 @@ class OpenChatResponse(BaseModel): agent_id: str | None thinking_enabled: bool thread_id: str + files: list[OpenChatFile] = Field(default_factory=list, description="本次生成的输出文件(含内容)") def _resolve_agent_id(agent_id: str | None, agent_name: str | None) -> str | None: @@ -171,6 +185,76 @@ def _extract_answer(result: Any, *, strip_think: bool) -> str: return answer +_OUTPUTS_VIRTUAL_PREFIX = "/mnt/user-data/outputs" +_MAX_OUTPUT_FILES = 20 +_MAX_OUTPUT_FILE_BYTES = 5 * 1024 * 1024 # 单文件 5MB 上限,避免把大二进制塞进 JSON 响应 + + +def _collect_output_files(state: Any, *, include: bool) -> list[OpenChatFile]: + """读取本次运行在沙箱 outputs 目录生成的文件,随结果一并回传。 + + 开放问答是无状态的(不持久化线程),客户端没法用 artifacts 路由按线程取件;但本端点 + 就跑在服务端进程里,沙箱把生成的文件落在 ``state['thread_data']['outputs_path']``, + 所以这里直接扫描该目录,把文件内容带回(文本按 utf-8,二进制按 base64)。 + 受 ``_MAX_OUTPUT_FILES`` / ``_MAX_OUTPUT_FILE_BYTES`` 限制。 + """ + if not include or not isinstance(state, dict): + return [] + thread_data = state.get("thread_data") or {} + outputs_path = thread_data.get("outputs_path") + if not outputs_path: + return [] + base = Path(outputs_path) + if not base.is_dir(): + return [] + try: + entries = sorted((p for p in base.rglob("*") if p.is_file()), key=lambda p: p.as_posix()) + except OSError: + return [] + + files: list[OpenChatFile] = [] + for entry in entries: + if len(files) >= _MAX_OUTPUT_FILES: + break + try: + if entry.stat().st_size > _MAX_OUTPUT_FILE_BYTES: + continue + data = entry.read_bytes() + except OSError: + continue + rel = entry.relative_to(base).as_posix() + try: + content = data.decode("utf-8") + encoding: Literal["text", "base64"] = "text" + except UnicodeDecodeError: + content = base64.b64encode(data).decode("ascii") + encoding = "base64" + files.append(OpenChatFile(name=rel, virtual_path=f"{_OUTPUTS_VIRTUAL_PREFIX}/{rel}", encoding=encoding, content=content)) + return files + + +def _cleanup_ephemeral_thread(state: Any, thread_id: str | None) -> None: + """删除本次开放问答的临时线程目录(open-*),落实"聊天记录不存储"。 + + 生成文件已在 _collect_output_files 读进内存(随响应回传),这里把磁盘上整个临时线程 + 目录清掉,避免无状态调用在服务端堆积残留。仅对 ``open-*`` 线程生效,best-effort。 + """ + if not thread_id or not str(thread_id).startswith("open-"): + return + thread_data = state.get("thread_data") if isinstance(state, dict) else None + outputs_path = (thread_data or {}).get("outputs_path") + if not outputs_path: + return + try: + p = Path(outputs_path).resolve() + # outputs_path 形如 .../threads//user-data/outputs;只删到 这层 + thread_root = next((parent for parent in p.parents if parent.name == thread_id), None) + if thread_root is not None and thread_root.exists(): + shutil.rmtree(thread_root, ignore_errors=True) + except Exception as exc: # noqa: BLE001 — 清理失败不影响返回结果 + logger.warning("清理临时线程目录失败 (%s): %s", thread_id, exc) + + # 流式响应每隔这么久没有真实事件就发一个 SSE 注释心跳,持续制造下行字节, # 让前置反向代理(nginx proxy_read_timeout 默认 60s 等)不会把长任务判定成空闲超时。 _HEARTBEAT_SECONDS = 15.0 @@ -188,6 +272,10 @@ def _build_run_context(body: OpenChatRequest) -> tuple[dict[str, Any], str | Non "thinking_force_disabled": not body.thinking_enabled, # 一次性问答:跳过自动起标题(省一次 LLM 调用)、不挂定时任务工具。 "is_scheduled_run": True, + # 临时/无状态运行:跳过记忆与知识库的读写(不召回、不留存、不自动沉淀), + # 做到"通过本接口的聊天记录不存储"(见 _build_middlewares 的 is_ephemeral)。 + "ephemeral": True, + "memory_injection_enabled": False, } if agent_id: configurable["agent_id"] = agent_id @@ -220,8 +308,10 @@ async def open_chat(body: OpenChatRequest) -> OpenChatResponse: raise HTTPException(status_code=502, detail=f"智能体调用失败:{exc}") from exc answer = _extract_answer(result, strip_think=body.strip_think and not body.thinking_enabled) - if not answer: - raise HTTPException(status_code=502, detail="智能体未返回任何文本答案。") + files = _collect_output_files(result, include=body.include_files) + _cleanup_ephemeral_thread(result, thread_id) # 取完文件即清理临时目录,不在服务端留存 + if not answer and not files: + raise HTTPException(status_code=502, detail="智能体未返回任何文本答案或文件。") return OpenChatResponse( answer=answer, @@ -230,6 +320,7 @@ async def open_chat(body: OpenChatRequest) -> OpenChatResponse: agent_id=agent_id, thinking_enabled=body.thinking_enabled, thread_id=thread_id, + files=files, ) @@ -289,8 +380,10 @@ async def _open_chat_event_stream(body: OpenChatRequest) -> AsyncIterator[str]: yield _sse(event, data) answer = _extract_answer(final_state, strip_think=strip_think) - if not answer: - yield _sse("error", {"detail": "智能体未返回任何文本答案。"}) + files = _collect_output_files(final_state, include=body.include_files) + _cleanup_ephemeral_thread(final_state, thread_id) # 取完文件即清理临时目录,不在服务端留存 + if not answer and not files: + yield _sse("error", {"detail": "智能体未返回任何文本答案或文件。"}) return yield _sse( "result", @@ -301,6 +394,7 @@ async def _open_chat_event_stream(body: OpenChatRequest) -> AsyncIterator[str]: "agent_id": agent_id, "thinking_enabled": body.thinking_enabled, "thread_id": thread_id, + "files": [f.model_dump() for f in files], }, ) finally: 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 e2d27c2f..66cf4d88 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 @@ -306,17 +306,25 @@ def _build_middlewares( if not cfg.get("is_scheduled_run", False): middlewares.append(TitleMiddleware(app_config=resolved_app_config)) - # Add MemoryMiddleware (after TitleMiddleware) — skip in bootstrap mode (agent creation flow) + # Ephemeral runs (e.g. the open Q&A API /api/open/chat) must NOT touch any + # server-side per-user state: no memory recall/retain, no knowledge-base + # recall/auto-ingest. Those runs use the shared "default" user bucket, so + # persisting their conversations there would both store records the caller + # explicitly asked NOT to keep and leak content across unrelated callers. + # Skip both write-capable middlewares entirely. + is_ephemeral = cfg.get("ephemeral", False) + + # Add MemoryMiddleware (after TitleMiddleware) — skip in bootstrap mode (agent creation flow) and ephemeral runs memory_injection_enabled = cfg.get("memory_injection_enabled", True) - if not is_bootstrap and not is_writing_setup: + if not is_bootstrap and not is_writing_setup and not is_ephemeral: middlewares.append(MemoryMiddleware(agent_name=agent_name, memory_config=resolved_app_config.memory, injection_enabled=memory_injection_enabled)) # KnowledgeRagMiddleware — inject knowledge-base recall (phase 2). Gated # per-conversation by the ``knowledge_rag_enabled`` runtime flag, defaulting # to the global ``knowledge.rag_enabled`` config. No-op when knowledge is - # disabled. Skipped in bootstrap/writing-setup like MemoryMiddleware. + # disabled. Skipped in bootstrap/writing-setup/ephemeral like MemoryMiddleware. knowledge_cfg = getattr(resolved_app_config, "knowledge", None) - if not is_bootstrap and not is_writing_setup and knowledge_cfg is not None and knowledge_cfg.enabled: + if not is_bootstrap and not is_writing_setup and not is_ephemeral and knowledge_cfg is not None and knowledge_cfg.enabled: from deerflow.agents.middlewares.knowledge_rag_middleware import KnowledgeRagMiddleware middlewares.append(KnowledgeRagMiddleware(limit=knowledge_cfg.rag_limit, default_enabled=knowledge_cfg.rag_enabled))