xuhdandClaude Sonnet 4.6
adeace0f7b
添加 multi-llm 路由智能体（LangGraph），支持多 LLM 分发能力；在 Gateway 中注册 /api/multi-llm 路由，并豁免其 CSRF 和认证校验；同时新增 asyncmy 依赖和聊天流格式文档。 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
"""Multi-LLM routing agent — SSE streaming endpoint.
|
|
|
|
POST /api/multi-llm/{thread_id}/stream
|
|
Body: {"message": "..."}
|
|
Returns text/event-stream with frames (all as "data: <json>\\n\\n"):
|
|
{"type": "routing", "system": "B"} ← router decision
|
|
{"type": "thinking", "content": "...", "system": "B"} ← reasoning (if available)
|
|
{"type": "text", "content": "...", "system": "B"} ← final text chunks
|
|
{"type": "done"} ← stream end
|
|
{"type": "error", "message": "..."} ← on failure
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from fastapi.responses import StreamingResponse
|
|
from langchain_core.messages import HumanMessage
|
|
from langgraph.types import Checkpointer
|
|
from pydantic import BaseModel
|
|
|
|
from app.gateway.deps import get_checkpointer
|
|
from deerflow.agents.multi_llm.graph import build_multi_llm_graph
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/multi-llm", tags=["multi-llm"])
|
|
|
|
# Lazy singleton: built once on first request, reusing the app-scoped checkpointer.
|
|
_graph = None
|
|
_graph_lock = asyncio.Lock()
|
|
|
|
|
|
async def _get_graph(checkpointer: Checkpointer):
|
|
global _graph
|
|
if _graph is None:
|
|
async with _graph_lock:
|
|
if _graph is None:
|
|
_graph = build_multi_llm_graph(checkpointer=checkpointer)
|
|
return _graph
|
|
|
|
|
|
class ChatRequest(BaseModel):
|
|
message: str
|
|
|
|
|
|
@router.post("/{thread_id}/stream", summary="Multi-LLM routing agent (SSE stream)")
|
|
async def multi_llm_stream(
|
|
thread_id: str,
|
|
body: ChatRequest,
|
|
request: Request,
|
|
checkpointer: Checkpointer = Depends(get_checkpointer),
|
|
) -> StreamingResponse:
|
|
graph = await _get_graph(checkpointer)
|
|
lg_config = {"configurable": {"thread_id": thread_id}}
|
|
input_state = {"messages": [HumanMessage(content=body.message)]}
|
|
|
|
async def generate():
|
|
try:
|
|
async for event in graph.astream_events(input_state, config=lg_config, version="v2"):
|
|
if await request.is_disconnected():
|
|
break
|
|
|
|
ev = event["event"]
|
|
|
|
if ev == "on_custom_event" and event.get("name") == "llm_chunk":
|
|
d = event["data"]
|
|
yield f"data: {json.dumps({'type': d['type'], 'content': d['content'], 'system': d['system']}, ensure_ascii=False)}\n\n"
|
|
|
|
elif ev == "on_chain_end" and event.get("name") == "router":
|
|
output = event.get("data", {}).get("output", {})
|
|
system = output.get("selected_system", "") if isinstance(output, dict) else ""
|
|
if system:
|
|
yield f"data: {json.dumps({'type': 'routing', 'system': system})}\n\n"
|
|
|
|
yield f"data: {json.dumps({'type': 'done'})}\n\n"
|
|
|
|
except Exception as e:
|
|
logger.exception("multi_llm stream failed: thread=%s", thread_id)
|
|
yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
|
|
|
|
return StreamingResponse(
|
|
generate(),
|
|
media_type="text/event-stream",
|
|
headers={"X-Accel-Buffering": "no", "Cache-Control": "no-cache"},
|
|
)
|