365 lines
14 KiB
Python
365 lines
14 KiB
Python
"""Runs a DeerFlow agent for one workflow ``agent`` / ``skill`` node.
|
|
|
|
This is the single place where the workflow runtime touches LangGraph. We build
|
|
the lead agent with :func:`make_lead_agent` (so the node inherits the same
|
|
middleware chain, SOUL, skills and model resolution as a normal chat turn) and
|
|
stream ``stream_mode=["values", "messages"]`` so the node can forward token
|
|
deltas into the workflow event log.
|
|
|
|
Each node gets its own thread id (``wf-{run_id}-{node_id}``) unless the node
|
|
asks for ``shared_run_thread``, which keeps the whole run in one conversation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
|
|
from langchain_core.messages import AIMessage, ToolMessage
|
|
|
|
from deerflow.workflows.errors import WorkflowError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _extract_text(content: Any) -> str:
|
|
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 block.get("type") == "text":
|
|
parts.append(str(block.get("text") or ""))
|
|
return "".join(parts)
|
|
return ""
|
|
|
|
|
|
def _extract_reasoning(message: AIMessage) -> str:
|
|
"""Read provider reasoning that LangChain keeps outside ``content``.
|
|
|
|
Patched OpenAI/vLLM providers preserve the streaming field under
|
|
``additional_kwargs.reasoning_content``. Some compatible providers use
|
|
the shorter ``reasoning`` spelling, so accept both without making either
|
|
field part of the workflow's executable output.
|
|
"""
|
|
|
|
additional_kwargs = getattr(message, "additional_kwargs", None)
|
|
if not isinstance(additional_kwargs, dict):
|
|
return ""
|
|
for key in ("reasoning_content", "reasoning"):
|
|
value = additional_kwargs.get(key)
|
|
text = _extract_text(value)
|
|
if text:
|
|
return text
|
|
if isinstance(value, str) and value:
|
|
return value
|
|
return ""
|
|
|
|
|
|
def _thread_id(run_id: str, node_id: str, thread_mode: str) -> str:
|
|
if thread_mode == "shared_run_thread":
|
|
return f"wf-{run_id}"
|
|
return f"wf-{run_id}-{node_id}"
|
|
|
|
|
|
def _native_message_frame(chunk: Any) -> list[Any]:
|
|
"""Keep the DeerFlow/LangGraph ``messages`` tuple intact for the client.
|
|
|
|
The workflow event envelope adds run/node identity, but the nested value is
|
|
deliberately the same two-item shape consumed by DeerFlow's chat UI:
|
|
``[message, metadata]``. Do not project tool args, result content, or
|
|
``additional_kwargs`` here; those fields are what power rich tool and user
|
|
assistance cards after a live stream reconnects.
|
|
"""
|
|
|
|
def encode(value: Any) -> Any:
|
|
if hasattr(value, "model_dump"):
|
|
return encode(value.model_dump(mode="json"))
|
|
if isinstance(value, dict):
|
|
return {str(key): encode(item) for key, item in value.items()}
|
|
if isinstance(value, (list, tuple)):
|
|
return [encode(item) for item in value]
|
|
# Force a JSON-compatible copy without discarding or truncating any
|
|
# field. LangGraph metadata is normally already JSON-native; ``str``
|
|
# is only a serialization fallback for provider-specific objects.
|
|
try:
|
|
json.dumps(value, ensure_ascii=False)
|
|
except (TypeError, ValueError):
|
|
return str(value)
|
|
return value
|
|
|
|
return encode(chunk if isinstance(chunk, tuple) else (chunk, {}))
|
|
|
|
|
|
def _resume_message(payload: dict[str, Any]) -> str:
|
|
"""Represent a workflow resume as the next user turn in the same thread."""
|
|
return (
|
|
"用户已回复你刚才的协助请求。请基于已有对话继续当前任务，不要重复询问已经回答的信息。\n"
|
|
"<workflow-user-assistance-response>\n"
|
|
f"{json.dumps(payload, ensure_ascii=False, separators=(',', ':'), default=str)}\n"
|
|
"</workflow-user-assistance-response>"
|
|
)
|
|
|
|
|
|
def _clarification_pending(message: ToolMessage) -> dict[str, Any] | None:
|
|
"""Identify DeerFlow's native ``ask_clarification`` tool result.
|
|
|
|
The complete card payload continues to live in the untouched ToolMessage
|
|
sent through ``node.message``. The pause descriptor only carries the
|
|
stable tool-call identity required to reconnect it to this node run.
|
|
"""
|
|
if str(getattr(message, "name", "") or "") != "ask_clarification":
|
|
return None
|
|
additional_kwargs = getattr(message, "additional_kwargs", None)
|
|
clarification = (
|
|
additional_kwargs.get("clarification")
|
|
if isinstance(additional_kwargs, dict)
|
|
else None
|
|
)
|
|
if not isinstance(clarification, dict):
|
|
return None
|
|
call_id = str(getattr(message, "tool_call_id", "") or getattr(message, "id", "") or "")
|
|
question = str(clarification.get("question") or "")
|
|
return {
|
|
"toolCallId": call_id,
|
|
"prompt": question or "智能体需要你补充信息后才能继续。",
|
|
"actions": ["submit"],
|
|
}
|
|
|
|
|
|
class WorkflowAgentRunner:
|
|
def __init__(self, app: Any) -> None:
|
|
self._app = app
|
|
|
|
async def run_agent(
|
|
self,
|
|
*,
|
|
run_id: str,
|
|
node_id: str,
|
|
owner_id: str,
|
|
agent_id: str,
|
|
prompt: str,
|
|
skill_names: list[str] | None = None,
|
|
thread_mode: str = "isolated_per_node",
|
|
cancel: Any = None,
|
|
on_delta: Callable[[str], Awaitable[None]] | None = None,
|
|
on_tool: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
|
|
on_message: Callable[[list[Any]], Awaitable[None]] | None = None,
|
|
resume_payload: dict[str, Any] | None = None,
|
|
model_name: str | None = None,
|
|
disable_tools: bool = False,
|
|
thinking_enabled: bool | None = None,
|
|
force_disable_thinking: bool = False,
|
|
) -> dict[str, Any]:
|
|
from deerflow.agents.lead_agent.agent import make_lead_agent
|
|
from deerflow.config.app_config import get_app_config
|
|
|
|
thread_id = _thread_id(run_id, node_id, thread_mode)
|
|
configurable: dict[str, Any] = {
|
|
"thread_id": thread_id,
|
|
"agent_id": agent_id,
|
|
"agent_name": agent_id,
|
|
"user_id": owner_id,
|
|
"model_name": model_name or "",
|
|
# Workflow nodes are non-interactive: no plan mode, no clarification.
|
|
"is_plan_mode": False,
|
|
# Control-plane callers (such as candidate planning) may need a
|
|
# model's reasoning but must not expose any callable capability.
|
|
"disable_tools": disable_tools,
|
|
"thinking_force_disabled": force_disable_thinking,
|
|
"workflow_run_id": run_id,
|
|
"workflow_node_id": node_id,
|
|
}
|
|
if thinking_enabled is not None:
|
|
configurable["thinking_enabled"] = thinking_enabled
|
|
if skill_names:
|
|
configurable["skill_names"] = list(skill_names)
|
|
# Tool-using research agents need more than the old chat-sized 60
|
|
# super-steps (each model → tool exchange is roughly two steps).
|
|
config = {
|
|
"configurable": configurable,
|
|
"recursion_limit": get_app_config().workflows.agent_recursion_limit,
|
|
}
|
|
|
|
try:
|
|
agent = make_lead_agent(config)
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("failed to build agent %s for workflow node %s", agent_id, node_id, exc_info=True)
|
|
raise WorkflowError(
|
|
"WORKFLOW_RESOURCE_MISSING",
|
|
"引用的智能体不存在或配置无效",
|
|
node_id=node_id,
|
|
details={"agentId": agent_id},
|
|
) from exc
|
|
|
|
checkpointer = getattr(self._app.state, "checkpointer", None)
|
|
if checkpointer is not None:
|
|
agent.checkpointer = checkpointer
|
|
store = getattr(self._app.state, "store", None)
|
|
if store is not None:
|
|
agent.store = store
|
|
|
|
seen_tool_calls: set[str] = set()
|
|
finished_tool_calls: set[str] = set()
|
|
tool_names: dict[str, str] = {}
|
|
final_state: dict[str, Any] | None = None
|
|
streamed: dict[str, list[str]] = {}
|
|
last_message_id = ""
|
|
open_reasoning_message_ids: set[str] = set()
|
|
|
|
async def emit_display_delta(text: str) -> None:
|
|
if on_delta is not None and text:
|
|
await on_delta(text)
|
|
|
|
async def close_reasoning(*, except_message_id: str | None = None) -> None:
|
|
"""Keep thinking blocks well-formed across tool/model turns."""
|
|
|
|
for message_id in tuple(open_reasoning_message_ids):
|
|
if message_id == except_message_id:
|
|
continue
|
|
open_reasoning_message_ids.remove(message_id)
|
|
await emit_display_delta("</think>")
|
|
|
|
initial_messages = [
|
|
{
|
|
"role": "user",
|
|
"content": _resume_message(resume_payload)
|
|
if resume_payload is not None
|
|
else prompt,
|
|
}
|
|
]
|
|
awaiting_input: dict[str, Any] | None = None
|
|
|
|
async for mode, chunk in agent.astream(
|
|
{"messages": initial_messages},
|
|
config=config,
|
|
stream_mode=["values", "messages"],
|
|
):
|
|
if cancel is not None and getattr(cancel, "cancelled", False):
|
|
raise WorkflowError("WORKFLOW_CANCELLED", "运行已被取消", node_id=node_id)
|
|
if mode == "values":
|
|
if isinstance(chunk, dict):
|
|
final_state = chunk
|
|
continue
|
|
if mode != "messages":
|
|
continue
|
|
if on_message is not None:
|
|
await on_message(_native_message_frame(chunk))
|
|
message = chunk[0] if isinstance(chunk, tuple) and chunk else chunk
|
|
if isinstance(message, ToolMessage):
|
|
# Tool result for an earlier call: close its step in the event log.
|
|
await close_reasoning()
|
|
call_id = str(getattr(message, "tool_call_id", "") or "")
|
|
if call_id and call_id not in finished_tool_calls:
|
|
finished_tool_calls.add(call_id)
|
|
if on_tool is not None:
|
|
await on_tool(
|
|
{
|
|
"phase": "finished",
|
|
"id": call_id,
|
|
"name": tool_names.get(call_id) or getattr(message, "name", "") or "",
|
|
"ok": getattr(message, "status", None) != "error",
|
|
"result": getattr(message, "content", ""),
|
|
}
|
|
)
|
|
awaiting_input = _clarification_pending(message)
|
|
if awaiting_input is not None:
|
|
break
|
|
continue
|
|
if not isinstance(message, AIMessage):
|
|
continue
|
|
message_id = str(getattr(message, "id", "") or "__workflow_reasoning__")
|
|
reasoning = _extract_reasoning(message)
|
|
if reasoning:
|
|
if message_id not in open_reasoning_message_ids:
|
|
open_reasoning_message_ids.add(message_id)
|
|
await emit_display_delta("<think>")
|
|
await emit_display_delta(reasoning)
|
|
text = _extract_text(getattr(message, "content", ""))
|
|
if text:
|
|
await close_reasoning()
|
|
streamed.setdefault(message_id, []).append(text)
|
|
last_message_id = message_id
|
|
await emit_display_delta(text)
|
|
for call in getattr(message, "tool_calls", None) or []:
|
|
call_id = str(call.get("id") or "")
|
|
if not call_id or call_id in seen_tool_calls:
|
|
continue
|
|
seen_tool_calls.add(call_id)
|
|
tool_names[call_id] = str(call.get("name") or "")
|
|
if on_tool is not None:
|
|
await on_tool(
|
|
{
|
|
"phase": "started",
|
|
"id": call_id,
|
|
"name": tool_names[call_id],
|
|
"args": call.get("args"),
|
|
}
|
|
)
|
|
|
|
# A reasoning-only turn that delegates to a tool must end before
|
|
# the tool result and next model turn arrive. Otherwise a later
|
|
# answer could be swallowed by one long unclosed `<think>` block.
|
|
if getattr(message, "tool_calls", None):
|
|
await close_reasoning()
|
|
|
|
await close_reasoning()
|
|
|
|
if awaiting_input is not None:
|
|
return {
|
|
"awaiting_input": awaiting_input,
|
|
"thread_id": thread_id,
|
|
"messages": [],
|
|
"artifacts": self._artifacts(final_state),
|
|
"usage": {},
|
|
}
|
|
|
|
final_text = "".join(streamed.get(last_message_id, ())) or self._final_text(final_state)
|
|
if not final_text.strip():
|
|
raise WorkflowError(
|
|
"WORKFLOW_AGENT_FAILED",
|
|
"智能体没有返回任何内容",
|
|
retryable=True,
|
|
node_id=node_id,
|
|
)
|
|
return {
|
|
"text": final_text,
|
|
"thread_id": thread_id,
|
|
"messages": [],
|
|
"artifacts": self._artifacts(final_state),
|
|
"usage": {},
|
|
}
|
|
|
|
def _final_text(self, state: dict[str, Any] | None) -> str:
|
|
for message in reversed((state or {}).get("messages") or []):
|
|
if isinstance(message, AIMessage):
|
|
text = _extract_text(getattr(message, "content", "")).strip()
|
|
if text:
|
|
return text
|
|
return ""
|
|
|
|
def _artifacts(self, state: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
raw = (state or {}).get("artifacts")
|
|
if not isinstance(raw, list):
|
|
return []
|
|
out: list[dict[str, Any]] = []
|
|
for item in raw:
|
|
if isinstance(item, dict) and item.get("name"):
|
|
out.append(
|
|
{
|
|
"name": str(item.get("name")),
|
|
"path": str(item.get("path") or ""),
|
|
"mime_type": str(item.get("mime_type") or item.get("mimeType") or ""),
|
|
"size_bytes": item.get("size_bytes") or item.get("sizeBytes"),
|
|
"preview": item.get("preview"),
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
__all__ = ["WorkflowAgentRunner"]
|