feat: 新增 multi-llm 路由智能体及 API 端点

添加 multi-llm 路由智能体(LangGraph),支持多 LLM 分发能力;在 Gateway 中注册 /api/multi-llm 路由,并豁免其 CSRF 和认证校验;同时新增 asyncmy 依赖和聊天流格式文档。

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
xuhdandClaude Sonnet 4.6 committed 2026-05-21 18:48:16 +08:00
1 parent a8a545690f
commit adeace0f7b
15 files changed
+3789 -2767

No files matched your search

@@ -34,6 +34,7 @@ from app.gateway.routers import (
mcp,
memory,
models,
multi_llm,
notifications,
public_scheduled_tasks,
recommended_questions,
@@ -796,6 +797,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# AI Writing multi-agent API is mounted at /api/ai-writing
app.include_router(ai_writing.router)
# Multi-LLM routing agent API is mounted at /api/multi-llm
app.include_router(multi_llm.router)
# Scheduled task API is mounted at /api/scheduled-tasks
app.include_router(scheduled_tasks.router)
@@ -33,6 +33,7 @@ _PUBLIC_PATH_PREFIXES: tuple[str, ...] = (
"/redoc",
"/openapi.json",
"/api/public/",
"/api/multi-llm/",
)
# Exact auth paths that are public (login/register/status check).
@@ -41,6 +41,9 @@ def should_check_csrf(request: Request) -> bool:
# Exempt /api/v1/auth/me endpoint
if path == "/api/v1/auth/me":
return False
# Exempt multi-llm streaming endpoint (API-to-API calls, no browser session)
if path.startswith("/api/multi-llm"):
return False
return True
@@ -1,3 +1,3 @@
from . import ai_writing, artifacts, assistants_compat, mcp, models, recommended_questions, skills, suggestions, thread_runs, threads, uploads, writing
from . import ai_writing, artifacts, assistants_compat, mcp, models, multi_llm, recommended_questions, skills, suggestions, thread_runs, threads, uploads, writing
__all__ = ["ai_writing", "artifacts", "assistants_compat", "mcp", "models", "recommended_questions", "skills", "suggestions", "threads", "thread_runs", "uploads", "writing"]
__all__ = ["ai_writing", "artifacts", "assistants_compat", "mcp", "models", "multi_llm", "recommended_questions", "skills", "suggestions", "threads", "thread_runs", "uploads", "writing"]
@@ -0,0 +1,89 @@
"""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"},
)
@@ -0,0 +1,51 @@
"""Multi-LLM routing graph.
Systems: fast, bigmodel, hxchat, moreModes, rag, taskChat
Graph layout:
START → router → (conditional) → fast / bigmodel / hxchat / more_modes / rag / task_chat → END
"""
from __future__ import annotations
from langgraph.graph import END, START, StateGraph
from deerflow.agents.multi_llm.nodes.router import route_by_system, router_node
from deerflow.agents.multi_llm.nodes.systems import (
bigmodel_node,
fast_node,
hxchat_node,
more_modes_node,
rag_node,
task_chat_node,
)
from deerflow.agents.multi_llm.state import MultiLLMState
# LangGraph node name → node function
_SYSTEM_NODES = {
"fast": fast_node,
"bigmodel": bigmodel_node,
"hxchat": hxchat_node,
"more_modes": more_modes_node,
"rag": rag_node,
"task_chat": task_chat_node,
}
def build_multi_llm_graph(checkpointer=None):
g = StateGraph(MultiLLMState)
g.add_node("router", router_node)
for name, fn in _SYSTEM_NODES.items():
g.add_node(name, fn)
g.add_edge(START, "router")
g.add_conditional_edges(
"router",
route_by_system,
{name: name for name in _SYSTEM_NODES},
)
for name in _SYSTEM_NODES:
g.add_edge(name, END)
return g.compile(checkpointer=checkpointer)
@@ -0,0 +1,77 @@
"""Router node: uses an LLM to decide which system to invoke."""
from __future__ import annotations
import logging
from typing import Literal
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.runnables import RunnableConfig
from deerflow.agents.multi_llm.state import MultiLLMState
from deerflow.config.app_config import get_app_config
from deerflow.models import create_chat_model
logger = logging.getLogger(__name__)
SystemName = Literal["fast", "bigmodel", "hxchat", "moreModes", "rag", "taskChat"]
_VALID_SYSTEMS: tuple[str, ...] = ("fast", "bigmodel", "hxchat", "moreModes", "rag", "taskChat")
_ROUTER_SYSTEM_PROMPT = f"""你是一个路由助手,负责将用户请求分配给最合适的AI系统。
可用系统(共 {len(_VALID_SYSTEMS)} 个):
- fast : 快速问答,适合通用问题、知识库检索
- bigmodel : 大模型,适合复杂推理、长文本分析
- hxchat : HX 对话,适合...(TODO: 补充说明)
- moreModes : 多模态,适合...(TODO: 补充说明)
- rag : RAG 检索增强,适合需要查找文档的问题
- taskChat : 任务对话,适合需要执行多步骤任务的场景
请根据用户请求,仅输出一个系统名称(区分大小写),不要输出任何其他内容。
有效值:fast / bigmodel / hxchat / moreModes / rag / taskChat"""
async def router_node(state: MultiLLMState, config: RunnableConfig) -> dict:
messages = state.get("messages", [])
if not messages:
return {"selected_system": "fast"}
last_human = next(
(
m
for m in reversed(messages)
if getattr(m, "type", None) == "human" or (isinstance(m, dict) and m.get("role") == "user")
),
None,
)
user_content = ""
if last_human is not None:
user_content = last_human.content if hasattr(last_human, "content") else last_human.get("content", "")
try:
app_config = get_app_config()
llm = create_chat_model(app_config=app_config)
prompt = [
SystemMessage(content=_ROUTER_SYSTEM_PROMPT),
HumanMessage(content=user_content or "请帮我处理这个请求"),
]
response = await llm.ainvoke(prompt, config={"run_name": "multi_llm_router"})
raw = (response.content or "").strip()
selected = raw if raw in _VALID_SYSTEMS else "fast"
except Exception:
logger.exception("multi_llm router failed, defaulting to fast")
selected = "fast"
return {"selected_system": selected}
def route_by_system(state: MultiLLMState) -> str:
system = state.get("selected_system", "fast")
if system not in _VALID_SYSTEMS:
system = "fast"
# LangGraph node names use underscores
return system.lower() if system in ("fast", "bigmodel", "hxchat", "rag") else {
"moreModes": "more_modes",
"taskChat": "task_chat",
}[system]
@@ -0,0 +1,223 @@
"""System nodes: call each external LLM system and stream chunks via custom events.
Systems: fast, bigmodel, hxchat, moreModes, rag, taskChat
System URLs are read from config.yaml under multi_llm_systems.<name>_url,
falling back to environment variables FAST_URL / BIGMODEL_URL / etc.
fast uses a custom request format (Vercel AI SDK / POST /new_chat):
{"chat_id": <thread_id>, "search_space_id": <id>, "user_query": "..."}
search_space_id from config multi_llm_systems.fast_search_space_id
or env FAST_SEARCH_SPACE_ID (default 0).
All other systems use generic {"messages": [...]} format (placeholder until
their real formats are confirmed).
"""
from __future__ import annotations
import logging
import os
import httpx
from langchain_core.callbacks.manager import adispatch_custom_event
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.runnables import RunnableConfig
from deerflow.agents.multi_llm.parsers import PARSERS
from deerflow.agents.multi_llm.state import MultiLLMState
from deerflow.config.app_config import get_app_config
from deerflow.models import create_chat_model
logger = logging.getLogger(__name__)
SYSTEM_NAMES = ["fast", "bigmodel", "hxchat", "moreModes", "rag", "taskChat"]
# Maps system name → env var suffix for URL lookup
_ENV_VAR_MAP = {
"fast": "FAST_URL",
"bigmodel": "BIGMODEL_URL",
"hxchat": "HXCHAT_URL",
"moreModes": "MOREMODES_URL",
"rag": "RAG_URL",
"taskChat": "TASKCHAT_URL",
}
def _get_multi_llm_config() -> dict:
try:
cfg = get_app_config()
return getattr(cfg, "multi_llm_systems", None) or {}
except Exception:
return {}
def _get_url(system: str) -> str:
c = _get_multi_llm_config()
config_key = f"{system}_url"
env_key = _ENV_VAR_MAP.get(system, f"{system.upper()}_URL")
return c.get(config_key) or os.environ.get(env_key, "")
def _extract_last_user_content(messages: list) -> str:
for m in reversed(messages):
if hasattr(m, "type") and m.type == "human":
return m.content if isinstance(m.content, str) else str(m.content)
if isinstance(m, dict) and m.get("role") == "user":
return m.get("content", "")
return ""
def _build_generic_payload(messages: list) -> dict:
result = []
for m in messages:
if hasattr(m, "type"):
role = "user" if m.type == "human" else "assistant"
content = m.content if isinstance(m.content, str) else str(m.content)
elif isinstance(m, dict):
role = m.get("role", "user")
content = m.get("content", "")
else:
continue
result.append({"role": role, "content": content})
return {"messages": result}
_FAST_TEST_URL = "http://47.88.25.99:8000/api/v1/new_chat"
_FAST_TEST_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiI5MmFhNmRkMC03NDA5LTQ0NDEtOGQzNC1iNTY4YmZjMzg5ODIiLCJhdWQiOlsiZmFzdGFwaS11c2VyczphdXRoIl0sImV4cCI6MTc3OTQzNjY0NH0.Da_KDlQaEc-cH95AyUzcx82pwZmU3t-9fGmJC9tRF7E"
_FAST_TEST_MODEL = "Qwen3.5-plus"
_FAST_TEST_SEARCH_SPACE_ID = 197
_FAST_TEST_CHAT_ID = 600
def _build_fast_payload(messages: list, thread_id: str = "") -> dict: # noqa: ARG001
"""Build the /new_chat request payload for the fast system."""
return {
"model_name": _FAST_TEST_MODEL,
"chat_id": _FAST_TEST_CHAT_ID,
"search_space_id": _FAST_TEST_SEARCH_SPACE_ID,
"user_query": _extract_last_user_content(messages),
}
async def _call_system(system: str, messages: list, thread_id: str = "") -> str:
if system == "fast":
url = _FAST_TEST_URL
headers = {"Authorization": f"Bearer {_FAST_TEST_TOKEN}"}
else:
url = _get_url(system)
headers = {}
if not url:
raise ValueError(
f"No URL configured for system '{system}'. "
f"Set multi_llm_systems.{system}_url in config.yaml or {_ENV_VAR_MAP.get(system, system.upper() + '_URL')} env var."
)
parser = PARSERS[system]
payload = _build_fast_payload(messages, thread_id) if system == "fast" else _build_generic_payload(messages)
text_buf = ""
async with httpx.AsyncClient(timeout=120) as client:
async with client.stream("POST", url, json=payload, headers=headers) as resp:
resp.raise_for_status()
async for chunk in parser.parse(resp):
if chunk.content:
if chunk.type == "thinking":
await adispatch_custom_event("llm_chunk", {"type": "thinking", "content": chunk.content, "system": system})
else:
text_buf += chunk.content
await adispatch_custom_event("llm_chunk", {"type": "text", "content": chunk.content, "system": system})
return text_buf
def _extract_chunk_parts(chunk: object) -> tuple[str, str]:
"""Return (thinking_delta, text_delta) from a LangChain streaming chunk."""
thinking = ""
text = ""
content = getattr(chunk, "content", "")
if isinstance(content, str):
text = content
elif isinstance(content, list):
for block in content:
if isinstance(block, str):
text += block
elif isinstance(block, dict):
btype = block.get("type", "")
if btype in ("thinking", "thinking_delta"):
thinking += block.get("thinking", "")
elif btype in ("text", "text_delta"):
text += block.get("text", "")
elif btype == "reasoning":
thinking += block.get("content", "")
if not thinking:
kwargs = getattr(chunk, "additional_kwargs", {}) or {}
r = kwargs.get("reasoning_content") or ""
if isinstance(r, str):
thinking = r
return thinking, text
async def _fallback_to_local_llm(messages: list, config: RunnableConfig) -> str:
"""Fall back to the local configured LLM when an external system fails."""
app_config = get_app_config()
llm = create_chat_model(app_config=app_config)
lc_messages = []
for m in messages:
if hasattr(m, "type"):
content = m.content if isinstance(m.content, str) else str(m.content)
lc_messages.append(HumanMessage(content=content) if m.type == "human" else AIMessage(content=content))
elif isinstance(m, dict):
content = m.get("content", "")
lc_messages.append(HumanMessage(content=content) if m.get("role") == "user" else AIMessage(content=content))
text_buf = ""
async for chunk in llm.astream(lc_messages):
thinking, text = _extract_chunk_parts(chunk)
if thinking:
await adispatch_custom_event("llm_chunk", {"type": "thinking", "content": thinking, "system": "local"})
if text:
text_buf += text
await adispatch_custom_event("llm_chunk", {"type": "text", "content": text, "system": "local"})
return text_buf
async def _system_node(system: str, state: MultiLLMState, config: RunnableConfig) -> dict:
messages = state.get("messages", [])
thread_id = (config or {}).get("configurable", {}).get("thread_id", "")
try:
result = await _call_system(system, messages, thread_id=thread_id)
except Exception as e:
logger.warning("multi_llm system '%s' failed (%s), falling back to local LLM", system, e)
await adispatch_custom_event("llm_chunk", {"type": "fallback", "system": "local", "reason": str(e)})
try:
result = await _fallback_to_local_llm(messages, config)
except Exception as e2:
logger.exception("local LLM fallback also failed")
result = f"[系统异常,请稍后重试: {e2}]"
return {"messages": [AIMessage(content=result)]}
async def fast_node(state: MultiLLMState, config: RunnableConfig) -> dict:
return await _system_node("fast", state, config)
async def bigmodel_node(state: MultiLLMState, config: RunnableConfig) -> dict:
return await _system_node("bigmodel", state, config)
async def hxchat_node(state: MultiLLMState, config: RunnableConfig) -> dict:
return await _system_node("hxchat", state, config)
async def more_modes_node(state: MultiLLMState, config: RunnableConfig) -> dict:
return await _system_node("moreModes", state, config)
async def rag_node(state: MultiLLMState, config: RunnableConfig) -> dict:
return await _system_node("rag", state, config)
async def task_chat_node(state: MultiLLMState, config: RunnableConfig) -> dict:
return await _system_node("taskChat", state, config)
@@ -0,0 +1,88 @@
"""Stream parsers for 6 external LLM systems.
Each system returns a different streaming format. All parsers normalize the
raw HTTP byte stream into a sequence of StreamChunk(type, content) where
type is either "thinking" (reasoning process) or "text" (final output).
Systems: fast, bigmodel, hxchat, moreModes, rag, taskChat
Currently implemented: fast (Vercel AI SDK Data Stream Protocol / POST /new_chat)
Others: TODO — using plain-text placeholder until formats are confirmed.
"""
from __future__ import annotations
import abc
import json
from dataclasses import dataclass
from typing import AsyncIterator, Literal
@dataclass
class StreamChunk:
type: Literal["thinking", "text"]
content: str
class BaseSystemParser(abc.ABC):
"""Parse a raw httpx response stream into normalized StreamChunks."""
@abc.abstractmethod
async def parse(self, response) -> AsyncIterator[StreamChunk]:
...
class FastParser(BaseSystemParser):
"""fast: Vercel AI SDK Data Stream Protocol (POST /new_chat).
Relevant event types:
text-delta → StreamChunk(type="text", content=payload["delta"])
reasoning-delta → StreamChunk(type="thinking", content=payload["delta"])
All other event types (start, finish, start-step, finish-step,
tool-input-*, tool-output-*, data-*, source-*, error, etc.) are silently
skipped — they are UI-control signals, not content.
"""
async def parse(self, response) -> AsyncIterator[StreamChunk]:
async for line in response.aiter_lines():
if not line.startswith("data:"):
continue
raw = line[5:].strip()
if not raw or raw == "[DONE]":
continue
try:
payload = json.loads(raw)
except json.JSONDecodeError:
continue
ev_type = payload.get("type", "")
if ev_type == "text-delta":
delta = payload.get("delta", "")
if delta:
yield StreamChunk(type="text", content=delta)
elif ev_type == "reasoning-delta":
delta = payload.get("delta", "")
if delta:
yield StreamChunk(type="thinking", content=delta)
class PlainTextParser(BaseSystemParser):
"""Fallback: plain text stream, no thinking content.
Used as placeholder for bigmodel / hxchat / moreModes / rag / taskChat
until their real streaming formats are confirmed.
"""
async def parse(self, response) -> AsyncIterator[StreamChunk]:
async for chunk in response.aiter_text():
if chunk:
yield StreamChunk(type="text", content=chunk)
PARSERS: dict[str, BaseSystemParser] = {
"fast": FastParser(),
"bigmodel": PlainTextParser(), # TODO: replace with real parser
"hxchat": PlainTextParser(), # TODO: replace with real parser
"moreModes": PlainTextParser(), # TODO: replace with real parser
"rag": PlainTextParser(), # TODO: replace with real parser
"taskChat": PlainTextParser(), # TODO: replace with real parser
}
@@ -0,0 +1,9 @@
from typing import Annotated
from langgraph.graph.message import add_messages
from typing_extensions import NotRequired, TypedDict
class MultiLLMState(TypedDict):
messages: Annotated[list, add_messages]
selected_system: NotRequired[str]
+49 -48
View File
@@ -1,51 +1,52 @@
[project]
name = "deer-flow"
version = "0.1.0"
description = "LangGraph-based AI agent system with sandbox execution capabilities"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"deerflow-harness",
"fastapi>=0.115.0",
"httpx>=0.28.0",
"python-multipart>=0.0.26",
"sse-starlette>=2.1.0",
"uvicorn[standard]>=0.34.0",
"lark-oapi>=1.4.0",
"slack-sdk>=3.33.0",
"python-telegram-bot>=21.0",
"langgraph-sdk>=0.1.51",
"markdown-to-mrkdwn>=0.3.1",
"wecom-aibot-python-sdk>=0.1.6",
"dingtalk-stream>=0.24.3",
"bcrypt>=4.0.0",
"pyjwt>=2.9.0",
"email-validator>=2.0.0",
"croniter>=6.2.2",
]
[project]
name = "deer-flow"
version = "0.1.0"
description = "LangGraph-based AI agent system with sandbox execution capabilities"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"deerflow-harness",
"fastapi>=0.115.0",
"httpx>=0.28.0",
"python-multipart>=0.0.26",
"sse-starlette>=2.1.0",
"uvicorn[standard]>=0.34.0",
"lark-oapi>=1.4.0",
"slack-sdk>=3.33.0",
"python-telegram-bot>=21.0",
"langgraph-sdk>=0.1.51",
"markdown-to-mrkdwn>=0.3.1",
"wecom-aibot-python-sdk>=0.1.6",
"dingtalk-stream>=0.24.3",
"bcrypt>=4.0.0",
"pyjwt>=2.9.0",
"email-validator>=2.0.0",
"croniter>=6.2.2",
"asyncmy>=0.2.11",
]
[project.optional-dependencies]
postgres = ["deerflow-harness[postgres]"]
mysql = ["deerflow-harness[mysql]"]
[dependency-groups]
dev = [
"prompt-toolkit>=3.0.0",
"pytest>=9.0.3",
"pytest-asyncio>=1.3.0",
"ruff>=0.14.11",
]
[tool.pytest.ini_options]
markers = [
"no_auto_user: disable the conftest autouse contextvar fixture for this test",
]
[tool.uv]
index-url = "https://pypi.org/simple"
[tool.uv.workspace]
members = ["packages/harness"]
[tool.uv.sources]
deerflow-harness = { workspace = true }
[dependency-groups]
dev = [
"prompt-toolkit>=3.0.0",
"pytest>=9.0.3",
"pytest-asyncio>=1.3.0",
"ruff>=0.14.11",
]
[tool.pytest.ini_options]
markers = [
"no_auto_user: disable the conftest autouse contextvar fixture for this test",
]
[tool.uv]
index-url = "https://pypi.org/simple"
[tool.uv.workspace]
members = ["packages/harness"]
[tool.uv.sources]
deerflow-harness = { workspace = true }
+2719 -2717
View File
File diff suppressed because it is too large. Load diff