From 00270a5c01e77ef7a521686dd014fc0504bd6945 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Sat, 11 Jul 2026 22:26:31 +0800 Subject: [PATCH] Add Dify integration and enhance workflow node specifications - Introduce new fields `dify_api_url` and `dify_api_key` in `AssistantConfig` for Dify API integration. - Update `requirements.txt` to include `dify-client-python` for Dify SDK support. - Modify `config_resolver` to handle Dify connection information. - Add a new `globalNode` type in workflow specifications to provide unified settings across workflows. - Enhance node specifications with additional constraints and default values for better configuration management. - Update frontend components to support the new `globalNode` type and its properties, improving workflow editor functionality. --- backend/models.py | 2 + backend/requirements.txt | 3 + backend/schemas.py | 4 +- backend/services/brains/__init__.py | 4 +- backend/services/brains/base.py | 111 +++++- backend/services/brains/dify_brain.py | 61 +++ backend/services/brains/dify_llm.py | 127 +++++++ backend/services/brains/fastgpt_brain.py | 15 +- backend/services/brains/internal_brain.py | 37 -- backend/services/brains/prompt_brain.py | 106 ++++++ backend/services/brains/registry.py | 39 +- backend/services/brains/workflow_brain.py | 188 ++++++++++ backend/services/config_resolver.py | 2 + backend/services/node_specs.py | 105 +++++- backend/services/pipecat/call_lifecycle.py | 60 +++ backend/services/pipecat/pipeline.py | 351 +++--------------- backend/services/pipecat/service_factory.py | 4 +- backend/services/workflow_engine.py | 26 +- backend/tests/test_brains.py | 303 +++++++++++++++ .../src/components/workflow/GenericNode.tsx | 1 + .../components/workflow/WorkflowEditor.tsx | 74 +++- frontend/src/components/workflow/specs.ts | 58 ++- frontend/src/lib/api.ts | 3 + 23 files changed, 1270 insertions(+), 414 deletions(-) create mode 100644 backend/services/brains/dify_brain.py create mode 100644 backend/services/brains/dify_llm.py delete mode 100644 backend/services/brains/internal_brain.py create mode 100644 backend/services/brains/prompt_brain.py create mode 100644 backend/services/brains/workflow_brain.py create mode 100644 backend/services/pipecat/call_lifecycle.py create mode 100644 backend/tests/test_brains.py diff --git a/backend/models.py b/backend/models.py index 5209de6..5097682 100644 --- a/backend/models.py +++ b/backend/models.py @@ -79,6 +79,8 @@ class AssistantConfig(BaseModel): graph: dict = {} # 外部托管类型(fastgpt/dify/opencode)的连接信息:context/KB/tools 由对方服务端接管。 + dify_api_url: str = "" + dify_api_key: str = "" fastgpt_api_url: str = "" fastgpt_api_key: str = "" fastgpt_app_id: str = "" diff --git a/backend/requirements.txt b/backend/requirements.txt index 9b06227..516d0ee 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -8,6 +8,9 @@ Pillow>=11.1.0,<13 # FastGPT 类型助手:本地 SDK(包 /api/v1/chat/completions 流式 + chatId 会话) fastgpt-client @ file:///Users/wangx/Code/AI-VideoAssistant-Project/fastgpt-python-sdk +# Dify 类型助手:异步流式 Runtime API SDK +dify-client-python==1.0.3 + fastapi httpx uvicorn[standard] diff --git a/backend/schemas.py b/backend/schemas.py index 562bef8..0ed1acf 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -24,10 +24,10 @@ ToolParameterLocation = Literal["path", "query", "body", "header"] # 外部应用类型:其 config.apiKey 是该助手私有密钥,读时打码 / 写时哨兵 EXTERNAL_TYPES = {"dify", "fastgpt", "opencode"} -# 支持 realtime(语音到语音)的类型;外部托管大脑只能走 cascade。 +# MVP 仅 PromptBrain 支持 realtime;Workflow 和外部托管大脑只走 pipeline。 # 与 services.brains 各 BrainSpec.supported_runtime_modes 对齐(此处独立声明, # 避免 HTTP schema 层为做校验而引入 pipecat 重依赖)。 -REALTIME_CAPABLE_TYPES = {"prompt", "workflow"} +REALTIME_CAPABLE_TYPES = {"prompt"} class CamelModel(BaseModel): diff --git a/backend/services/brains/__init__.py b/backend/services/brains/__init__.py index 8613af0..f4069b6 100644 --- a/backend/services/brains/__init__.py +++ b/backend/services/brains/__init__.py @@ -1,6 +1,6 @@ """可插拔的「大脑」:把不同助手类型在运行时的差异收口到各自的 Brain 实现。""" -from services.brains.base import Brain, BrainSpec +from services.brains.base import Brain, BrainRuntime, BrainSpec from services.brains.registry import SPECS, build_brain -__all__ = ["Brain", "BrainSpec", "SPECS", "build_brain"] +__all__ = ["Brain", "BrainRuntime", "BrainSpec", "SPECS", "build_brain"] diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index 44bd293..de1e6d5 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -1,46 +1,117 @@ -"""「大脑」抽象:把不同助手类型(prompt/workflow/fastgpt/…)在运行时的差异收口。 +"""Conversation-brain contracts shared by every assistant type. -cascade 管线骨架对所有类型一致(STT → LLM 槽 → TTS),变化的只有: - - 谁产出助手文本(LLM 槽里塞什么)——build_llm - - 开场白来源(静态 / 外部异步拉取)——greeting - - 对话上下文归谁维护——spec.owns_context - - 是否支持 realtime——spec.supported_runtime_modes - -阶段 1 只抽到「够 fastgpt 用」的程度;workflow 编排仍内联在 pipeline.py, -待阶段 2 再搬进 WorkflowBrain 收口。 +Brain selects who owns reasoning and conversation state. The Pipecat pipeline +still owns media transport, STT/TTS, transcript delivery, and interruption +semantics. This keeps assistant-specific orchestration out of pipeline.py +without coupling brains to Pipecat internals more than necessary. """ from __future__ import annotations +from collections.abc import Awaitable, Callable from dataclasses import dataclass -from typing import Protocol, runtime_checkable +from typing import Any, Protocol, runtime_checkable from models import AssistantConfig +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.frames.frames import Frame from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameProcessor @dataclass(frozen=True) class BrainSpec: - """类型元数据。单一来源,供运行时门控与上下文归属决策复用。""" + """Static capabilities used by validation and runtime dispatch.""" type: str supported_runtime_modes: frozenset[str] - # True:由本服务维护 LLMContext(prompt/workflow); - # False:上下文/知识库/工具由外部服务端接管(fastgpt/dify),本地不写 context。 + # False means context, knowledge bases, and tools live on an external agent. owns_context: bool -@runtime_checkable -class Brain(Protocol): - """每通电话 new 一个实例(可持有 chatId / 当前节点等会话状态)。""" +class CallEndPort(Protocol): + """Small call-lifecycle surface available to a brain.""" + + @property + def ending(self) -> bool: ... + + def begin(self, reason: str) -> None: ... + + def arm_after_speech(self) -> None: ... + + async def finish(self) -> None: ... + + +@dataclass(frozen=True) +class BrainRuntime: + """Pipeline-owned capabilities injected into one brain session.""" + + context: LLMContext + llm: Any + queue_frame: Callable[[Frame], Awaitable[None]] + set_system_prompt: Callable[[str], None] + set_tools: Callable[[list[FunctionSchema] | None], None] + call_end: CallEndPort + + +class BaseBrain: + """No-op lifecycle defaults for brains without local orchestration.""" spec: BrainSpec async def greeting(self, cfg: AssistantConfig) -> str: - """开场白。内部类型通常直接用 cfg.greeting;外部类型异步拉取后端配置。""" - ... + return cfg.greeting + + def system_prompt(self, cfg: AssistantConfig) -> str: + return cfg.prompt if self.spec.owns_context else "" def build_llm(self, cfg: AssistantConfig, context: LLMContext) -> FrameProcessor: - """返回丢进管线 LLM 槽位的帧处理器(标准 LLMService 或外部托管的伪 LLM)。""" - ... + raise NotImplementedError + + async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: + """Register tools and initialize per-call orchestration.""" + + async def on_connected(self) -> None: + """Handle a connected client after the common greeting is queued.""" + + def record_user_message(self, content: str) -> None: + """Observe a committed user message for brain-owned routing state.""" + + async def on_assistant_text_start(self, turn_id: str) -> None: + """Observe the start of a generated assistant turn.""" + + async def on_assistant_text_end( + self, + turn_id: str, + content: str, + interrupted: bool, + ) -> None: + """Observe the completion of a generated assistant turn.""" + + +@runtime_checkable +class Brain(Protocol): + """One instance per call; implementations may keep conversation state.""" + + spec: BrainSpec + + async def greeting(self, cfg: AssistantConfig) -> str: ... + + def system_prompt(self, cfg: AssistantConfig) -> str: ... + + def build_llm(self, cfg: AssistantConfig, context: LLMContext) -> FrameProcessor: ... + + async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: ... + + async def on_connected(self) -> None: ... + + def record_user_message(self, content: str) -> None: ... + + async def on_assistant_text_start(self, turn_id: str) -> None: ... + + async def on_assistant_text_end( + self, + turn_id: str, + content: str, + interrupted: bool, + ) -> None: ... diff --git a/backend/services/brains/dify_brain.py b/backend/services/brains/dify_brain.py new file mode 100644 index 0000000..bee51e1 --- /dev/null +++ b/backend/services/brains/dify_brain.py @@ -0,0 +1,61 @@ +"""Dify-hosted brain: prompt, workflow, tools, and context live in Dify.""" + +from __future__ import annotations + +from uuid import uuid4 + +from dify_client import AsyncClient +from loguru import logger +from models import AssistantConfig +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.frame_processor import FrameProcessor + +from services.brains.base import BaseBrain, BrainSpec +from services.brains.dify_llm import DifyLLMService, normalize_api_base + + +class DifyBrain(BaseBrain): + spec = BrainSpec( + type="dify", + supported_runtime_modes=frozenset({"pipeline"}), + owns_context=False, + ) + + def __init__(self): + self._user_id = f"ai-video-{uuid4().hex}" + self._client: AsyncClient | None = None + + def _get_client(self, cfg: AssistantConfig) -> AsyncClient: + if self._client is None: + if not cfg.dify_api_key: + raise ValueError("缺少 Dify Agent apiKey") + self._client = AsyncClient( + api_key=cfg.dify_api_key, + api_base=normalize_api_base(cfg.dify_api_url), + ) + return self._client + + async def greeting(self, cfg: AssistantConfig) -> str: + """Use Dify's opening statement, with the local greeting as fallback.""" + try: + api_base = normalize_api_base(cfg.dify_api_url) + response = await self._get_client(cfg).arequest( + f"{api_base}/parameters", + "GET", + params={"user": self._user_id}, + timeout=15.0, + ) + opening = str(response.json().get("opening_statement") or "").strip() + return opening or cfg.greeting + except ValueError: + raise + except Exception as exc: # noqa: BLE001 - greeting failure should not block a call + logger.warning(f"Dify 获取开场白失败,回退 cfg.greeting: {exc}") + return cfg.greeting + + def build_llm(self, cfg: AssistantConfig, context: LLMContext) -> FrameProcessor: + return DifyLLMService( + cfg, + client=self._get_client(cfg), + user_id=self._user_id, + ) diff --git a/backend/services/brains/dify_llm.py b/backend/services/brains/dify_llm.py new file mode 100644 index 0000000..849d36e --- /dev/null +++ b/backend/services/brains/dify_llm.py @@ -0,0 +1,127 @@ +"""Dify chat applications exposed as a Pipecat LLM processor.""" + +from __future__ import annotations + +from uuid import uuid4 + +from dify_client import AsyncClient, models +from loguru import logger +from models import AssistantConfig +from pipecat.frames.frames import ( + Frame, + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMTextFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.llm_service import LLMService +from pipecat.services.settings import LLMSettings + + +def normalize_api_base(url: str) -> str: + """Accept a Dify host, /v1 base URL, or full chat endpoint.""" + base = (url or "https://api.dify.ai").strip().rstrip("/") + if base.endswith("/chat-messages"): + base = base[: -len("/chat-messages")] + if not base.endswith("/v1"): + base = f"{base}/v1" + return base + + +def last_user_text(messages: list[dict]) -> str: + for message in reversed(messages or []): + if message.get("role") != "user": + continue + content = message.get("content") + if isinstance(content, str): + return content + if isinstance(content, list): + return "".join( + str(part.get("text") or "") + for part in content + if isinstance(part, dict) + ) + return "" + + +class DifyLLMService(LLMService): + """Stream Dify answer events into Pipecat's standard text frames.""" + + def __init__( + self, + cfg: AssistantConfig, + *, + client: AsyncClient | None = None, + user_id: str | None = None, + ): + super().__init__( + settings=LLMSettings( + model=None, + system_instruction=None, + temperature=None, + max_tokens=None, + top_p=None, + top_k=None, + frequency_penalty=None, + presence_penalty=None, + seed=None, + filter_incomplete_user_turns=None, + user_turn_completion_config=None, + ) + ) + self._client = client or AsyncClient( + api_key=cfg.dify_api_key, + api_base=normalize_api_base(cfg.dify_api_url), + ) + self._user_id = user_id or f"ai-video-{uuid4().hex}" + self._conversation_id = "" + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if not isinstance(frame, LLMContextFrame): + await self.push_frame(frame, direction) + return + + user_text = last_user_text(frame.context.get_messages()) + if not user_text: + return + + await self.push_frame(LLMFullResponseStartFrame()) + try: + request = models.ChatRequest( + query=user_text, + inputs={}, + user=self._user_id, + response_mode=models.ResponseMode.STREAMING, + conversation_id=self._conversation_id, + auto_generate_name=False, + ) + events = await self._client.achat_messages(request, timeout=120.0) + async for event in events: + conversation_id = getattr(event, "conversation_id", "") + if conversation_id: + self._conversation_id = conversation_id + + event_name = str(getattr(event, "event", "")) + if event_name == "error": + logger.error( + "Dify 流式错误: " + f"code={getattr(event, 'code', '')} " + f"message={getattr(event, 'message', '')}" + ) + continue + text = ( + getattr(event, "answer", "") + if event_name in {"message", "agent_message"} + else "" + ) + if event_name == "text_chunk": + text = getattr(getattr(event, "data", None), "text", "") + if text: + await self.push_frame(LLMTextFrame(text)) + except Exception as exc: # noqa: BLE001 - one failed turn must not kill the call + logger.error(f"Dify 调用失败: {exc}") + finally: + await self.push_frame(LLMFullResponseEndFrame()) diff --git a/backend/services/brains/fastgpt_brain.py b/backend/services/brains/fastgpt_brain.py index b3e8c96..deb8882 100644 --- a/backend/services/brains/fastgpt_brain.py +++ b/backend/services/brains/fastgpt_brain.py @@ -15,7 +15,7 @@ from models import AssistantConfig from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameProcessor -from services.brains.base import BrainSpec +from services.brains.base import BaseBrain, BrainSpec from services.brains.fastgpt_llm import FastGPTLLMService, normalize_base_url @@ -38,15 +38,16 @@ def _extract_welcome(payload: Any) -> str: return "" -class FastGPTBrain: +class FastGPTBrain(BaseBrain): def __init__(self): - self.spec = BrainSpec( - type="fastgpt", - supported_runtime_modes=frozenset({"pipeline"}), - owns_context=False, - ) self._chat_id = uuid4().hex + spec = BrainSpec( + type="fastgpt", + supported_runtime_modes=frozenset({"pipeline"}), + owns_context=False, + ) + async def greeting(self, cfg: AssistantConfig) -> str: """优先用 FastGPT 后台配置的开场白;无 app_id 或取不到时回退 cfg.greeting。""" if not cfg.fastgpt_app_id: diff --git a/backend/services/brains/internal_brain.py b/backend/services/brains/internal_brain.py deleted file mode 100644 index 605b026..0000000 --- a/backend/services/brains/internal_brain.py +++ /dev/null @@ -1,37 +0,0 @@ -"""内部 LLM 大脑:prompt 与 workflow。 - -二者都用本地维护的 LLMContext + OpenAI 兼容 LLM,支持 cascade 与 realtime。 -workflow 的图编排(切提示/转移工具/node-active)阶段 1 仍内联在 pipeline.py, -这里只负责提供 LLM 槽位与元数据,行为与改造前完全一致。 -""" - -from __future__ import annotations - -from models import AssistantConfig -from pipecat.processors.aggregators.llm_context import LLMContext -from pipecat.processors.frame_processor import FrameProcessor - -from services.brains.base import BrainSpec - -_CASCADE_AND_REALTIME = frozenset({"pipeline", "realtime"}) - - -class InternalBrain: - """prompt / workflow 共用。""" - - def __init__(self, brain_type: str): - self.spec = BrainSpec( - type=brain_type, - supported_runtime_modes=_CASCADE_AND_REALTIME, - owns_context=True, - ) - - async def greeting(self, cfg: AssistantConfig) -> str: - # 内部类型的开场白由 pipeline.py 现有逻辑(workflow 起始节点 / cfg.greeting)决定, - # 该方法仅为满足 Brain 协议,实际不在内部路径上被调用。 - return cfg.greeting - - def build_llm(self, cfg: AssistantConfig, context: LLMContext) -> FrameProcessor: - from services.pipecat.service_factory import create_llm - - return create_llm(cfg) diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py new file mode 100644 index 0000000..24448cc --- /dev/null +++ b/backend/services/brains/prompt_brain.py @@ -0,0 +1,106 @@ +"""Local prompt assistant, including prompt-only reusable tools.""" + +from __future__ import annotations + +from uuid import uuid4 + +from models import AssistantConfig +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.frames.frames import OutputTransportMessageUrgentFrame, TTSSpeakFrame +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.frame_processor import FrameProcessor +from pipecat.services.llm_service import ( + FunctionCallParams, + FunctionCallResultProperties, +) +from pipecat.utils.time import time_now_iso8601 + +from services.brains.base import BaseBrain, BrainRuntime, BrainSpec + + +class PromptBrain(BaseBrain): + spec = BrainSpec( + type="prompt", + supported_runtime_modes=frozenset({"pipeline", "realtime"}), + owns_context=True, + ) + + def build_llm(self, cfg: AssistantConfig, context: LLMContext) -> FrameProcessor: + from services.pipecat.service_factory import create_llm + + return create_llm(cfg) + + async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: + schemas: list[FunctionSchema] = [] + for tool in cfg.tools: + if tool.type != "end_call": + continue + schema, handler = self._make_end_call_tool(tool, runtime) + schemas.append(schema) + runtime.llm.register_function(tool.function_name, handler) + runtime.set_tools(schemas) + + @staticmethod + def _make_end_call_tool(tool, runtime: BrainRuntime): + config = (tool.definition or {}).get("config") or {} + message_type = str(config.get("message_type") or "none") + custom_message = str(config.get("custom_message") or "").strip() + capture_reason = bool(config.get("capture_reason", True)) + + async def end_call(params: FunctionCallParams) -> None: + reason = str(params.arguments.get("reason") or "end_call_tool").strip() + runtime.call_end.begin(reason) + await params.result_callback( + {"status": "success", "action": "ending_call"}, + properties=FunctionCallResultProperties(run_llm=False), + ) + + if message_type != "custom" or not custom_message: + await runtime.call_end.finish() + return + + turn_id = uuid4().hex + timestamp = time_now_iso8601() + for message in ( + { + "type": "assistant-text-start", + "turn_id": turn_id, + "timestamp": timestamp, + }, + { + "type": "assistant-text-delta", + "turn_id": turn_id, + "delta": custom_message, + }, + { + "type": "assistant-text-end", + "turn_id": turn_id, + "content": custom_message, + "interrupted": False, + }, + ): + await runtime.queue_frame( + OutputTransportMessageUrgentFrame(message=message) + ) + runtime.call_end.arm_after_speech() + await runtime.queue_frame( + TTSSpeakFrame(custom_message, append_to_context=False) + ) + + properties = ( + { + "reason": { + "type": "string", + "description": "结束本次通话的简短原因。", + } + } + if capture_reason + else {} + ) + schema = FunctionSchema( + name=tool.function_name, + description=tool.description or "结束当前通话。", + properties=properties, + required=["reason"] if capture_reason else [], + ) + return schema, end_call diff --git a/backend/services/brains/registry.py b/backend/services/brains/registry.py index f8a3a0e..cfa3fe1 100644 --- a/backend/services/brains/registry.py +++ b/backend/services/brains/registry.py @@ -1,25 +1,40 @@ -"""类型 → Brain 工厂。新增一种大脑 = 加一个 brain 文件 + 在此注册一行。""" +"""Explicit assistant-type registry. Unsupported types never silently degrade.""" from __future__ import annotations +from collections.abc import Callable + from models import AssistantConfig from services.brains.base import Brain, BrainSpec +from services.brains.dify_brain import DifyBrain from services.brains.fastgpt_brain import FastGPTBrain -from services.brains.internal_brain import InternalBrain +from services.brains.prompt_brain import PromptBrain +from services.brains.workflow_brain import WorkflowBrain + + +def _workflow(cfg: AssistantConfig) -> Brain: + return WorkflowBrain(cfg.graph) + + +BRAIN_FACTORIES: dict[str, Callable[[AssistantConfig], Brain]] = { + "prompt": lambda _cfg: PromptBrain(), + "workflow": _workflow, + "dify": lambda _cfg: DifyBrain(), + "fastgpt": lambda _cfg: FastGPTBrain(), +} -# 各类型的元数据(供 schema 校验 / realtime 门控复用,无需实例化 Brain)。 SPECS: dict[str, BrainSpec] = { - "prompt": InternalBrain("prompt").spec, - "workflow": InternalBrain("workflow").spec, - "fastgpt": FastGPTBrain().spec, + "prompt": PromptBrain.spec, + "workflow": WorkflowBrain.spec, + "dify": DifyBrain.spec, + "fastgpt": FastGPTBrain.spec, } def build_brain(cfg: AssistantConfig) -> Brain: - """按 cfg.type 构造每通电话的 Brain 实例(未知类型回退 prompt)。""" - if cfg.type == "fastgpt": - return FastGPTBrain() - if cfg.type in ("prompt", "workflow"): - return InternalBrain(cfg.type) - return InternalBrain("prompt") + try: + factory = BRAIN_FACTORIES[cfg.type] + except KeyError as exc: + raise ValueError(f"尚未实现的助手类型: {cfg.type}") from exc + return factory(cfg) diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py new file mode 100644 index 0000000..ec90ce3 --- /dev/null +++ b/backend/services/brains/workflow_brain.py @@ -0,0 +1,188 @@ +"""Local graph-driven workflow assistant and its per-call state.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from loguru import logger +from models import AssistantConfig +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.frames.frames import OutputTransportMessageUrgentFrame, TTSSpeakFrame +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.frame_processor import FrameProcessor + +from services.brains.base import BaseBrain, BrainRuntime, BrainSpec +from services.workflow_engine import WorkflowEngine + + +@dataclass +class WorkflowState: + current: str + ended: bool = False + turns_in_node: int = 0 + end_turn_id: str | None = None + + +class WorkflowBrain(BaseBrain): + spec = BrainSpec( + type="workflow", + supported_runtime_modes=frozenset({"pipeline"}), + owns_context=True, + ) + _FALLBACK_AFTER_TURNS = 2 + + def __init__(self, graph: dict[str, Any]): + self._engine = WorkflowEngine(graph or {}) + if not self._engine.has_graph() or not self._engine.start_id: + raise ValueError("WorkflowBrain 缺少有效的 startCall 节点") + self._state = WorkflowState(current=self._engine.start_id) + self._history: list[dict[str, str]] = [] + self._cfg: AssistantConfig | None = None + self._runtime: BrainRuntime | None = None + + async def greeting(self, cfg: AssistantConfig) -> str: + return self._engine.greeting() or cfg.greeting + + def system_prompt(self, cfg: AssistantConfig) -> str: + return self._engine.system_prompt_for(self._state.current) + + def build_llm(self, cfg: AssistantConfig, context: LLMContext) -> FrameProcessor: + from services.pipecat.service_factory import create_llm + + return create_llm(cfg) + + async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: + self._cfg = cfg + self._runtime = runtime + for edge in self._engine.edges: + if edge.get("target"): + runtime.llm.register_function( + self._engine.edge_fn_name(edge), + self._make_transition_handler(edge), + ) + self._apply_node(self._state.current) + logger.info( + f"工作流模式启用: 起始节点={self._engine.name(self._state.current)}" + ) + + async def on_connected(self) -> None: + await self._emit_node_active(self._state.current) + + def record_user_message(self, content: str) -> None: + if content: + self._history.append({"role": "user", "content": content}) + + async def on_assistant_text_start(self, turn_id: str) -> None: + if self._state.ended and self._state.end_turn_id is None: + self._state.end_turn_id = turn_id + + async def on_assistant_text_end( + self, + turn_id: str, + content: str, + interrupted: bool, + ) -> None: + if not content or interrupted: + return + self._history.append({"role": "assistant", "content": content}) + if turn_id == self._state.end_turn_id: + runtime = self._require_runtime() + runtime.call_end.begin("completed") + runtime.call_end.arm_after_speech() + elif not self._state.ended: + self._state.turns_in_node += 1 + await self._fallback_route() + + def _apply_node(self, node_id: str) -> None: + runtime = self._require_runtime() + runtime.set_system_prompt(self._engine.system_prompt_for(node_id)) + if self._engine.is_end(node_id): + runtime.set_tools([]) + return + runtime.set_tools( + [ + FunctionSchema( + name=self._engine.edge_fn_name(edge), + description=self._engine.edge_description(edge), + properties={}, + required=[], + ) + for edge in self._engine.outgoing(node_id) + ] + ) + + async def _go_to_node(self, target: str) -> None: + self._state.current = target + self._state.turns_in_node = 0 + if self._engine.is_end(target): + self._state.ended = True + await self._emit_node_active(target) + self._apply_node(target) + + async def _emit_node_active(self, node_id: str | None) -> None: + if node_id: + await self._require_runtime().queue_frame( + OutputTransportMessageUrgentFrame( + message={"type": "node-active", "nodeId": node_id} + ) + ) + + async def _speak_transition(self, edge: dict | None) -> None: + speech = self._engine.edge_transition_speech(edge) + if speech: + await self._require_runtime().queue_frame( + TTSSpeakFrame(speech, append_to_context=False) + ) + + def _make_transition_handler(self, edge: dict): + target = str(edge.get("target")) + + async def handler(params) -> None: + logger.info(f"LLM 触发转移 → {self._engine.name(target)}") + if not self._engine.is_end(target): + await self._speak_transition(edge) + await self._go_to_node(target) + await params.result_callback({"status": "ok"}) + + return handler + + async def _fallback_route(self) -> None: + if self._state.ended: + return + if self._state.turns_in_node < self._FALLBACK_AFTER_TURNS: + return + if not self._engine.outgoing(self._state.current): + return + + cfg = self._require_config() + target = await self._engine.route( + self._state.current, + self._history, + api_key=self._require(cfg.llm_api_key, "LLM apiKey"), + base_url=self._require(cfg.llm_base_url, "LLM apiUrl"), + model=self._require(cfg.model, "LLM modelId"), + ) + if target and target != self._state.current: + logger.info(f"文本兜底触发转移 → {self._engine.name(target)}") + if not self._engine.is_end(target): + await self._speak_transition( + self._engine.find_edge(self._state.current, target) + ) + await self._go_to_node(target) + + def _require_runtime(self) -> BrainRuntime: + if self._runtime is None: + raise RuntimeError("WorkflowBrain 尚未绑定 pipeline runtime") + return self._runtime + + def _require_config(self) -> AssistantConfig: + if self._cfg is None: + raise RuntimeError("WorkflowBrain 尚未初始化配置") + return self._cfg + + @staticmethod + def _require(value: str, label: str) -> str: + if value: + return value + raise ValueError(f"缺少模型资源配置: {label}") diff --git a/backend/services/config_resolver.py b/backend/services/config_resolver.py index 7aff562..25e0b83 100644 --- a/backend/services/config_resolver.py +++ b/backend/services/config_resolver.py @@ -139,6 +139,8 @@ async def resolve_runtime_config( # workflow 图:仅 workflow 类型非空,引擎据此启用图驱动对话 graph=(assistant.graph or {}) if assistant.type == "workflow" else {}, # 外部托管类型连接信息(DB 存真 key,直接注入) + dify_api_url=str(_value(agent_resource, "apiUrl", assistant.api_url)), + dify_api_key=_secret(agent_resource, "apiKey", assistant.api_key), fastgpt_api_url=str(_value(agent_resource, "apiUrl", assistant.api_url)), fastgpt_api_key=_secret(agent_resource, "apiKey", assistant.api_key), fastgpt_app_id=str(_value(agent_resource, "appId", assistant.app_id)), diff --git a/backend/services/node_specs.py b/backend/services/node_specs.py index 3ad9da2..cd562c8 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -1,6 +1,6 @@ """工作流节点规格 + 图校验(对齐 dograh 的 node-spec / GraphConstraints 思路)。 -当前实现 3 个核心节点:开始(startCall)/智能体(agentNode)/结束(endCall)。 +当前实现 4 个核心节点:开始(startCall)/智能体(agentNode)/结束(endCall)/全局(globalNode)。 本模块是「节点类型」的唯一事实源: - /api/node-types 接口直接吐这里的规格; - 助手保存时用这里的约束校验 workflow 图。 @@ -13,7 +13,7 @@ from __future__ import annotations from typing import Any # 规格版本号:节点定义有破坏性变更时 +1,前端可据此判断是否需要刷新缓存。 -SPEC_VERSION = "1" +SPEC_VERSION = "2" # 每个节点的图约束。None 表示不限制。 # min_incoming / max_incoming:入边数量 @@ -27,11 +27,28 @@ NODE_SPECS: list[dict[str, Any]] = [ "icon": "Play", "accent": "mint", "addable": False, - "constraints": {"minIncoming": 0, "maxIncoming": 0}, + "constraints": { + "minIncoming": 0, + "maxIncoming": 0, + "minInstances": 1, + "maxInstances": 1, + }, "fields": [ - {"key": "name", "label": "节点名称", "type": "text"}, - {"key": "greeting", "label": "开场白", "type": "textarea"}, - {"key": "prompt", "label": "节点提示词", "type": "textarea"}, + {"key": "name", "label": "节点名称", "type": "text", "default": "开始"}, + {"key": "greeting", "label": "开场白", "type": "textarea", "default": ""}, + {"key": "prompt", "label": "节点提示词", "type": "textarea", "default": ""}, + { + "key": "allowInterrupt", + "label": "允许用户打断", + "type": "switch", + "default": True, + }, + { + "key": "addGlobalPrompt", + "label": "应用全局提示词", + "type": "switch", + "default": True, + }, ], }, { @@ -44,9 +61,26 @@ NODE_SPECS: list[dict[str, Any]] = [ "addable": True, "constraints": {"minIncoming": 1}, "fields": [ - {"key": "name", "label": "节点名称", "type": "text"}, - {"key": "prompt", "label": "节点提示词", "type": "textarea", "required": True}, - {"key": "allowInterrupt", "label": "允许用户打断", "type": "switch"}, + {"key": "name", "label": "节点名称", "type": "text", "default": "智能体节点"}, + { + "key": "prompt", + "label": "节点提示词", + "type": "textarea", + "required": True, + "default": "", + }, + { + "key": "allowInterrupt", + "label": "允许用户打断", + "type": "switch", + "default": True, + }, + { + "key": "addGlobalPrompt", + "label": "应用全局提示词", + "type": "switch", + "default": True, + }, ], }, { @@ -59,8 +93,40 @@ NODE_SPECS: list[dict[str, Any]] = [ "addable": True, "constraints": {"minIncoming": 1, "minOutgoing": 0, "maxOutgoing": 0}, "fields": [ - {"key": "name", "label": "节点名称", "type": "text"}, - {"key": "prompt", "label": "结束语提示词", "type": "textarea"}, + {"key": "name", "label": "节点名称", "type": "text", "default": "结束"}, + {"key": "prompt", "label": "结束语提示词", "type": "textarea", "default": ""}, + { + "key": "addGlobalPrompt", + "label": "应用全局提示词", + "type": "switch", + "default": False, + }, + ], + }, + { + "name": "globalNode", + "displayName": "全局节点", + "category": "global_node", + "description": "为整个工作流提供统一的人设、语气和公共规则。无需连线,每个流程最多一个。", + "icon": "Globe2", + "accent": "lavender", + "addable": True, + "constraints": { + "minIncoming": 0, + "maxIncoming": 0, + "minOutgoing": 0, + "maxOutgoing": 0, + "maxInstances": 1, + }, + "fields": [ + {"key": "name", "label": "节点名称", "type": "text", "default": "全局设定"}, + { + "key": "prompt", + "label": "全局提示词", + "type": "textarea", + "required": True, + "default": "你是一个友好、专业的语音助手。请使用简短、自然、适合口语表达的句子。", + }, ], }, ] @@ -79,7 +145,7 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: 基础规则(对齐 dograh 的核心不变量): 1. 节点类型必须是已注册类型; 2. 有且仅有一个 startCall; - 3. 至少有一个 endCall; + 3. 至少有一个 endCall,全局节点最多一个; 4. 边的 source/target 必须指向存在的节点; 5. 入边/出边数量满足各节点类型的约束。 @@ -120,6 +186,21 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: if type_counts.get("endCall", 0) == 0: errors.append("工作流至少需要一个「结束」节点") + for node_type, spec in _SPEC_BY_NAME.items(): + # 开始节点上方已有更明确的中文错误提示,避免重复报错。 + if node_type == "startCall": + continue + constraints = spec["constraints"] + count = type_counts.get(node_type, 0) + _check_count( + errors, + count, + constraints, + "Instances", + node_type, + "实例", + ) + # 统计入边/出边 incoming: dict[str, int] = {nid: 0 for nid in node_ids} outgoing: dict[str, int] = {nid: 0 for nid in node_ids} diff --git a/backend/services/pipecat/call_lifecycle.py b/backend/services/pipecat/call_lifecycle.py new file mode 100644 index 0000000..27eb6c5 --- /dev/null +++ b/backend/services/pipecat/call_lifecycle.py @@ -0,0 +1,60 @@ +"""Shared call termination timing for prompt tools and workflow end nodes.""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable + +from loguru import logger +from pipecat.frames.frames import BotStartedSpeakingFrame, BotStoppedSpeakingFrame +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +class CallEndCoordinator: + """End immediately or after the currently armed closing speech finishes.""" + + def __init__(self, queue_end: Callable[[str], Awaitable[None]]): + self._queue_end = queue_end + self._ending = False + self._armed = False + self._speaking = False + self._finished = False + self._reason = "completed" + + @property + def ending(self) -> bool: + return self._ending + + def begin(self, reason: str) -> None: + self._ending = True + self._reason = reason or "completed" + + def arm_after_speech(self) -> None: + self._armed = True + + async def finish(self) -> None: + if self._finished: + return + self._finished = True + await self._queue_end(self._reason) + + async def observe(self, frame) -> None: + if isinstance(frame, BotStartedSpeakingFrame) and self._armed: + self._speaking = True + elif ( + isinstance(frame, BotStoppedSpeakingFrame) + and self._armed + and self._speaking + ): + logger.info("结束语播报完毕,挂断通话") + await self.finish() + + +class EndCallAfterSpeechProcessor(FrameProcessor): + def __init__(self, coordinator: CallEndCoordinator): + super().__init__() + self._coordinator = coordinator + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + await self.push_frame(frame, direction) + await self._coordinator.observe(frame) diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index d95315f..63fe50d 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -3,7 +3,7 @@ 关键设计:**transport 由调用方传入**,管线本身不关心是 WebRTC 还是 WS。 这就是"同时支持多种输出"的落点——加输出方式不用动这里。 -对应 dograh 的 pipeline_builder.py + run_pipeline.py(已砍掉 workflow 引擎/DB/录音/指标)。 +对话编排交给 Brain;本文件只保留共享媒体管线、输入输出和通话生命周期。 """ import asyncio @@ -16,21 +16,22 @@ from loguru import logger from models import AssistantConfig from openai import AsyncOpenAI from PIL import Image -from services.brains import build_brain +from services.brains import Brain, BrainRuntime, build_brain from services.conversation_history import ConversationRecorder +from services.pipecat.call_lifecycle import ( + CallEndCoordinator, + EndCallAfterSpeechProcessor, +) from services.pipecat.service_factory import ( create_realtime_service, create_stt, create_tts, ) -from services.workflow_engine import WorkflowEngine from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, EndFrame, InputTransportMessageFrame, InterruptionFrame, @@ -57,10 +58,7 @@ from pipecat.runner.utils import ( get_transport_client_id, maybe_capture_participant_camera, ) -from pipecat.services.llm_service import ( - FunctionCallParams, - FunctionCallResultProperties, -) +from pipecat.services.llm_service import FunctionCallParams from pipecat.turns.user_start import ( TranscriptionUserTurnStartStrategy, VADUserTurnStartStrategy, @@ -399,8 +397,7 @@ async def run_pipeline( cfg.runtimeMode == "realtime" and "realtime" not in brain.spec.supported_runtime_modes ): - logger.warning(f"类型 {cfg.type} 不支持 realtime,回退 cascade") - cfg.runtimeMode = "pipeline" + raise ValueError(f"类型 {cfg.type} 不支持 realtime 运行模式") if cfg.runtimeMode == "realtime": if vision_enabled: @@ -408,6 +405,7 @@ async def run_pipeline( await run_realtime_pipeline( transport, cfg, + brain=brain, assistant_id=assistant_id, channel=channel, ) @@ -416,45 +414,24 @@ async def run_pipeline( stt = create_stt(cfg) tts = create_tts(cfg) - # ---- workflow 图引擎(可选)---- - # 有节点图时按图驱动:开场白/系统提示来自起始节点,每轮回复后按条件路由。 - engine = WorkflowEngine(cfg.graph or {}) - workflow_active = engine.has_graph() - wf_state = { - # 开始节点本身就是会话节点(有自己的 prompt,可多轮),从它开始 - "current": engine.start_id if workflow_active else None, - "ended": False, - "turns_in_node": 0, - # 结束流程的精确计时:只在「结束节点自己的结束语」真正说完时挂断。 - "end_turn_id": None, # 结束节点回复的 turn_id(其 text_start 在 ended 之后) - "end_armed": False, # 结束语文本已生成完(已下发 data channel) - "end_speaking": False, # 结束语音频已开始播报 - "end_frame_queued": False, - } - call_end_state = { - "ending": False, - "armed": False, - "speaking": False, - "frame_queued": False, - "reason": "completed", - } - history: list[dict] = [] - # 当前节点没有可调用转移工具(全是空条件)时,才启用文本兜底路由 - FALLBACK_AFTER_TURNS = 2 + greeting = await brain.greeting(cfg) + system_content = brain.system_prompt(cfg) - if workflow_active: - greeting = engine.greeting() or cfg.greeting - system_content = engine.system_prompt_for(wf_state["current"]) - logger.info( - f"工作流模式启用: 起始节点={engine.name(wf_state['current'])}" + worker_holder: dict = {} + + async def queue_call_end(reason: str) -> None: + worker = worker_holder.get("worker") + if worker is None: + return + logger.info(f"结束通话: reason={reason}") + await worker.queue_frame( + OutputTransportMessageUrgentFrame( + message={"type": "call-ended", "reason": reason} + ) ) - elif brain.spec.owns_context: - greeting = cfg.greeting - system_content = cfg.prompt - else: - # 外部托管(fastgpt 等):开场白来自对方后台,系统提示/上下文不归我们维护 - greeting = await brain.greeting(cfg) - system_content = "" + await worker.queue_frame(EndFrame()) + + call_end = CallEndCoordinator(queue_call_end) def with_vision_hint(text: str) -> str: if not vision_enabled: @@ -466,7 +443,7 @@ async def run_pipeline( context = LLMContext( messages=[{"role": "system", "content": with_vision_hint(system_content)}] ) - # LLM 槽由大脑提供:内部类型=OpenAI 兼容服务;fastgpt=包 SDK 的伪 LLM。 + # LLM 槽由大脑提供:本地模型或 Dify/FastGPT 外部托管适配器。 llm = brain.build_llm(cfg, context) user_aggregator = LLMUserAggregator( context, @@ -474,9 +451,7 @@ async def run_pipeline( vad_analyzer=SileroVADAnalyzer(), user_mute_strategies=[ FunctionCallUserMuteStrategy(), - CallEndingUserMuteStrategy( - lambda: bool(call_end_state["ending"]) - ), + CallEndingUserMuteStrategy(lambda: call_end.ending), ], user_turn_strategies=UserTurnStrategies( start=[ @@ -489,9 +464,7 @@ async def run_pipeline( ), ) assistant_aggregator = PassthroughLLMAssistantAggregator(context) - text_input = TextInputProcessor( - should_ignore_input=lambda: bool(call_end_state["ending"]) - ) + text_input = TextInputProcessor(should_ignore_input=lambda: call_end.ending) vision_capture = VisionCaptureProcessor() vision_native_mode = vision_enabled and _vision_uses_main_llm(cfg) vision_state: dict[str, str | None] = {"client_id": None} @@ -572,99 +545,6 @@ async def run_pipeline( if vision_enabled: llm.register_function(VISION_TOOL_NAME, fetch_user_image) - end_call_tools = [ - tool - for tool in cfg.tools - if cfg.type == "prompt" and tool.type == "end_call" - ] - end_call_schemas: list[FunctionSchema] = [] - worker_holder: dict = {} - - async def queue_call_end(reason: str) -> None: - if call_end_state["frame_queued"] or worker_holder.get("worker") is None: - return - call_end_state["frame_queued"] = True - logger.info(f"结束通话: reason={reason}") - await worker_holder["worker"].queue_frame( - OutputTransportMessageUrgentFrame( - message={"type": "call-ended", "reason": reason} - ) - ) - await worker_holder["worker"].queue_frame(EndFrame()) - - def make_end_call_handler(tool): - config = (tool.definition or {}).get("config") or {} - message_type = str(config.get("message_type") or "none") - custom_message = str(config.get("custom_message") or "").strip() - capture_reason = bool(config.get("capture_reason", True)) - - async def end_call(params: FunctionCallParams) -> None: - reason = str(params.arguments.get("reason") or "end_call_tool").strip() - call_end_state["ending"] = True - logger.info( - f"End Call Tool EXECUTED: {tool.function_name}, reason={reason}" - ) - await params.result_callback( - {"status": "success", "action": "ending_call"}, - properties=FunctionCallResultProperties(run_llm=False), - ) - - if message_type != "custom" or not custom_message: - await queue_call_end(reason) - return - - call_end_state["reason"] = reason - call_end_state["armed"] = True - turn_id = uuid4().hex - timestamp = time_now_iso8601() - for message in ( - { - "type": "assistant-text-start", - "turn_id": turn_id, - "timestamp": timestamp, - }, - { - "type": "assistant-text-delta", - "turn_id": turn_id, - "delta": custom_message, - }, - { - "type": "assistant-text-end", - "turn_id": turn_id, - "content": custom_message, - "interrupted": False, - }, - ): - await worker_holder["worker"].queue_frame( - OutputTransportMessageUrgentFrame(message=message) - ) - await worker_holder["worker"].queue_frame( - TTSSpeakFrame(custom_message, append_to_context=False) - ) - - properties = ( - { - "reason": { - "type": "string", - "description": "结束本次通话的简短原因。", - } - } - if capture_reason - else {} - ) - schema = FunctionSchema( - name=tool.function_name, - description=tool.description or "结束当前通话。", - properties=properties, - required=["reason"] if capture_reason else [], - ) - return schema, end_call - - for end_call_tool in end_call_tools: - schema, handler = make_end_call_handler(end_call_tool) - end_call_schemas.append(schema) - llm.register_function(end_call_tool.function_name, handler) - def set_visible_tools(schemas: list[FunctionSchema] | None = None) -> None: tools = list(schemas or []) if vision_enabled: @@ -674,31 +554,6 @@ async def run_pipeline( else: context.set_tools() - # Workflow 结束节点和 end_call 固定结束语都等到 BotStoppedSpeakingFrame - # 再挂断,确保文字(data channel)与音频完整送达。 - class EndCallAfterSpeech(FrameProcessor): - async def process_frame(self, frame, direction: FrameDirection): - await super().process_frame(frame, direction) - await self.push_frame(frame, direction) - armed = wf_state["end_armed"] or call_end_state["armed"] - # 结束语文本生成完(end_armed)→ 其音频开始(end_speaking)→ 音频说完才挂断。 - # 配对 started/stopped,避免被结束节点之前的话(如先答一句再转移)的 - # stopped 事件提前触发,导致结束语被截断。 - if isinstance(frame, BotStartedSpeakingFrame) and armed: - if wf_state["end_armed"]: - wf_state["end_speaking"] = True - if call_end_state["armed"]: - call_end_state["speaking"] = True - elif ( - isinstance(frame, BotStoppedSpeakingFrame) - and (wf_state["end_speaking"] or call_end_state["speaking"]) - and worker_holder.get("worker") is not None - ): - logger.info("结束语播报完毕,挂断通话") - wf_state["end_frame_queued"] = True - reason = str(call_end_state["reason"] or "completed") - await queue_call_end(reason) - recorder = await ConversationRecorder.start( assistant_id=assistant_id, assistant_name=cfg.name, @@ -718,7 +573,7 @@ async def run_pipeline( # waiting for a TTS provider to emit spoken-text/timestamp frames. assistant_aggregator, tts, - EndCallAfterSpeech(), + EndCallAfterSpeechProcessor(call_end), ConversationHistoryProcessor(recorder), transport.output(), ] @@ -749,15 +604,6 @@ async def run_pipeline( greeting_transcript_sent = False pending_text_inputs: list[str] = [] - async def emit_node_active(node_id: str | None) -> None: - """通知前端当前激活的节点,画布据此高亮。""" - if node_id: - await worker.queue_frame( - OutputTransportMessageUrgentFrame( - message={"type": "node-active", "nodeId": node_id} - ) - ) - def set_system_prompt(text: str) -> None: """替换上下文里的系统提示(节点切换时整体替换,而非追加)。""" messages = context.get_messages() @@ -767,89 +613,18 @@ async def run_pipeline( else: messages.insert(0, {"role": "system", "content": content}) - def apply_node(node_id: str | None) -> None: - """进入节点:设置系统提示 + 把出边注册为可调用的转移工具。""" - set_system_prompt(engine.system_prompt_for(node_id)) - if engine.is_end(node_id): - set_visible_tools([]) # 终止节点不展示转移工具,但保留视觉工具 - return - schemas = [ - FunctionSchema( - name=engine.edge_fn_name(edge), - description=engine.edge_description(edge), - properties={}, - required=[], - ) - for edge in engine.outgoing(node_id) - ] - set_visible_tools(schemas) - - async def go_to_node(target: str) -> None: - """执行转移:切当前节点、重置计数、点亮画布、设置提示/工具。 - - 结束节点:设 ended 标记,apply_node 会清空工具,模型据结束语提示说完后, - on_assistant_text_end 里排入 EndFrame 挂断,不再多轮。 - """ - wf_state["current"] = target - wf_state["turns_in_node"] = 0 - if engine.is_end(target): - wf_state["ended"] = True - await emit_node_active(target) - apply_node(target) - - async def speak_transition(edge: dict | None) -> None: - """切换瞬间播报过渡语(可选),掩盖切节点/新一轮生成的延迟。不写入上下文。""" - speech = engine.edge_transition_speech(edge) - if speech: - await worker.queue_frame(TTSSpeakFrame(speech, append_to_context=False)) - - def make_transition_handler(edge: dict): - target = edge.get("target") - - async def handler(params): - logger.info(f"LLM 触发转移 → {engine.name(target)}") - # 进结束节点不播过渡语(结束语本身就是收尾,避免打断挂断时序) - if not engine.is_end(target): - await speak_transition(edge) - await go_to_node(target) - # 返回工具结果,pipecat 随即在新节点的提示/工具下继续生成 - await params.result_callback({"status": "ok"}) - - return handler - - async def fallback_route() -> None: - """文本兜底:模型迟迟不调用转移工具时,用一次轻量分类器判断是否转移。""" - if not workflow_active or wf_state["ended"]: - return - if wf_state["turns_in_node"] < FALLBACK_AFTER_TURNS: - return - if not engine.outgoing(wf_state["current"]): - return - target = await engine.route( - wf_state["current"], - history, - api_key=_require(cfg.llm_api_key, "LLM apiKey"), - base_url=_require(cfg.llm_base_url, "LLM apiUrl"), - model=_require(cfg.model, "LLM modelId"), - ) - if target and target != wf_state["current"]: - logger.info(f"文本兜底触发转移 → {engine.name(target)}") - if not engine.is_end(target): - await speak_transition(engine.find_edge(wf_state["current"], target)) - # 仅切换节点提示/工具,下一轮用户输入即在新节点处理 - await go_to_node(target) - - # 把每条边注册成 LLM 可调用的转移函数(按边唯一命名,处理器全局注册一次, - # 由各节点的 context.tools 控制当前可见哪些)。 - if workflow_active: - for edge in engine.edges: - if edge.get("target"): - llm.register_function( - engine.edge_fn_name(edge), make_transition_handler(edge) - ) - apply_node(wf_state["current"]) # 设初始节点的提示与工具 - else: - set_visible_tools(end_call_schemas) + set_visible_tools([]) + await brain.setup( + cfg, + BrainRuntime( + context=context, + llm=llm, + queue_frame=worker.queue_frame, + set_system_prompt=set_system_prompt, + set_tools=set_visible_tools, + call_end=call_end, + ), + ) async def append_user_text_to_context(text: str, *, run_llm: bool) -> None: await worker.queue_frame( @@ -862,19 +637,12 @@ async def run_pipeline( @user_aggregator.event_handler("on_user_turn_stopped") async def on_user_turn_stopped(_aggregator, _strategy, message): if message.content: - history.append({"role": "user", "content": message.content}) + brain.record_user_message(message.content) await queue_transcript("user", message.content, message.timestamp) @assistant_aggregator.event_handler("on_assistant_text_start") async def on_assistant_text_start(_aggregator, turn_id, timestamp): - # 进入结束节点后,第一条「开始生成」的回复就是结束节点自己的结束语 - # (其 text_start 发生在 ended 置位之后,不会误认转移前的那句)。 - if ( - workflow_active - and wf_state["ended"] - and wf_state["end_turn_id"] is None - ): - wf_state["end_turn_id"] = turn_id + await brain.on_assistant_text_start(turn_id) await worker.queue_frame( OutputTransportMessageUrgentFrame( message={ @@ -909,24 +677,12 @@ async def run_pipeline( } ) ) - # 助手把话说完(未被打断)后:累加本节点轮次,必要时走文本兜底路由。 - # 正常情况下转移由 LLM 直接调用转移工具完成(go_to_node),无需这里处理。 - if content and not interrupted and workflow_active: - history.append({"role": "assistant", "content": content}) - if turn_id == wf_state["end_turn_id"]: - # 结束节点的结束语文本已生成完(也已下发 data channel),武装挂断; - # 真正的 EndFrame 由 EndCallAfterSpeech 在结束语「说完」时排入。 - wf_state["end_armed"] = True - elif not wf_state["ended"]: - wf_state["turns_in_node"] += 1 - await fallback_route() - elif content and not interrupted: - history.append({"role": "assistant", "content": content}) + await brain.on_assistant_text_end(turn_id, content, interrupted) @text_input.event_handler("on_text_input") async def on_text_input(_processor, text): pending_text_inputs.append(text) - history.append({"role": "user", "content": text}) + brain.record_user_message(text) # 前端显示不依赖 interruption 后续事件,必须在打断前先排入发送队列。 await queue_transcript("user", text, time_now_iso8601()) @@ -941,7 +697,7 @@ async def run_pipeline( @text_input.event_handler("on_text_append") async def on_text_append(_processor, text): # 静默追加:写进上下文但不打断、不触发推理;transcript 照常上报 - history.append({"role": "user", "content": text}) + brain.record_user_message(text) await queue_transcript("user", text, time_now_iso8601()) await append_user_text_to_context(text, run_llm=False) @@ -969,9 +725,7 @@ async def run_pipeline( if brain.spec.owns_context: context.add_message({"role": "assistant", "content": greeting}) await worker.queue_frame(TTSSpeakFrame(greeting, append_to_context=False)) - # 工作流:点亮当前(开始)节点。开始节点即首个会话节点。 - if workflow_active: - await emit_node_active(wf_state["current"]) + await brain.on_connected() @transport.event_handler("on_client_disconnected") async def on_client_disconnected(_transport, _client): @@ -996,12 +750,17 @@ async def run_realtime_pipeline( transport, cfg: AssistantConfig, *, + brain: Brain, assistant_id: str | None = None, channel: str = "webrtc", ) -> None: """Run a speech-to-speech model that owns ASR, reasoning, and synthesis.""" - realtime = create_realtime_service(cfg) + realtime = create_realtime_service( + cfg, + instructions=brain.system_prompt(cfg), + ) text_input = RealtimeTextInputProcessor() + greeting = await brain.greeting(cfg) recorder = await ConversationRecorder.start( assistant_id=assistant_id, @@ -1058,8 +817,8 @@ async def run_realtime_pipeline( @transport.event_handler("on_client_connected") async def on_client_connected(_transport, _client): - if cfg.greeting: - await realtime.speak(cfg.greeting) + if greeting: + await realtime.speak(greeting) @transport.event_handler("on_client_disconnected") async def on_client_disconnected(_transport, _client): diff --git a/backend/services/pipecat/service_factory.py b/backend/services/pipecat/service_factory.py index 898ea3c..ce2ff59 100644 --- a/backend/services/pipecat/service_factory.py +++ b/backend/services/pipecat/service_factory.py @@ -142,7 +142,7 @@ def create_tts(cfg: AssistantConfig): ) -def create_realtime_service(cfg: AssistantConfig): +def create_realtime_service(cfg: AssistantConfig, *, instructions: str): """Create a speech-to-speech service that owns STT, LLM, and TTS.""" if cfg.realtime_interface_type == "stepfun-realtime": from services.pipecat.stepfun_realtime import StepFunRealtimeService @@ -151,7 +151,7 @@ def create_realtime_service(cfg: AssistantConfig): api_key=_require(cfg.realtime_api_key, "Realtime apiKey"), model=_require(cfg.realtimeModel, "Realtime modelId"), base_url=_require(cfg.realtime_base_url, "Realtime apiUrl"), - instructions=cfg.prompt, + instructions=instructions, voice=str(cfg.realtime_values.get("voice") or "linjiajiejie"), input_sample_rate=int( cfg.realtime_values.get("inputSampleRate") or 24000 diff --git a/backend/services/workflow_engine.py b/backend/services/workflow_engine.py index 1b893fa..29e13af 100644 --- a/backend/services/workflow_engine.py +++ b/backend/services/workflow_engine.py @@ -3,6 +3,7 @@ 对应 dograh 的 pipecat_engine.py,极简实现: - 单个 startCall 入口,开场白来自该节点; - agentNode 用各自的 prompt 驱动多轮对话; + - globalNode 不参与连线,按节点开关向会话节点注入统一提示词; - 每轮助手回复后,用一次轻量 LLM「路由」判断是否满足某条出边的 condition, 满足则切换当前节点(linear = 单边;branching = 多边按条件分流); - 到达 endCall 播放结束语并停止路由。 @@ -27,6 +28,10 @@ class WorkflowEngine: (nid for nid, n in self.nodes.items() if n.get("type") == "startCall"), None, ) + self.global_id: str | None = next( + (nid for nid, n in self.nodes.items() if n.get("type") == "globalNode"), + None, + ) # ---- 结构查询 ---- def node_type(self, nid: str | None) -> str | None: @@ -80,10 +85,25 @@ class WorkflowEngine: return self.data(self.start_id).get("greeting") or "" def system_prompt_for(self, nid: str | None) -> str: - """节点系统提示:仅用该节点自己的 prompt(开始节点也是会话节点)。""" + """组合当前节点提示与可选的全局提示(开始节点也是会话节点)。""" header = f"[当前节点:{self.name(nid)}]" - prompt = str(self.data(nid).get("prompt") or "").strip() - return f"{header}\n{prompt}" if prompt else header + node_data = self.data(nid) + prompt = str(node_data.get("prompt") or "").strip() + node_type = self.node_type(nid) + default_add_global = node_type in {"startCall", "agentNode"} + add_global = bool(node_data.get("addGlobalPrompt", default_add_global)) + global_prompt = ( + str(self.data(self.global_id).get("prompt") or "").strip() + if add_global and self.global_id + else "" + ) + + sections = [header] + if global_prompt: + sections.append(f"[全局规则]\n{global_prompt}") + if prompt: + sections.append(f"[当前节点任务]\n{prompt}") + return "\n\n".join(sections) # ---- 路由:决定下一节点 ---- async def route( diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py new file mode 100644 index 0000000..48d6a2c --- /dev/null +++ b/backend/tests/test_brains.py @@ -0,0 +1,303 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace + +from models import AssistantConfig, RuntimeTool +from pipecat.frames.frames import ( + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMTextFrame, +) +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.frame_processor import FrameDirection +from schemas import AssistantUpsert, REALTIME_CAPABLE_TYPES +from services.brains import BrainRuntime, SPECS, build_brain +from services.brains.dify_llm import ( + DifyLLMService, + last_user_text, + normalize_api_base, +) +from services.brains.workflow_brain import WorkflowBrain + + +class FakeLLM: + def __init__(self): + self.functions = {} + + def register_function(self, name, handler): + self.functions[name] = handler + + +class FakeCallEnd: + def __init__(self): + self.ending = False + self.reason = "" + self.armed = False + self.finished = False + + def begin(self, reason: str) -> None: + self.ending = True + self.reason = reason + + def arm_after_speech(self) -> None: + self.armed = True + + async def finish(self) -> None: + self.finished = True + + +class FakeFunctionParams: + def __init__(self, arguments=None): + self.arguments = arguments or {} + self.result = None + self.properties = None + + async def result_callback(self, result, properties=None): + self.result = result + self.properties = properties + + +class BrainRegistryTests(unittest.TestCase): + def test_capability_matrix(self): + self.assertEqual( + { + name: spec.supported_runtime_modes + for name, spec in SPECS.items() + }, + { + "prompt": frozenset({"pipeline", "realtime"}), + "workflow": frozenset({"pipeline"}), + "dify": frozenset({"pipeline"}), + "fastgpt": frozenset({"pipeline"}), + }, + ) + self.assertEqual( + REALTIME_CAPABLE_TYPES, + { + name + for name, spec in SPECS.items() + if "realtime" in spec.supported_runtime_modes + }, + ) + + def test_unknown_brain_does_not_fallback_to_prompt(self): + with self.assertRaisesRegex(ValueError, "尚未实现"): + build_brain(AssistantConfig(type="opencode")) + + def test_workflow_realtime_is_rejected_at_schema_boundary(self): + with self.assertRaises(ValueError): + AssistantUpsert( + name="workflow", + type="workflow", + runtimeMode="realtime", + ) + + +class DifyHelpersTests(unittest.TestCase): + def test_normalize_api_base(self): + self.assertEqual( + normalize_api_base("https://api.dify.ai"), + "https://api.dify.ai/v1", + ) + self.assertEqual( + normalize_api_base("https://example.test/v1/chat-messages"), + "https://example.test/v1", + ) + + def test_last_user_text(self): + self.assertEqual( + last_user_text( + [ + {"role": "user", "content": "first"}, + {"role": "assistant", "content": "answer"}, + { + "role": "user", + "content": [{"type": "text", "text": "latest"}], + }, + ] + ), + "latest", + ) + + +class DifyLLMServiceTests(unittest.IsolatedAsyncioTestCase): + async def test_streams_sdk_events_and_keeps_conversation_id(self): + class FakeDifyClient: + requests = [] + + async def achat_messages(self, request, **_kwargs): + self.requests.append(request) + + async def events(): + yield SimpleNamespace( + event="message", + answer="你好", + conversation_id="conversation-1", + ) + yield SimpleNamespace( + event="message_end", + conversation_id="conversation-1", + ) + + return events() + + client = FakeDifyClient() + service = DifyLLMService( + AssistantConfig(type="dify"), + client=client, + user_id="test-user", + ) + frames = [] + + async def push_frame(frame, *_args, **_kwargs): + frames.append(frame) + + service.push_frame = push_frame + context = LLMContext(messages=[{"role": "user", "content": "问题"}]) + await service.process_frame( + LLMContextFrame(context), + FrameDirection.DOWNSTREAM, + ) + + self.assertIsInstance(frames[0], LLMFullResponseStartFrame) + self.assertIsInstance(frames[1], LLMTextFrame) + self.assertEqual(frames[1].text, "你好") + self.assertIsInstance(frames[-1], LLMFullResponseEndFrame) + self.assertEqual(service._conversation_id, "conversation-1") + + context.add_message({"role": "user", "content": "追问"}) + await service.process_frame( + LLMContextFrame(context), + FrameDirection.DOWNSTREAM, + ) + self.assertEqual(client.requests[-1].conversation_id, "conversation-1") + + +class PromptBrainTests(unittest.IsolatedAsyncioTestCase): + async def test_end_call_tool_is_owned_by_prompt_brain(self): + brain = build_brain( + AssistantConfig( + type="prompt", + tools=[ + RuntimeTool( + id="end-call", + name="结束通话", + function_name="end_call", + type="end_call", + definition={ + "config": { + "message_type": "none", + "capture_reason": True, + } + }, + ) + ], + ) + ) + llm = FakeLLM() + call_end = FakeCallEnd() + visible_tools = [] + + async def queue_frame(_frame): + pass + + await brain.setup( + AssistantConfig( + type="prompt", + tools=[ + RuntimeTool( + id="end-call", + name="结束通话", + function_name="end_call", + type="end_call", + definition={"config": {"capture_reason": True}}, + ) + ], + ), + BrainRuntime( + context=LLMContext(messages=[]), + llm=llm, + queue_frame=queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda tools: visible_tools.extend(tools or []), + call_end=call_end, + ), + ) + + self.assertEqual(visible_tools[0].name, "end_call") + params = FakeFunctionParams({"reason": "用户已完成咨询"}) + await llm.functions["end_call"](params) + self.assertEqual(call_end.reason, "用户已完成咨询") + self.assertTrue(call_end.finished) + self.assertEqual(params.result["action"], "ending_call") + + +class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): + async def test_transition_and_end_are_owned_by_workflow_brain(self): + graph = { + "nodes": [ + { + "id": "start", + "type": "startCall", + "data": {"name": "开始", "prompt": "收集需求"}, + }, + { + "id": "end", + "type": "endCall", + "data": {"name": "结束", "prompt": "礼貌结束"}, + }, + ], + "edges": [ + { + "id": "finish", + "source": "start", + "target": "end", + "data": {"condition": "需求已收集"}, + } + ], + } + brain = WorkflowBrain(graph) + llm = FakeLLM() + context = LLMContext(messages=[]) + queued = [] + prompts = [] + visible_tools = [] + call_end = FakeCallEnd() + + async def queue_frame(frame): + queued.append(frame) + + runtime = BrainRuntime( + context=context, + llm=llm, + queue_frame=queue_frame, + set_system_prompt=prompts.append, + set_tools=lambda tools: visible_tools.append(tools or []), + call_end=call_end, + ) + await brain.setup(AssistantConfig(type="workflow", graph=graph), runtime) + + self.assertIn("goto_finish", llm.functions) + self.assertIn("收集需求", prompts[-1]) + self.assertEqual(visible_tools[-1][0].name, "goto_finish") + + params = FakeFunctionParams() + await llm.functions["goto_finish"](params) + self.assertEqual(params.result, {"status": "ok"}) + self.assertIn("礼貌结束", prompts[-1]) + self.assertEqual(visible_tools[-1], []) + + await brain.on_assistant_text_start("closing-turn") + await brain.on_assistant_text_end( + "closing-turn", + "感谢来电，再见。", + False, + ) + self.assertTrue(call_end.ending) + self.assertTrue(call_end.armed) + + +if __name__ == "__main__": + unittest.main() diff --git a/frontend/src/components/workflow/GenericNode.tsx b/frontend/src/components/workflow/GenericNode.tsx index f65e374..aedd38b 100644 --- a/frontend/src/components/workflow/GenericNode.tsx +++ b/frontend/src/components/workflow/GenericNode.tsx @@ -144,4 +144,5 @@ export const nodeTypes = { startCall: GenericNode, agentNode: GenericNode, endCall: GenericNode, + globalNode: GenericNode, }; diff --git a/frontend/src/components/workflow/WorkflowEditor.tsx b/frontend/src/components/workflow/WorkflowEditor.tsx index c4ee1d9..f3948dc 100644 --- a/frontend/src/components/workflow/WorkflowEditor.tsx +++ b/frontend/src/components/workflow/WorkflowEditor.tsx @@ -90,6 +90,14 @@ export type WorkflowEditorProps = { let nodeSeq = 0; const NONE = "__none__"; +function defaultNodeData(spec: RuntimeNodeSpec): WorkflowNodeData { + const data: WorkflowNodeData = { name: spec.displayName }; + for (const field of spec.fields) { + if (field.default !== undefined) data[field.key] = field.default; + } + return data; +} + function toFlow(graph: WorkflowGraph): { nodes: Node[]; edges: Edge[] } { return { nodes: graph.nodes.map((n) => ({ @@ -179,12 +187,34 @@ function Canvas({ const isValidConnection = useCallback( (c: Connection | Edge) => { if (c.source === c.target) return false; + const source = nodes.find((n) => n.id === c.source); const target = nodes.find((n) => n.id === c.target); - if (!target) return false; - const spec = specsByType[target.type as string]; - return !!spec?.hasTarget; + if (!source || !target) return false; + + const sourceSpec = specsByType[source.type as string]; + const targetSpec = specsByType[target.type as string]; + if (!sourceSpec?.hasSource || !targetSpec?.hasTarget) return false; + if (edges.some((e) => e.source === c.source && e.target === c.target)) { + return false; + } + + const sourceLimit = sourceSpec.constraints.maxOutgoing; + if ( + sourceLimit !== undefined && + edges.filter((e) => e.source === c.source).length >= sourceLimit + ) { + return false; + } + const targetLimit = targetSpec.constraints.maxIncoming; + if ( + targetLimit !== undefined && + edges.filter((e) => e.target === c.target).length >= targetLimit + ) { + return false; + } + return true; }, - [nodes, specsByType], + [edges, nodes, specsByType], ); const addNode = useCallback( @@ -195,11 +225,7 @@ function Canvas({ x: window.innerWidth / 2, y: window.innerHeight / 2, }); - const data: WorkflowNodeData = { name: spec.displayName }; - if (spec.type === "agentNode") { - data.allowInterrupt = true; - data.prompt = ""; - } + const data = defaultNodeData(spec); setNodes((ns) => [...ns, { id, type: spec.type, position, data }]); setAddOpen(false); setEditingId(id); @@ -281,6 +307,14 @@ function Canvas({ const editingSpec = editingNode ? specsByType[editingNode.type as string] : null; const editingEdge = edges.find((e) => e.id === editingEdgeId); const addableSpecs = Object.values(specsByType).filter((s) => s.addable); + const canAddSpec = useCallback( + (spec: RuntimeNodeSpec) => { + const limit = spec.constraints.maxInstances; + if (limit === undefined) return true; + return nodes.filter((node) => node.type === spec.type).length < limit; + }, + [nodes], + ); return ( @@ -398,11 +432,13 @@ function Canvas({ ) : null} {addableSpecs.map((spec) => { const Icon = spec.icon; + const canAdd = canAddSpec(spec); return ( + ); diff --git a/frontend/src/components/workflow/specs.ts b/frontend/src/components/workflow/specs.ts index 8e25355..04e01ec 100644 --- a/frontend/src/components/workflow/specs.ts +++ b/frontend/src/components/workflow/specs.ts @@ -11,7 +11,11 @@ import { Circle, type LucideIcon } from "lucide-react"; import type { NodeSpecDto } from "@/lib/api"; -export type WorkflowNodeType = "startCall" | "agentNode" | "endCall"; +export type WorkflowNodeType = + | "startCall" + | "agentNode" + | "endCall" + | "globalNode"; export type WorkflowNodeData = { /** 节点显示名 */ @@ -22,6 +26,8 @@ export type WorkflowNodeData = { prompt?: string; /** 允许打断(仅 agentNode) */ allowInterrupt?: boolean; + /** 是否合并全局节点提示词(start/agent 默认开启,end 默认关闭) */ + addGlobalPrompt?: boolean; [key: string]: unknown; }; @@ -30,6 +36,7 @@ export type FieldSpec = { label: string; type: "text" | "textarea" | "switch"; required?: boolean; + default?: unknown; }; /** 解析后的运行期节点规格(DTO + 解析出的 React 图标 + 派生句柄) */ @@ -44,6 +51,14 @@ export type RuntimeNodeSpec = { hasTarget: boolean; /** 出边句柄(结束节点没有) */ hasSource: boolean; + constraints: { + minIncoming?: number; + maxIncoming?: number; + minOutgoing?: number; + maxOutgoing?: number; + minInstances?: number; + maxInstances?: number; + }; fields: FieldSpec[]; }; @@ -77,11 +92,13 @@ export function toRuntimeSpec(dto: NodeSpecDto): RuntimeNodeSpec { addable: dto.addable, hasTarget: dto.constraints.maxIncoming !== 0, hasSource: dto.constraints.maxOutgoing !== 0, + constraints: dto.constraints, fields: dto.fields.map((f) => ({ key: f.key, label: f.label, type: f.type, required: f.required, + default: f.default, })), }; } @@ -104,27 +121,52 @@ export type WorkflowGraph = { viewport?: { x: number; y: number; zoom: number }; }; -/** 新建工作流的默认图:开始 → 智能体 → 结束 */ +/** 新建工作流的默认图:全局规则 + 开始 → 智能体 → 结束 */ export function defaultGraph(): WorkflowGraph { return { nodes: [ { id: "start", type: "startCall", - position: { x: 80, y: 160 }, - data: { name: "开始", greeting: "你好,我是 AI 视频助手,有什么可以帮你?" }, + position: { x: 100, y: 120 }, + data: { + name: "开始", + greeting: "你好,我是 AI 视频助手,有什么可以帮你?", + prompt: "了解用户的需求,并在信息明确后进入下一节点。", + allowInterrupt: true, + addGlobalPrompt: true, + }, }, { id: "agent-1", type: "agentNode", - position: { x: 400, y: 160 }, - data: { name: "智能体节点", prompt: "", allowInterrupt: true }, + position: { x: 420, y: 120 }, + data: { + name: "智能体节点", + prompt: "根据用户需求提供清晰、准确的帮助。", + allowInterrupt: true, + addGlobalPrompt: true, + }, }, { id: "end", type: "endCall", - position: { x: 720, y: 160 }, - data: { name: "结束", prompt: "感谢你的来电,再见!" }, + position: { x: 740, y: 120 }, + data: { + name: "结束", + prompt: "总结已经完成的事项,礼貌道别并结束通话。", + addGlobalPrompt: false, + }, + }, + { + id: "global", + type: "globalNode", + position: { x: 100, y: 400 }, + data: { + name: "全局设定", + prompt: + "你是一个友好、专业的语音助手。请使用简短、自然、适合口语表达的句子。", + }, }, ], edges: [ diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 954d83d..eda4b00 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -338,6 +338,7 @@ export type NodeFieldSpec = { label: string; type: "text" | "textarea" | "switch"; required?: boolean; + default?: unknown; }; export type NodeConstraints = { @@ -345,6 +346,8 @@ export type NodeConstraints = { maxIncoming?: number; minOutgoing?: number; maxOutgoing?: number; + minInstances?: number; + maxInstances?: number; }; export type NodeSpecDto = {