mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-30 20:35:22 +08:00
539 lines
20 KiB
Python
539 lines
20 KiB
Python
"""DeerFlow Agent default runner implementation.
|
|
|
|
Ported from LangBot's legacy deerflow-api provider runner and adapted to
|
|
Runner protocol v1.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import typing
|
|
from collections import deque
|
|
|
|
from langbot_plugin.api.definition.components.runner.runner import Runner
|
|
from langbot_plugin.api.entities.builtin.provider.message import Message, MessageChunk
|
|
from langbot_plugin.api.entities.builtin.runner import (
|
|
RunnerContext,
|
|
RunnerResult,
|
|
)
|
|
from pkg import stream_utils
|
|
from pkg.deerflow_client import AsyncDeerFlowClient
|
|
from pkg.errors import DeerFlowAPIError, DeerFlowConfigError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_MAX_VALUES_HISTORY = 200
|
|
|
|
|
|
class _StreamState:
|
|
"""State used while consuming a DeerFlow stream."""
|
|
|
|
def __init__(self) -> None:
|
|
self.latest_text = ""
|
|
self.clarification_text = ""
|
|
self.task_failures: list[str] = []
|
|
self.seen_message_ids: set[str] = set()
|
|
self.seen_message_order: deque[str] = deque()
|
|
self.no_id_message_fingerprints: dict[int, str] = {}
|
|
self.baseline_initialized = False
|
|
self.has_values_text = False
|
|
self.run_values_messages: list[dict[str, typing.Any]] = []
|
|
|
|
|
|
def _content_get(content: typing.Any, key: str, default: typing.Any = None) -> typing.Any:
|
|
if isinstance(content, dict):
|
|
return content.get(key, default)
|
|
return getattr(content, key, default)
|
|
|
|
|
|
def _attachment_get(attachment: typing.Any, key: str, default: typing.Any = None) -> typing.Any:
|
|
if isinstance(attachment, dict):
|
|
return attachment.get(key, default)
|
|
return getattr(attachment, key, default)
|
|
|
|
|
|
def _image_url_value(value: typing.Any) -> str:
|
|
if isinstance(value, str):
|
|
return value
|
|
if isinstance(value, dict):
|
|
return str(value.get("url", ""))
|
|
return str(getattr(value, "url", "") or "")
|
|
|
|
|
|
class DefaultRunner(Runner):
|
|
"""Real Runner for DeerFlow LangGraph HTTP API.
|
|
|
|
Configuration is read from ctx.config using the legacy deerflow-api field
|
|
names so host-side migration can map old pipeline configs directly.
|
|
|
|
Runtime state:
|
|
- external.thread_id: DeerFlow LangGraph thread id for this conversation.
|
|
"""
|
|
|
|
def _validate_config(self, ctx: RunnerContext) -> dict[str, typing.Any]:
|
|
config = ctx.config or {}
|
|
|
|
api_base = self._as_string(config, "api-base", "").strip()
|
|
if not api_base or not api_base.startswith(("http://", "https://")):
|
|
raise DeerFlowConfigError("api-base must start with http:// or https://")
|
|
|
|
return {
|
|
"api_base": api_base,
|
|
"api_key": self._as_string(config, "api-key", ""),
|
|
"auth_header": self._as_string(config, "auth-header", ""),
|
|
"assistant_id": self._as_string(config, "assistant-id", "lead_agent"),
|
|
"model_name": self._as_string(config, "model-name", ""),
|
|
"thinking_enabled": self._as_bool(config, "thinking-enabled", False),
|
|
"plan_mode": self._as_bool(config, "plan-mode", False),
|
|
"subagent_enabled": self._as_bool(config, "subagent-enabled", False),
|
|
"max_concurrent_subagents": self._as_int(config, "max-concurrent-subagents", 3),
|
|
"timeout": self._as_int(config, "timeout", 300),
|
|
"recursion_limit": self._as_int(config, "recursion-limit", 1000),
|
|
}
|
|
|
|
def _as_string(self, config: dict[str, typing.Any], name: str, default: str) -> str:
|
|
value = config.get(name, default)
|
|
if not isinstance(value, str):
|
|
raise DeerFlowConfigError(f"{name} must be a string")
|
|
# Credentials and provider identifiers are opaque, including whitespace.
|
|
return value
|
|
|
|
def _as_bool(self, config: dict[str, typing.Any], name: str, default: bool) -> bool:
|
|
value = config.get(name, default)
|
|
if not isinstance(value, bool):
|
|
raise DeerFlowConfigError(f"{name} must be a boolean")
|
|
return value
|
|
|
|
def _as_int(self, config: dict[str, typing.Any], name: str, default: int) -> int:
|
|
raw_value = config.get(name, default)
|
|
if isinstance(raw_value, bool) or not isinstance(raw_value, (int, str)):
|
|
raise DeerFlowConfigError(f"{name} must be an integer")
|
|
try:
|
|
return int(raw_value)
|
|
except (TypeError, ValueError):
|
|
raise DeerFlowConfigError(f"{name} must be an integer") from None
|
|
|
|
def _should_stream(self, ctx: RunnerContext) -> bool:
|
|
configured = (ctx.config or {}).get("streaming")
|
|
if configured is not None:
|
|
return bool(configured)
|
|
return bool(ctx.delivery.supports_streaming or ctx.runtime.metadata.get("streaming_supported", False))
|
|
|
|
def _fingerprint_message(self, message: dict[str, typing.Any]) -> str:
|
|
try:
|
|
raw = json.dumps(message, sort_keys=True, ensure_ascii=False, default=str)
|
|
except (TypeError, ValueError):
|
|
raw = repr(message)
|
|
return hashlib.sha1(raw.encode("utf-8", errors="ignore")).hexdigest()
|
|
|
|
def _remember_seen_message_id(self, state: _StreamState, msg_id: str) -> None:
|
|
if not msg_id or msg_id in state.seen_message_ids:
|
|
return
|
|
state.seen_message_ids.add(msg_id)
|
|
state.seen_message_order.append(msg_id)
|
|
while len(state.seen_message_order) > _MAX_VALUES_HISTORY:
|
|
dropped = state.seen_message_order.popleft()
|
|
state.seen_message_ids.discard(dropped)
|
|
|
|
def _extract_new_messages_from_values(
|
|
self,
|
|
values_messages: list[typing.Any],
|
|
state: _StreamState,
|
|
) -> list[dict[str, typing.Any]]:
|
|
new_messages: list[dict[str, typing.Any]] = []
|
|
no_id_indexes_seen: set[int] = set()
|
|
for idx, msg in enumerate(values_messages):
|
|
if not isinstance(msg, dict):
|
|
continue
|
|
msg_id = stream_utils.get_message_id(msg)
|
|
if msg_id:
|
|
if msg_id in state.seen_message_ids:
|
|
continue
|
|
self._remember_seen_message_id(state, msg_id)
|
|
new_messages.append(msg)
|
|
continue
|
|
|
|
no_id_indexes_seen.add(idx)
|
|
fingerprint = self._fingerprint_message(msg)
|
|
if state.no_id_message_fingerprints.get(idx) == fingerprint:
|
|
continue
|
|
state.no_id_message_fingerprints[idx] = fingerprint
|
|
new_messages.append(msg)
|
|
|
|
for idx in list(state.no_id_message_fingerprints.keys()):
|
|
if idx not in no_id_indexes_seen:
|
|
state.no_id_message_fingerprints.pop(idx, None)
|
|
return new_messages
|
|
|
|
def _build_user_content(
|
|
self,
|
|
prompt: str,
|
|
image_urls: list[str],
|
|
) -> typing.Any:
|
|
if not image_urls:
|
|
return prompt
|
|
|
|
content: list[dict[str, typing.Any]] = []
|
|
if prompt:
|
|
content.append({"type": "text", "text": prompt})
|
|
for url in image_urls:
|
|
url = url.strip()
|
|
if not url:
|
|
continue
|
|
if url.startswith(("http://", "https://", "data:")):
|
|
content.append({"type": "image_url", "image_url": {"url": url}})
|
|
return content if content else prompt
|
|
|
|
def _extract_image_urls(self, ctx: RunnerContext) -> list[str]:
|
|
image_urls: list[str] = []
|
|
|
|
for item in ctx.input.contents or []:
|
|
item_type = _content_get(item, "type")
|
|
if item_type == "image_base64":
|
|
value = _content_get(item, "image_base64")
|
|
if isinstance(value, str) and value:
|
|
if not value.startswith("data:"):
|
|
value = f"data:image/png;base64,{value}"
|
|
image_urls.append(value)
|
|
elif item_type == "image_url":
|
|
value = _image_url_value(_content_get(item, "image_url"))
|
|
if value:
|
|
image_urls.append(value)
|
|
|
|
for attachment in ctx.input.attachments or []:
|
|
url = _attachment_get(attachment, "url")
|
|
if isinstance(url, str) and url:
|
|
image_urls.append(url)
|
|
continue
|
|
|
|
content = _attachment_get(attachment, "content")
|
|
mime_type = _attachment_get(attachment, "mime_type") or "image/png"
|
|
artifact_type = str(_attachment_get(attachment, "artifact_type") or "").lower()
|
|
if (
|
|
isinstance(content, str)
|
|
and content
|
|
and (artifact_type == "image" or str(mime_type).startswith("image/"))
|
|
):
|
|
if not content.startswith("data:"):
|
|
content = f"data:{mime_type};base64,{content}"
|
|
image_urls.append(content)
|
|
|
|
return image_urls
|
|
|
|
def _build_messages(
|
|
self,
|
|
prompt: str,
|
|
image_urls: list[str],
|
|
system_prompt: str = "",
|
|
) -> list[dict[str, typing.Any]]:
|
|
messages: list[dict[str, typing.Any]] = []
|
|
if system_prompt:
|
|
messages.append({"role": "system", "content": system_prompt})
|
|
messages.append(
|
|
{
|
|
"role": "user",
|
|
"content": self._build_user_content(prompt, image_urls),
|
|
}
|
|
)
|
|
return messages
|
|
|
|
def _build_runtime_configurable(self, config: dict[str, typing.Any], thread_id: str) -> dict[str, typing.Any]:
|
|
runtime_configurable: dict[str, typing.Any] = {
|
|
"thread_id": thread_id,
|
|
"thinking_enabled": config["thinking_enabled"],
|
|
"is_plan_mode": config["plan_mode"],
|
|
"subagent_enabled": config["subagent_enabled"],
|
|
}
|
|
if config["subagent_enabled"]:
|
|
runtime_configurable["max_concurrent_subagents"] = config["max_concurrent_subagents"]
|
|
if config["model_name"]:
|
|
runtime_configurable["model_name"] = config["model_name"]
|
|
return runtime_configurable
|
|
|
|
def _build_payload(
|
|
self,
|
|
config: dict[str, typing.Any],
|
|
thread_id: str,
|
|
prompt: str,
|
|
image_urls: list[str],
|
|
system_prompt: str = "",
|
|
) -> dict[str, typing.Any]:
|
|
runtime_configurable = self._build_runtime_configurable(config, thread_id)
|
|
return {
|
|
"assistant_id": config["assistant_id"],
|
|
"input": {
|
|
"messages": self._build_messages(prompt, image_urls, system_prompt),
|
|
},
|
|
"stream_mode": ["values", "messages-tuple", "custom"],
|
|
"context": dict(runtime_configurable),
|
|
"config": {
|
|
"recursion_limit": config["recursion_limit"],
|
|
"configurable": runtime_configurable,
|
|
},
|
|
}
|
|
|
|
async def _ensure_thread_id(
|
|
self,
|
|
ctx: RunnerContext,
|
|
client: AsyncDeerFlowClient,
|
|
timeout: int,
|
|
) -> tuple[str, bool]:
|
|
thread_id = ctx.state.conversation.get("external.thread_id")
|
|
if thread_id:
|
|
return str(thread_id), False
|
|
|
|
thread = await client.create_thread(timeout=min(30, timeout))
|
|
thread_id = thread.get("thread_id", "")
|
|
if not thread_id:
|
|
raise DeerFlowAPIError(
|
|
f"DeerFlow create thread response missing thread_id: {thread}",
|
|
code="deerflow.api_error",
|
|
)
|
|
return str(thread_id), True
|
|
|
|
def _handle_values_event(
|
|
self,
|
|
data: typing.Any,
|
|
state: _StreamState,
|
|
) -> str | None:
|
|
values_messages = stream_utils.extract_messages_from_values_data(data)
|
|
if not values_messages:
|
|
return None
|
|
|
|
if not state.baseline_initialized:
|
|
state.baseline_initialized = True
|
|
new_messages: list[dict[str, typing.Any]] = []
|
|
for idx, msg in enumerate(values_messages):
|
|
if not isinstance(msg, dict):
|
|
continue
|
|
new_messages.append(msg)
|
|
msg_id = stream_utils.get_message_id(msg)
|
|
if msg_id:
|
|
self._remember_seen_message_id(state, msg_id)
|
|
continue
|
|
state.no_id_message_fingerprints[idx] = self._fingerprint_message(msg)
|
|
else:
|
|
new_messages = self._extract_new_messages_from_values(values_messages, state)
|
|
|
|
latest_text = ""
|
|
if new_messages:
|
|
state.run_values_messages.extend(new_messages)
|
|
if len(state.run_values_messages) > _MAX_VALUES_HISTORY:
|
|
state.run_values_messages = state.run_values_messages[-_MAX_VALUES_HISTORY:]
|
|
latest_text = stream_utils.extract_latest_ai_text(state.run_values_messages)
|
|
if latest_text:
|
|
state.has_values_text = True
|
|
latest_clarification = stream_utils.extract_latest_clarification_text(state.run_values_messages)
|
|
if latest_clarification:
|
|
state.clarification_text = latest_clarification
|
|
|
|
return latest_text or None
|
|
|
|
def _handle_message_event(
|
|
self,
|
|
data: typing.Any,
|
|
state: _StreamState,
|
|
) -> str | None:
|
|
delta = stream_utils.extract_ai_delta_from_event_data(data)
|
|
if delta and not state.has_values_text:
|
|
state.latest_text += delta
|
|
return delta
|
|
|
|
maybe_clarification = stream_utils.extract_clarification_from_event_data(data)
|
|
if maybe_clarification:
|
|
state.clarification_text = maybe_clarification
|
|
return None
|
|
|
|
def _build_final_text(self, state: _StreamState) -> str:
|
|
if state.clarification_text:
|
|
return state.clarification_text
|
|
|
|
latest_ai = stream_utils.extract_latest_ai_message(state.run_values_messages)
|
|
if latest_ai:
|
|
text = stream_utils.extract_text(latest_ai.get("content"))
|
|
if text:
|
|
return text
|
|
|
|
if state.latest_text:
|
|
return state.latest_text
|
|
|
|
failure_text = stream_utils.build_task_failure_summary(state.task_failures)
|
|
if failure_text:
|
|
return failure_text
|
|
|
|
return "DeerFlow returned an empty response"
|
|
|
|
async def run(self, ctx: RunnerContext) -> typing.AsyncGenerator[RunnerResult, None]:
|
|
"""Run the DeerFlow LangGraph agent."""
|
|
try:
|
|
config = self._validate_config(ctx)
|
|
except DeerFlowConfigError as e:
|
|
yield RunnerResult.run_failed(
|
|
ctx.run_id,
|
|
error=e.message,
|
|
code=e.code,
|
|
retryable=getattr(e, "retryable", False),
|
|
)
|
|
return
|
|
|
|
client = AsyncDeerFlowClient(
|
|
api_base=config["api_base"],
|
|
api_key=config["api_key"],
|
|
auth_header=config["auth_header"],
|
|
)
|
|
|
|
try:
|
|
thread_id, _created = await self._ensure_thread_id(ctx, client, config["timeout"])
|
|
prompt = ctx.input.to_text() or "continue"
|
|
image_urls = self._extract_image_urls(ctx)
|
|
payload = self._build_payload(
|
|
config=config,
|
|
thread_id=thread_id,
|
|
prompt=prompt,
|
|
image_urls=image_urls,
|
|
)
|
|
|
|
if self._should_stream(ctx):
|
|
async for result in self._stream_run(ctx, client, config, thread_id, payload):
|
|
yield result
|
|
else:
|
|
message = await self._complete_run(client, config, thread_id, payload)
|
|
yield RunnerResult.message_completed(ctx.run_id, message)
|
|
|
|
yield RunnerResult.state_updated(
|
|
ctx.run_id,
|
|
"external.thread_id",
|
|
thread_id,
|
|
scope="conversation",
|
|
)
|
|
yield RunnerResult.run_completed(ctx.run_id)
|
|
except DeerFlowAPIError as e:
|
|
yield RunnerResult.run_failed(
|
|
ctx.run_id,
|
|
error=e.message,
|
|
code=e.code,
|
|
retryable=getattr(e, "retryable", False),
|
|
)
|
|
return
|
|
except Exception as e:
|
|
logger.exception(f"DeerFlow runner unexpected error: {e}")
|
|
yield RunnerResult.run_failed(
|
|
ctx.run_id,
|
|
error=f"DeerFlow runner error: {e}",
|
|
code="deerflow.unexpected_error",
|
|
)
|
|
return
|
|
|
|
async def _stream_run(
|
|
self,
|
|
ctx: RunnerContext,
|
|
client: AsyncDeerFlowClient,
|
|
config: dict[str, typing.Any],
|
|
thread_id: str,
|
|
payload: dict[str, typing.Any],
|
|
) -> typing.AsyncGenerator[RunnerResult, None]:
|
|
state = _StreamState()
|
|
prev_text = ""
|
|
|
|
try:
|
|
async for event in client.stream_run(
|
|
thread_id=thread_id,
|
|
payload=payload,
|
|
timeout=config["timeout"],
|
|
):
|
|
event_type = event.get("event")
|
|
data = event.get("data")
|
|
|
|
if event_type == "values":
|
|
new_full = self._handle_values_event(data, state)
|
|
if new_full and new_full != prev_text:
|
|
changed_text = new_full[len(prev_text) :] if new_full.startswith(prev_text) else new_full
|
|
prev_text = new_full
|
|
if changed_text:
|
|
yield RunnerResult.message_delta(
|
|
ctx.run_id,
|
|
MessageChunk(role="assistant", content=new_full, is_final=False),
|
|
)
|
|
continue
|
|
|
|
if event_type in {"messages-tuple", "messages", "message"}:
|
|
delta = self._handle_message_event(data, state)
|
|
if delta:
|
|
prev_text = state.latest_text
|
|
yield RunnerResult.message_delta(
|
|
ctx.run_id,
|
|
MessageChunk(role="assistant", content=prev_text, is_final=False),
|
|
)
|
|
continue
|
|
|
|
if event_type == "custom":
|
|
state.task_failures.extend(stream_utils.extract_task_failures_from_custom_event(data))
|
|
continue
|
|
|
|
if event_type == "error":
|
|
raise DeerFlowAPIError(f"DeerFlow stream error event: {data}", code="deerflow.api_error")
|
|
|
|
if event_type == "end":
|
|
break
|
|
except TimeoutError:
|
|
logger.warning(f"DeerFlow stream timed out after {config['timeout']}s for thread_id={thread_id}")
|
|
raise DeerFlowAPIError(
|
|
f"DeerFlow stream timed out after {config['timeout']}s",
|
|
code="deerflow.timeout",
|
|
retryable=True,
|
|
) from None
|
|
|
|
final_text = self._build_final_text(state)
|
|
yield RunnerResult.message_delta(
|
|
ctx.run_id,
|
|
MessageChunk(role="assistant", content=final_text, is_final=True),
|
|
)
|
|
|
|
async def _complete_run(
|
|
self,
|
|
client: AsyncDeerFlowClient,
|
|
config: dict[str, typing.Any],
|
|
thread_id: str,
|
|
payload: dict[str, typing.Any],
|
|
) -> Message:
|
|
state = _StreamState()
|
|
|
|
try:
|
|
async for event in client.stream_run(
|
|
thread_id=thread_id,
|
|
payload=payload,
|
|
timeout=config["timeout"],
|
|
):
|
|
event_type = event.get("event")
|
|
data = event.get("data")
|
|
|
|
if event_type == "values":
|
|
self._handle_values_event(data, state)
|
|
continue
|
|
|
|
if event_type in {"messages-tuple", "messages", "message"}:
|
|
self._handle_message_event(data, state)
|
|
continue
|
|
|
|
if event_type == "custom":
|
|
state.task_failures.extend(stream_utils.extract_task_failures_from_custom_event(data))
|
|
continue
|
|
|
|
if event_type == "error":
|
|
raise DeerFlowAPIError(f"DeerFlow stream error event: {data}", code="deerflow.api_error")
|
|
|
|
if event_type == "end":
|
|
break
|
|
except TimeoutError:
|
|
logger.warning(f"DeerFlow stream timed out after {config['timeout']}s for thread_id={thread_id}")
|
|
raise DeerFlowAPIError(
|
|
f"DeerFlow stream timed out after {config['timeout']}s",
|
|
code="deerflow.timeout",
|
|
retryable=True,
|
|
) from None
|
|
|
|
final_text = self._build_final_text(state)
|
|
return Message(role="assistant", content=final_text)
|