150 lines
4.9 KiB
Python
150 lines
4.9 KiB
Python
"""Direct LLM chat — no lead agent, tools, skills, or memory.
|
|
|
|
Used by the frontend 「极速」mode: stream tokens from ``create_chat_model``
|
|
with thinking forced off so first-token latency matches a bare chat client.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from collections.abc import AsyncIterator
|
|
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.config import get_app_config
|
|
from deerflow.models import create_chat_model
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/direct-chat", tags=["direct-chat"])
|
|
|
|
_HEARTBEAT_SECONDS = 15.0
|
|
_MAX_MESSAGES = 40
|
|
|
|
|
|
class DirectChatMessage(BaseModel):
|
|
role: Literal["user", "assistant", "system"]
|
|
content: str = ""
|
|
|
|
|
|
class DirectChatRequest(BaseModel):
|
|
messages: list[DirectChatMessage] = Field(..., min_length=1)
|
|
model_name: str | None = None
|
|
|
|
|
|
def _chunk_text(chunk: Any) -> str:
|
|
content = getattr(chunk, "content", None)
|
|
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) and isinstance(block.get("text"), str):
|
|
parts.append(block["text"])
|
|
return "".join(parts)
|
|
return ""
|
|
|
|
|
|
def _build_lc_messages(body: DirectChatRequest) -> list[BaseMessage]:
|
|
built: list[BaseMessage] = []
|
|
for item in body.messages[-_MAX_MESSAGES:]:
|
|
text = (item.content or "").strip()
|
|
if not text:
|
|
continue
|
|
if item.role == "user":
|
|
built.append(HumanMessage(content=text))
|
|
elif item.role == "assistant":
|
|
built.append(AIMessage(content=text))
|
|
else:
|
|
built.append(SystemMessage(content=text))
|
|
if not any(isinstance(msg, HumanMessage) for msg in built):
|
|
raise HTTPException(status_code=400, detail="至少需要一条 user 消息")
|
|
return built
|
|
|
|
|
|
def _sse(event: str, data: Any) -> str:
|
|
return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
|
|
|
|
|
|
def _resolve_model_name(name: str | None) -> str:
|
|
config = get_app_config()
|
|
if not name:
|
|
if not config.models:
|
|
raise HTTPException(status_code=400, detail="未配置可用模型")
|
|
return config.models[0].name
|
|
if config.get_model_config(name) is None:
|
|
raise HTTPException(status_code=400, detail=f"模型不存在: {name}")
|
|
return name
|
|
|
|
|
|
async def _direct_chat_event_stream(body: DirectChatRequest, model_name: str) -> AsyncIterator[str]:
|
|
lc_messages = _build_lc_messages(body)
|
|
queue: asyncio.Queue[tuple[str, Any]] = asyncio.Queue()
|
|
pieces: list[str] = []
|
|
|
|
async def produce() -> None:
|
|
try:
|
|
model = create_chat_model(
|
|
name=model_name,
|
|
thinking_enabled=False,
|
|
force_disable_thinking=True,
|
|
)
|
|
async for chunk in model.astream(lc_messages):
|
|
text = _chunk_text(chunk)
|
|
if text:
|
|
pieces.append(text)
|
|
await queue.put(("delta", {"text": text}))
|
|
except Exception as exc: # noqa: BLE001 — stream errors become SSE error events
|
|
logger.exception("direct_chat 调用模型失败 (model=%s)", model_name)
|
|
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"
|
|
continue
|
|
if event == "__end__":
|
|
break
|
|
if event == "error":
|
|
yield _sse("error", data)
|
|
return
|
|
yield _sse(event, data)
|
|
|
|
answer = "".join(pieces)
|
|
if not answer:
|
|
yield _sse("error", {"detail": "模型未返回任何文本。"})
|
|
return
|
|
yield _sse("result", {"answer": answer, "model_name": model_name})
|
|
finally:
|
|
if not task.done():
|
|
task.cancel()
|
|
|
|
|
|
@router.post("/stream")
|
|
async def direct_chat_stream(body: DirectChatRequest) -> StreamingResponse:
|
|
"""Stream a plain chat completion. No agent graph, tools, or skills."""
|
|
model_name = _resolve_model_name(body.model_name)
|
|
_build_lc_messages(body)
|
|
return StreamingResponse(
|
|
_direct_chat_event_stream(body, model_name),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"Connection": "keep-alive",
|
|
"X-Accel-Buffering": "no",
|
|
},
|
|
)
|