433 lines
19 KiB
Python
433 lines
19 KiB
Python
"""开放问答接口(无需登录)。
|
||
|
||
给外部系统一个最简单的 HTTP 入口来「调用本项目的智能体做问答」:
|
||
|
||
- 可指定 **模型**(``model_name``,对应 ``config.yaml`` 中 ``models[]`` 的 name)。
|
||
- 可指定 **智能体**(``agent_id``,对应 ``.deer-flow/agents/<id>/``;留空 = 默认 lead agent)。
|
||
- 可关闭 **大模型思考模式**(``thinking_enabled=false``,默认即关闭)。内网 vLLM/Qwen
|
||
模型仅靠 ``thinking_enabled=False`` 往往关不掉思考(其 config 只声明
|
||
``supports_thinking: true``),本接口在关闭思考时会把
|
||
``thinking_force_disabled`` 一并下发,经
|
||
``create_chat_model(force_disable_thinking=True)`` 强制注入
|
||
``extra_body.chat_template_kwargs.enable_thinking=false``(及 GLM 的
|
||
``thinking.type=disabled``)——与「生成标题」时关闭思考的做法一致。
|
||
|
||
该路由前缀 ``/api/open/`` 已在 ``auth_middleware`` 白名单里,**绕过登录鉴权**,
|
||
以 ``"default"`` 用户桶运行,只暴露受控的问答能力(无文件/线程持久化)。
|
||
|
||
请求示例::
|
||
|
||
POST /api/open/chat
|
||
{
|
||
"message": "用一句话介绍一下你自己",
|
||
"model_name": "deepseek-ai/DeepSeek-V4-Flash",
|
||
"agent_id": null,
|
||
"thinking_enabled": false
|
||
}
|
||
|
||
返回::
|
||
|
||
{"answer": "...", "model_name": "...", "agent_id": null,
|
||
"thinking_enabled": false, "thread_id": "open-...."}
|
||
"""
|
||
|
||
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
|
||
from fastapi.responses import StreamingResponse
|
||
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage
|
||
from pydantic import BaseModel, Field
|
||
|
||
from deerflow.agents.lead_agent.agent import make_lead_agent
|
||
from deerflow.config.agents_config import list_custom_agents, validate_agent_id
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/api/open", tags=["open"])
|
||
|
||
_THINK_TAG_RE = re.compile(r"<think>[\s\S]*?</think>", re.IGNORECASE)
|
||
|
||
|
||
class OpenChatMessage(BaseModel):
|
||
"""多轮历史中的一条消息。"""
|
||
|
||
role: Literal["user", "assistant", "system"] = "user"
|
||
content: str
|
||
|
||
|
||
class OpenChatRequest(BaseModel):
|
||
"""开放问答请求体。
|
||
|
||
``message`` 与 ``messages`` 二选一:单轮直接传 ``message``;需要带上下文时传
|
||
``messages``(完整历史,本接口本身无状态、不落库)。
|
||
"""
|
||
|
||
message: str | None = Field(default=None, description="单轮问题文本;与 messages 二选一")
|
||
messages: list[OpenChatMessage] | None = Field(default=None, description="多轮历史;与 message 二选一")
|
||
model_name: str | None = Field(default=None, description="models[] 中的模型名;留空用默认模型")
|
||
agent_name: str | None = Field(default=None, description="智能体名称(中文显示名);留空用默认 lead agent")
|
||
agent_id: str | None = Field(default=None, description="智能体 id;一般不用,优先级高于 agent_name")
|
||
thinking_enabled: bool = Field(default=False, description="是否开启思考;默认 False = 关闭思考(内网模型也会被强制关闭)")
|
||
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):
|
||
answer: str
|
||
model_name: str | None
|
||
agent_name: str | None
|
||
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:
|
||
"""把 agent_id / 中文名称 解析成运行 id。
|
||
|
||
优先级:显式 agent_id > agent_name(按 .deer-flow/agents/*/config.yaml 的 name 匹配)。
|
||
都没传 = None(默认 lead agent)。名称匹配先精确、再忽略大小写/首尾空格。
|
||
"""
|
||
if agent_id and agent_id.strip():
|
||
try:
|
||
return validate_agent_id(agent_id)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=422, detail=f"非法的 agent_id:{exc}") from exc
|
||
|
||
if not agent_name or not agent_name.strip():
|
||
return None
|
||
|
||
target = agent_name.strip()
|
||
agents = list_custom_agents()
|
||
for a in agents:
|
||
if (a.name or "").strip() == target:
|
||
return a.id
|
||
lowered = target.lower()
|
||
for a in agents:
|
||
if (a.name or "").strip().lower() == lowered:
|
||
return a.id
|
||
raise HTTPException(status_code=404, detail=f"未找到名称为「{agent_name}」的智能体。")
|
||
|
||
|
||
def _build_lc_messages(body: OpenChatRequest) -> list[BaseMessage]:
|
||
"""把请求体转成 LangChain 消息列表。"""
|
||
if body.messages:
|
||
out: list[BaseMessage] = []
|
||
for m in body.messages:
|
||
text = (m.content or "").strip()
|
||
if not text:
|
||
continue
|
||
if m.role == "assistant":
|
||
out.append(AIMessage(content=m.content))
|
||
elif m.role == "system":
|
||
out.append(SystemMessage(content=m.content))
|
||
else:
|
||
out.append(HumanMessage(content=m.content))
|
||
if out:
|
||
return out
|
||
if body.message and body.message.strip():
|
||
return [HumanMessage(content=body.message)]
|
||
raise HTTPException(status_code=422, detail="必须提供非空的 'message' 或 'messages'。")
|
||
|
||
|
||
def _extract_text(content: Any) -> str:
|
||
"""从 AIMessage.content(str 或 block 列表)抽取纯文本。"""
|
||
if isinstance(content, str):
|
||
return content
|
||
if isinstance(content, list):
|
||
parts: list[str] = []
|
||
for block in content:
|
||
if isinstance(block, str):
|
||
parts.append(block)
|
||
elif isinstance(block, dict):
|
||
text_value = block.get("text")
|
||
if isinstance(text_value, str):
|
||
parts.append(text_value)
|
||
return "".join(parts) if parts else ""
|
||
return ""
|
||
|
||
|
||
def _extract_answer(result: Any, *, strip_think: bool) -> str:
|
||
"""从 ainvoke 的最终状态里取最后一条 AI 文本。"""
|
||
messages = (result or {}).get("messages") if isinstance(result, dict) else None
|
||
answer = ""
|
||
if isinstance(messages, list):
|
||
for msg in reversed(messages):
|
||
if isinstance(msg, AIMessage):
|
||
text = _extract_text(msg.content).strip()
|
||
if text:
|
||
answer = text
|
||
break
|
||
if strip_think and answer:
|
||
answer = _THINK_TAG_RE.sub("", answer).strip()
|
||
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
|
||
|
||
|
||
def _build_run_context(body: OpenChatRequest) -> tuple[dict[str, Any], str | None, str]:
|
||
"""把请求体翻译成 (config, agent_id, thread_id)。两个端点共用,确保关思考等行为一致。"""
|
||
agent_id = _resolve_agent_id(body.agent_id, body.agent_name)
|
||
thread_id = body.thread_id or f"open-{uuid.uuid4()}"
|
||
configurable: dict[str, Any] = {
|
||
"thread_id": thread_id,
|
||
"model_name": body.model_name,
|
||
"thinking_enabled": body.thinking_enabled,
|
||
# 关闭思考时强制下发,确保内网模型也真的关掉(见模块 docstring)。
|
||
"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
|
||
config: dict[str, Any] = {"configurable": configurable, "recursion_limit": body.recursion_limit}
|
||
return config, agent_id, thread_id
|
||
|
||
|
||
def _sse(event: str | None, data: Any) -> str:
|
||
"""序列化一条 SSE 消息。event=None 时只发 data 行。"""
|
||
payload = json.dumps(data, ensure_ascii=False)
|
||
head = f"event: {event}\n" if event else ""
|
||
return f"{head}data: {payload}\n\n"
|
||
|
||
|
||
@router.post("/chat", response_model=OpenChatResponse)
|
||
async def open_chat(body: OpenChatRequest) -> OpenChatResponse:
|
||
"""指定模型 + 智能体做一次问答(无状态、可关闭思考)。
|
||
|
||
同步阻塞版:整段调研跑完才返回。**长任务(联网调研)请改用 /chat/stream**——
|
||
本端点期间不产生任何下行字节,容易被前置反代的 proxy_read_timeout 掐成 504。
|
||
"""
|
||
lc_messages = _build_lc_messages(body)
|
||
config, agent_id, thread_id = _build_run_context(body)
|
||
|
||
try:
|
||
agent = make_lead_agent(config)
|
||
result = await agent.ainvoke({"messages": lc_messages}, config=config)
|
||
except Exception as exc: # noqa: BLE001 — 开放接口需把内部错误归一成 502
|
||
logger.exception("open_chat 调用智能体失败 (model=%s, agent=%s)", body.model_name, agent_id)
|
||
raise HTTPException(status_code=502, detail=f"智能体调用失败:{exc}") from exc
|
||
|
||
answer = _extract_answer(result, strip_think=body.strip_think and not body.thinking_enabled)
|
||
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,
|
||
model_name=body.model_name,
|
||
agent_name=body.agent_name,
|
||
agent_id=agent_id,
|
||
thinking_enabled=body.thinking_enabled,
|
||
thread_id=thread_id,
|
||
files=files,
|
||
)
|
||
|
||
|
||
async def _open_chat_event_stream(body: OpenChatRequest) -> AsyncIterator[str]:
|
||
"""驱动智能体并产出 SSE 事件流。
|
||
|
||
- 边跑边转发 AI 文本增量(``event: delta``,仅供实时观看,客户端可忽略);
|
||
- 没有真实事件时每 ``_HEARTBEAT_SECONDS`` 发一个心跳注释行保活;
|
||
- 全部跑完后用 ``event: result`` 给出最终(已按需剥离 <think> 的)答案;
|
||
- 出错则 ``event: error``。
|
||
|
||
用「后台生产任务 + 队列 + 超时取队列」的方式把「智能体进度」与「定时心跳」解耦,
|
||
所以即便某次工具调用静默很久,心跳仍照常下行,连接不会被反代判空闲。
|
||
"""
|
||
lc_messages = _build_lc_messages(body)
|
||
config, agent_id, thread_id = _build_run_context(body)
|
||
strip_think = body.strip_think and not body.thinking_enabled
|
||
|
||
queue: asyncio.Queue[tuple[str, Any]] = asyncio.Queue()
|
||
final_state: dict[str, Any] = {}
|
||
|
||
async def produce() -> None:
|
||
nonlocal final_state
|
||
try:
|
||
agent = make_lead_agent(config)
|
||
async for mode, chunk in agent.astream({"messages": lc_messages}, config=config, stream_mode=["values", "messages"]):
|
||
if mode == "values" and isinstance(chunk, dict):
|
||
final_state = chunk
|
||
await queue.put(("progress", {"messages": len(chunk.get("messages") or [])}))
|
||
elif mode == "messages":
|
||
msg = chunk[0] if isinstance(chunk, tuple) and chunk else chunk
|
||
# 仅转发 AI 正文增量;DeepSeek 等把思考放在 additional_kwargs.reasoning_content,
|
||
# 不在 content 里,所以这里天然不会泄漏 <think>。
|
||
if isinstance(msg, AIMessage):
|
||
text = _extract_text(getattr(msg, "content", "")).strip()
|
||
if text:
|
||
await queue.put(("delta", {"text": text}))
|
||
except Exception as exc: # noqa: BLE001 — 开放接口需把内部错误归一成 error 事件
|
||
logger.exception("open_chat_stream 调用智能体失败 (model=%s, agent=%s)", body.model_name, agent_id)
|
||
await queue.put(("error", {"detail": f"智能体调用失败:{exc}"}))
|
||
finally:
|
||
await queue.put(("__end__", None))
|
||
|
||
task = asyncio.create_task(produce())
|
||
try:
|
||
while True:
|
||
try:
|
||
event, data = await asyncio.wait_for(queue.get(), timeout=_HEARTBEAT_SECONDS)
|
||
except asyncio.TimeoutError:
|
||
yield ": keepalive\n\n" # SSE 注释行,客户端忽略,仅用于保活
|
||
continue
|
||
if event == "__end__":
|
||
break
|
||
if event == "error":
|
||
yield _sse("error", data)
|
||
return
|
||
yield _sse(event, data)
|
||
|
||
answer = _extract_answer(final_state, strip_think=strip_think)
|
||
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",
|
||
{
|
||
"answer": answer,
|
||
"model_name": body.model_name,
|
||
"agent_name": body.agent_name,
|
||
"agent_id": agent_id,
|
||
"thinking_enabled": body.thinking_enabled,
|
||
"thread_id": thread_id,
|
||
"files": [f.model_dump() for f in files],
|
||
},
|
||
)
|
||
finally:
|
||
if not task.done():
|
||
task.cancel()
|
||
|
||
|
||
@router.post("/chat/stream")
|
||
async def open_chat_stream(body: OpenChatRequest) -> StreamingResponse:
|
||
"""流式版 /chat:边跑边发心跳保活,跑完用 ``event: result`` 给最终答案。
|
||
|
||
解决两件事:(1) 长任务(联网调研、报告)同步阻塞会被前置反代 proxy_read_timeout
|
||
掐成 504;流式持续有字节下行,代理不再判定空闲超时。(2) 客户端只需读到
|
||
``event: result`` 解析 ``answer`` 即可,中间的 ``delta``/``progress``/心跳都可忽略。
|
||
"""
|
||
# 入参校验放在流外:非法请求直接抛 4xx,而不是开了流再报错。
|
||
_build_lc_messages(body)
|
||
return StreamingResponse(
|
||
_open_chat_event_stream(body),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"Connection": "keep-alive",
|
||
# 关掉 nginx 对该响应的缓冲,确保 SSE/心跳实时下行(即便前置 nginx 默认开缓冲)。
|
||
"X-Accel-Buffering": "no",
|
||
},
|
||
)
|
||
|
||
|
||
@router.get("/agents")
|
||
async def open_list_agents() -> dict[str, Any]:
|
||
"""列出可用智能体的名称(方便知道 agent_name 该填什么)。"""
|
||
agents = list_custom_agents()
|
||
items = [{"name": a.name, "id": a.id} for a in agents if (a.name or "").strip()]
|
||
return {"agents": items, "count": len(items)}
|