feat(open-chat): 开放问答接口回传生成文件 + 临时不留存

- /api/open/chat[/stream] 新增 include_files(默认开),把智能体在 outputs 目录
  生成的文件(name/virtual_path/encoding(text|base64)/content)随结果回传,
  解决"报告写进文件、answer 只回一段总结"导致拿不到完整内容的问题;
  既无 answer 也无文件时返回 502。
- 无状态运行标记 ephemeral:lead agent 跳过记忆与知识库中间件(不召回/不留存/
  不自动沉淀),取完文件后清理临时线程目录,做到"通过本接口的聊天记录不存储"。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
吉鑫升andClaude Opus 4.8 committed 2026-06-22 19:36:30 +08:00
1 parent 06b169d7a1
commit 7c55a577ad
3 files changed
+111 -9

No files matched your search

+1 -1
View File
@@ -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`. |
@@ -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="关闭思考时,从答案里剥离残留的 <think>…</think> 块")
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/<thread_id>/user-data/outputs;只删到 <thread_id> 这层
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:
@@ -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))