Files
deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/agents/multi_llm/parsers.py
T
xuhdandClaude Sonnet 4.6 adeace0f7b feat: 新增 multi-llm 路由智能体及 API 端点
添加 multi-llm 路由智能体(LangGraph),支持多 LLM 分发能力;在 Gateway 中注册 /api/multi-llm 路由,并豁免其 CSRF 和认证校验;同时新增 asyncmy 依赖和聊天流格式文档。

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-21 18:48:16 +08:00

89 lines
3.0 KiB
Python

"""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
}