From 32aef14ddb89d991477fbdc91350a14669591e1d Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Mon, 13 Jul 2026 16:13:27 +0800 Subject: [PATCH] Add workflow support and enhance runtime configuration in models and services - Introduce RuntimeModelResource and RuntimeKnowledgeBase classes to manage workflow resources. - Update AssistantConfig to include workflow_model_resources and workflow_knowledge_bases for better integration. - Refactor validation and processing logic in routes and services to accommodate workflow types. - Implement dynamic variable support for workflow assistants and enhance graph normalization. - Add ToolExecutor for reusable tool execution across different assistant types. - Update various services to ensure compatibility with new workflow features and improve error handling. --- backend/models.py | 17 + backend/requirements.txt | 2 +- backend/routes/assistants.py | 54 +- backend/routes/knowledge_bases.py | 9 + backend/routes/model_registry.py | 18 + backend/schemas.py | 2 +- backend/services/auth.py | 11 + backend/services/brains/base.py | 9 +- backend/services/brains/prompt_brain.py | 123 +-- backend/services/brains/registry.py | 2 +- backend/services/brains/workflow_brain.py | 467 +++++++--- backend/services/config_resolver.py | 51 +- backend/services/node_specs.py | 525 +++++++---- backend/services/pipecat/pipeline.py | 233 ++++- backend/services/pipecat/service_factory.py | 31 +- backend/services/runtime_variables.py | 12 +- backend/services/tool_executor.py | 154 ++++ backend/services/workflow_engine.py | 242 +++-- backend/tests/test_brains.py | 131 ++- backend/tests/test_pipeline_knowledge.py | 25 +- backend/tests/test_workflow_v3.py | 118 +++ .../src/components/pages/AssistantPage.tsx | 129 ++- .../src/components/workflow/ConditionEdge.tsx | 19 +- .../src/components/workflow/GenericNode.tsx | 41 +- .../components/workflow/WorkflowEditor.tsx | 853 ++++++++++++++---- frontend/src/components/workflow/specs.ts | 162 ++-- frontend/src/lib/api.ts | 33 +- 27 files changed, 2563 insertions(+), 910 deletions(-) create mode 100644 backend/services/tool_executor.py create mode 100644 backend/tests/test_workflow_v3.py diff --git a/backend/models.py b/backend/models.py index e93f2c9..7c32dc3 100644 --- a/backend/models.py +++ b/backend/models.py @@ -27,6 +27,21 @@ class RuntimeTool(BaseModel): secrets: dict = Field(default_factory=dict) +class RuntimeModelResource(BaseModel): + id: str + name: str = "" + capability: str + interface_type: str + values: dict = Field(default_factory=dict) + secrets: dict = Field(default_factory=dict) + + +class RuntimeKnowledgeBase(BaseModel): + id: str + name: str = "" + description: str = "" + + class AssistantConfig(BaseModel): """运行时配置:前端可见部分(name/prompt/...) + 服务端注入部分(*_api_key/*_base_url)。""" @@ -93,6 +108,8 @@ class AssistantConfig(BaseModel): # workflow 类型:节点图(nodes/edges)。非 workflow 为空,引擎据此决定是否启用。 graph: dict = {} + workflow_model_resources: dict[str, RuntimeModelResource] = Field(default_factory=dict) + workflow_knowledge_bases: dict[str, RuntimeKnowledgeBase] = Field(default_factory=dict) # 外部托管类型(fastgpt/dify/opencode)的连接信息:context/KB/tools 由对方服务端接管。 dify_api_url: str = "" diff --git a/backend/requirements.txt b/backend/requirements.txt index b2c0792..b6c947a 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -2,7 +2,7 @@ # webrtc -> SmallWebRTCTransport / SmallWebRTCConnection + aiortc # silero -> 本地 VAD(判断用户说话起止),语音必备 # openai -> OpenAI 兼容的 LLM/STT/TTS 客户端(DeepSeek、SenseVoice、CosyVoice 都走它) -pipecat-ai[webrtc,websocket,silero,openai]==1.4.0 +pipecat-ai[webrtc,websocket,silero,openai]==1.5.0 Pillow>=11.1.0,<13 # FastGPT 类型助手:本地 SDK(包 /api/v1/chat/completions 流式 + chatId 会话) diff --git a/backend/routes/assistants.py b/backend/routes/assistants.py index 3161911..1f3c9ca 100644 --- a/backend/routes/assistants.py +++ b/backend/routes/assistants.py @@ -15,7 +15,7 @@ from fastapi import APIRouter, Depends, HTTPException from schemas import AssistantOut, AssistantUpsert from services.auth import require_admin from services.masking import mask, resolve_incoming_key -from services.node_specs import validate_graph +from services.node_specs import graph_references, normalize_graph, validate_graph from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -31,9 +31,45 @@ def _validate_workflow(body: AssistantUpsert) -> None: """workflow 类型:保存前校验图结构,不通过则 400。其他类型跳过。""" if body.type != "workflow": return - errors = validate_graph(body.graph or {}) + body.graph = normalize_graph(body.graph or {}) + errors = validate_graph(body.graph) if errors: raise HTTPException(400, "工作流校验失败:" + ";".join(errors)) + refs = graph_references(body.graph) + body.tool_ids = list(dict.fromkeys([*body.tool_ids, *sorted(refs["tools"])])) + + +async def _validate_workflow_references( + session: AsyncSession, body: AssistantUpsert +) -> None: + if body.type != "workflow" or not body.graph.get("nodes"): + return + graph = body.graph + settings = graph.get("settings") or {} + resource_expectations: dict[str, str] = {} + for key, capability in ( + ("defaultAsrResourceId", "ASR"), + ("defaultTtsResourceId", "TTS"), + ): + if settings.get(key): + resource_expectations[str(settings[key])] = capability + knowledge_ids: set[str] = set() + for node in graph.get("nodes") or []: + data = node.get("data") or {} + if data.get("asrResourceId"): + resource_expectations[str(data["asrResourceId"])] = "ASR" + if data.get("ttsResourceId"): + resource_expectations[str(data["ttsResourceId"])] = "TTS" + if data.get("knowledgeBaseId"): + knowledge_ids.add(str(data["knowledgeBaseId"])) + for resource_id, capability in resource_expectations.items(): + resource = await session.get(ModelResource, resource_id) + if not resource or not resource.enabled or resource.capability != capability: + raise HTTPException(400, f"Workflow 引用了无效的 {capability} 资源:{resource_id}") + for knowledge_id in knowledge_ids: + knowledge = await session.get(KnowledgeBase, knowledge_id) + if not knowledge or knowledge.status != "active": + raise HTTPException(400, f"Workflow 引用了无效知识库:{knowledge_id}") async def _validate_vision_model( @@ -101,7 +137,11 @@ async def _resource_ids(session: AsyncSession, assistant_id: str) -> dict[str, s async def _sync_tool_bindings( session: AsyncSession, assistant_id: str, assistant_type: str, tool_ids: list[str] ) -> None: - requested = list(dict.fromkeys(tool_ids)) if assistant_type == "prompt" else [] + requested = ( + list(dict.fromkeys(tool_ids)) + if assistant_type in {"prompt", "workflow"} + else [] + ) if requested: tools = ( await session.execute(select(Tool).where(Tool.id.in_(requested))) @@ -158,7 +198,11 @@ async def _to_out(session: AsyncSession, assistant: Assistant) -> AssistantOut: api_url=assistant.api_url, api_key=mask(assistant.api_key), app_id=assistant.app_id, - graph=assistant.graph or {}, + graph=( + normalize_graph(assistant.graph or {}) + if assistant.type == "workflow" + else {} + ), updated_at=assistant.updated_at.isoformat() if assistant.updated_at else None, ) @@ -176,6 +220,7 @@ async def create_assistant( body: AssistantUpsert, session: AsyncSession = Depends(get_session) ): _validate_workflow(body) + await _validate_workflow_references(session, body) await _validate_vision_model(session, body) await _validate_knowledge_base(session, body) data = body.model_dump() @@ -248,6 +293,7 @@ async def update_assistant( if not assistant: raise HTTPException(404, "助手不存在") _validate_workflow(body) + await _validate_workflow_references(session, body) await _validate_vision_model(session, body) await _validate_knowledge_base(session, body) data = body.model_dump() diff --git a/backend/routes/knowledge_bases.py b/backend/routes/knowledge_bases.py index f23755d..22e14e9 100644 --- a/backend/routes/knowledge_bases.py +++ b/backend/routes/knowledge_bases.py @@ -15,6 +15,7 @@ from schemas import ( ) from services.auth import require_admin from services.knowledge import create_document, delete_storage_object, process_document, search +from services.node_specs import graph_references import settings from sqlalchemy import select from sqlalchemy.exc import IntegrityError @@ -123,6 +124,14 @@ async def delete_knowledge_base( )).scalar_one_or_none() if referenced: raise HTTPException(409, "知识库正被助手引用,无法删除") + workflows = ( + await session.execute(select(Assistant).where(Assistant.type == "workflow")) + ).scalars().all() + if any( + kb_id in graph_references(assistant.graph or {})["knowledge_bases"] + for assistant in workflows + ): + raise HTTPException(409, "知识库正被 Workflow 节点引用,无法删除") documents = (await session.execute( select(KnowledgeDocument).where(KnowledgeDocument.knowledge_base_id == kb_id) )).scalars().all() diff --git a/backend/routes/model_registry.py b/backend/routes/model_registry.py index 4c89026..c463ad6 100644 --- a/backend/routes/model_registry.py +++ b/backend/routes/model_registry.py @@ -3,6 +3,7 @@ import uuid from db.models import ( + Assistant, AssistantModelBinding, InterfaceDefinition, KnowledgeBase, @@ -20,6 +21,7 @@ from services.auth import require_admin from services.interface_catalog import validate_fields from services.masking import mask_secrets, merge_secrets from services.model_resource_tester import test_model_resource +from services.node_specs import graph_references from sqlalchemy import delete, select, update from sqlalchemy.ext.asyncio import AsyncSession @@ -102,6 +104,14 @@ async def _clear_incompatible_references( ) -> None: if capability == resource.capability: return + workflows = ( + await session.execute(select(Assistant).where(Assistant.type == "workflow")) + ).scalars().all() + if any( + resource.id in graph_references(assistant.graph or {})["model_resources"] + for assistant in workflows + ): + raise HTTPException(409, "模型资源正被 Workflow 节点引用，不能修改能力类型") await session.execute( delete(AssistantModelBinding).where( AssistantModelBinding.model_resource_id == resource.id @@ -263,6 +273,14 @@ async def delete_model_resource( ).scalar_one_or_none() if in_use: raise HTTPException(409, "该模型资源仍被助手引用") + workflows = ( + await session.execute(select(Assistant).where(Assistant.type == "workflow")) + ).scalars().all() + if any( + resource_id in graph_references(assistant.graph or {})["model_resources"] + for assistant in workflows + ): + raise HTTPException(409, "该模型资源仍被 Workflow 节点引用") await session.delete(row) await session.commit() return {"ok": True} diff --git a/backend/schemas.py b/backend/schemas.py index 75ff066..684b375 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -138,7 +138,7 @@ class AssistantUpsert(CamelModel): setattr(self, field, "") if "graph" not in allowed: self.graph = {} - if self.type != "prompt": + if self.type not in {"prompt", "workflow"}: self.tool_ids = [] self.dynamic_variable_definitions = {} # 外部托管大脑只能 cascade,拦住不兼容的 realtime diff --git a/backend/services/auth.py b/backend/services/auth.py index d1e553f..c3f1c8d 100644 --- a/backend/services/auth.py +++ b/backend/services/auth.py @@ -109,6 +109,17 @@ def clear_auth_cookie(response: Response) -> None: async def require_admin(request: Request) -> AdminUser: user = verify_admin_token(request.cookies.get(settings.AUTH_COOKIE_NAME)) + if not user: + authorization = request.headers.get("authorization", "") + scheme, _, encoded_credentials = authorization.partition(" ") + if scheme.lower() == "basic" and encoded_credentials: + try: + credentials = base64.b64decode(encoded_credentials).decode("utf-8") + username, separator, password = credentials.partition(":") + if separator: + user = authenticate_admin(username, password) + except (ValueError, UnicodeDecodeError): + user = None if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index de1e6d5..ee07a90 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -9,7 +9,7 @@ without coupling brains to Pipecat internals more than necessary. from __future__ import annotations from collections.abc import Awaitable, Callable -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any, Protocol, runtime_checkable from models import AssistantConfig @@ -52,6 +52,13 @@ class BrainRuntime: set_system_prompt: Callable[[str], None] set_tools: Callable[[list[FunctionSchema] | None], None] call_end: CallEndPort + worker: Any = None + context_aggregator: Any = None + transport: Any = None + switch_services: Callable[[str | None, str | None], Awaitable[None]] | None = None + set_knowledge_scope: Callable[[dict[str, Any]], None] | None = None + set_input_enabled: Callable[[bool], None] | None = None + flow_global_functions: list[Any] = field(default_factory=list) class BaseBrain: diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py index 8abf092..7d1bc50 100644 --- a/backend/services/brains/prompt_brain.py +++ b/backend/services/brains/prompt_brain.py @@ -2,11 +2,8 @@ from __future__ import annotations -from copy import deepcopy -from urllib.parse import quote from uuid import uuid4 -import httpx from models import AssistantConfig from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.frames.frames import OutputTransportMessageUrgentFrame, TTSSpeakFrame @@ -20,10 +17,9 @@ from pipecat.utils.time import time_now_iso8601 from services.brains.base import BaseBrain, BrainRuntime, BrainSpec from services.runtime_variables import ( - DynamicVariableError, DynamicVariableStore, - value_at_path, ) +from services.tool_executor import ToolExecutionError, ToolExecutor class PromptBrain(BaseBrain): @@ -37,6 +33,7 @@ class PromptBrain(BaseBrain): self._cfg = cfg self._dynamic_enabled = True self._store = DynamicVariableStore.from_config(cfg) + self._tools = ToolExecutor(self._store) self._runtime: BrainRuntime | None = None async def greeting(self, cfg: AssistantConfig) -> str: @@ -85,120 +82,16 @@ class PromptBrain(BaseBrain): self._runtime.set_system_prompt(self._store.render(self._cfg.prompt)) def _make_http_tool(self, tool, runtime: BrainRuntime): - config = (tool.definition or {}).get("config") or {} - parameters = list(config.get("parameters") or []) - properties = { - str(parameter.get("name")): { - "type": str(parameter.get("type") or "string"), - "description": str(parameter.get("description") or ""), - } - for parameter in parameters - if parameter.get("name") - } - required = [ - str(parameter["name"]) - for parameter in parameters - if parameter.get("name") and parameter.get("required", True) - ] - - tool_secrets = tool.secrets or {} - dynamic_secrets = tool_secrets.get("dynamic_variables") or {} - for name, value in dynamic_secrets.items(): - if not str(name).startswith("secret__"): - raise DynamicVariableError(f"工具密钥变量必须以 secret__ 开头: {name}") - self._store.secrets[str(name)] = str(value) + properties, required = self._tools.schema_parts(tool) + self._tools.register_secrets(tool) async def call_http(params: FunctionCallParams) -> None: - arguments = params.arguments or {} - url = self._store.render(str(config.get("url") or "")) - configured_headers = self._store.render_data( - deepcopy(config.get("headers") or {}), allow_secrets=True - ) - secret_headers = self._store.render_data( - deepcopy(tool_secrets.get("headers") or {}), allow_secrets=True - ) - headers: dict[str, str] = {} - query: dict[str, object] = {} - body = self._store.render_data(deepcopy(config.get("body") or {})) - - for parameter in parameters: - name = str(parameter.get("name") or "") - if not name or name not in arguments: - continue - value = arguments[name] - location = str(parameter.get("location") or "body") - if location == "path": - encoded = quote(str(value), safe="") - url = url.replace(f"{{{name}}}", encoded) - elif location == "query": - query[name] = value - elif location == "header": - headers[name] = str(value) - else: - body[name] = value - - # Admin-configured headers win over model-provided arguments; secret - # headers are applied last so an LLM can never replace credentials. - headers.update( - {str(key): str(value) for key, value in configured_headers.items()} - ) - headers.update({str(key): str(value) for key, value in secret_headers.items()}) - try: - async with httpx.AsyncClient( - timeout=float(config.get("timeout_seconds") or 15), - follow_redirects=False, - ) as client: - response = await client.request( - str(config.get("method") or "GET"), - url, - headers=headers, - params=query, - json=body if body else None, - ) - response.raise_for_status() - if len(response.content) > 1_000_000: - raise DynamicVariableError("HTTP 工具响应超过 1 MB 限制") - try: - payload = response.json() - except ValueError: - payload = {"text": response.text[:8000]} - - updated: list[str] = [] - assignments = config.get("dynamic_variable_assignments") or {} - for variable_name, path in assignments.items(): - try: - value = value_at_path(payload, str(path)) - except KeyError: - try: - value = value_at_path({"response": payload}, str(path)) - except KeyError: - continue - self._store.assign(str(variable_name), value) - updated.append(str(variable_name)) - if updated: + result = await self._tools.execute(tool, dict(params.arguments or {})) + if result["updated_variables"]: self._refresh_prompt() - await params.result_callback( - { - "status": "ok", - "status_code": response.status_code, - "data": payload, - "updated_variables": updated, - } - ) - except httpx.TimeoutException: - await params.result_callback( - {"status": "error", "message": "HTTP 工具调用超时"} - ) - except httpx.HTTPStatusError as exc: - await params.result_callback( - { - "status": "error", - "status_code": exc.response.status_code, - "message": "HTTP 工具返回错误状态", - } - ) - except (httpx.RequestError, DynamicVariableError) as exc: + await params.result_callback(result) + except (ToolExecutionError, ValueError) as exc: await params.result_callback( {"status": "error", "message": f"HTTP 工具调用失败: {exc}"} ) diff --git a/backend/services/brains/registry.py b/backend/services/brains/registry.py index eebe5aa..1a7a059 100644 --- a/backend/services/brains/registry.py +++ b/backend/services/brains/registry.py @@ -14,7 +14,7 @@ from services.brains.workflow_brain import WorkflowBrain def _workflow(cfg: AssistantConfig) -> Brain: - return WorkflowBrain(cfg.graph) + return WorkflowBrain(cfg) BRAIN_FACTORIES: dict[str, Callable[[AssistantConfig], Brain]] = { diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index ec90ce3..2aedef5 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -1,27 +1,40 @@ -"""Local graph-driven workflow assistant and its per-call state.""" +"""Pipecat Flows-backed Workflow v3 brain.""" 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 models import AssistantConfig, RuntimeTool +from db.session import SessionLocal +from pipecat.flows import ( + ContextStrategy, + ContextStrategyConfig, + FlowManager, + FlowsFunctionSchema, + NodeConfig, +) +from pipecat.frames.frames import ( + LLMUpdateSettingsFrame, + OutputTransportMessageUrgentFrame, + TTSSpeakFrame, +) from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameProcessor +from pipecat.services.settings import LLMSettings from services.brains.base import BaseBrain, BrainRuntime, BrainSpec +from services.knowledge import search as search_knowledge +from services.runtime_variables import DynamicVariableStore +from services.tool_executor import ToolExecutionError, ToolExecutor 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 +MAX_AUTOMATIC_HOPS = 50 +AGENT_STAGE_INSTRUCTION = ( + "完成当前阶段任务。需要流转时必须调用对应的转移函数；" + "不要在调用转移函数后继续生成口头回复。" +) class WorkflowBrain(BaseBrain): @@ -30,22 +43,28 @@ class WorkflowBrain(BaseBrain): supported_runtime_modes=frozenset({"pipeline"}), owns_context=True, ) - _FALLBACK_AFTER_TURNS = 2 - def __init__(self, graph: dict[str, Any]): + def __init__(self, cfg_or_graph: AssistantConfig | dict[str, Any]): + cfg = cfg_or_graph if isinstance(cfg_or_graph, AssistantConfig) else None + graph = cfg.graph if cfg is not None else cfg_or_graph 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 + raise ValueError("WorkflowBrain 缺少有效的 Start 节点") + self._cfg = cfg + self._store = DynamicVariableStore.from_config(cfg or AssistantConfig(type="workflow")) + self._tools = ToolExecutor(self._store) + self._tool_by_id: dict[str, RuntimeTool] = { + tool.id: tool for tool in (cfg.tools if cfg else []) + } self._runtime: BrainRuntime | None = None + self._manager: FlowManager | None = None + self._ended = False async def greeting(self, cfg: AssistantConfig) -> str: - return self._engine.greeting() or cfg.greeting + return self._engine.greeting(self._store) or cfg.greeting def system_prompt(self, cfg: AssistantConfig) -> str: - return self._engine.system_prompt_for(self._state.current) + return self._store.render(self._engine.global_prompt()) def build_llm(self, cfg: AssistantConfig, context: LLMContext) -> FrameProcessor: from services.pipecat.service_factory import create_llm @@ -53,72 +72,331 @@ class WorkflowBrain(BaseBrain): return create_llm(cfg) async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: + if runtime.worker is None or runtime.context_aggregator is None: + raise RuntimeError("WorkflowBrain 需要 PipelineWorker 和 context aggregator pair") 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)}" + self._store = DynamicVariableStore.from_config(cfg) + self._tools = ToolExecutor(self._store) + self._tool_by_id = {tool.id: tool for tool in cfg.tools} + self._manager = FlowManager( + worker=runtime.worker, + llm=runtime.llm, + context_aggregator=runtime.context_aggregator, + transport=runtime.transport, + global_functions=runtime.flow_global_functions, ) + self._manager.state["variables"] = self._store.values async def on_connected(self) -> None: - await self._emit_node_active(self._state.current) + await self._emit_node_active(self._engine.start_id) + edge = self._engine.deterministic_edge( + self._engine.start_id, + self._store, + include_default=True, + ) + if not edge: + raise RuntimeError("Start 初始化后没有命中的表达式边或默认边") + node_config = await self._follow_edge(edge) + if self._manager is None: + raise RuntimeError("Workflow FlowManager 尚未初始化") + await self._manager.initialize(node_config) + logger.info(f"工作流模式启用: 当前节点={self._manager.current_node}") 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 + if content and not self._ended: + self._store.record("user", content) async def on_assistant_text_end( self, - turn_id: str, + _turn_id: str, content: str, interrupted: bool, ) -> None: - if not content or interrupted: + if not content or interrupted or self._ended: 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() + self._store.record("agent", content, completed_agent_turn=True) + manager = self._require_manager() + current = manager.current_node + if not current or self._engine.node_type(current) != "agent": + return + await self._refresh_agent_prompt(current) + edge = self._engine.deterministic_edge( + current, + self._store, + include_default=False, + ) + if edge and manager.current_node == current: + next_config = await self._follow_edge(edge) + await manager.set_node_from_config(next_config) - def _apply_node(self, node_id: str) -> None: + async def _refresh_agent_prompt(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=[], + await runtime.queue_frame( + LLMUpdateSettingsFrame( + delta=LLMSettings( + system_instruction=self._agent_role_message(node_id) ) - 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) + def _agent_role_message(self, node_id: str) -> str: + """Build one provider-compatible system instruction for an Agent stage.""" + stage_prompt = self._engine.prompt_for(node_id, self._store) + return f"{stage_prompt}\n\n[工作流执行规则]\n{AGENT_STAGE_INSTRUCTION}" + + async def _apply_agent_stage(self, node_id: str) -> None: + data = self._engine.data(node_id) + await self._emit_node_active(node_id) + if self._runtime and self._runtime.set_input_enabled: + self._runtime.set_input_enabled(True) + asr_id = str( + data.get("asrResourceId") + or self._engine.settings.get("defaultAsrResourceId") + or "" + ) + tts_id = str( + data.get("ttsResourceId") + or self._engine.settings.get("defaultTtsResourceId") + or "" + ) + runtime = self._require_runtime() + if runtime.switch_services: + await runtime.switch_services(asr_id or None, tts_id or None) + if runtime.set_knowledge_scope: + runtime.set_knowledge_scope( + { + "knowledge_base_id": data.get("knowledgeBaseId"), + "mode": data.get("knowledgeMode", "disabled"), + "top_n": int(data.get("knowledgeTopN") or 5), + "score_threshold": float(data.get("knowledgeScoreThreshold") or 0.0), + } + ) + + def _agent_config(self, node_id: str) -> NodeConfig: + data = self._engine.data(node_id) + strategy = ( + ContextStrategy.RESET + if data.get("contextPolicy") == "fresh" + else ContextStrategy.APPEND + ) + functions: list[FlowsFunctionSchema] = [] + for tool_id in data.get("toolIds") or []: + tool = self._tool_by_id.get(str(tool_id)) + if tool and tool.type == "http": + functions.append(self._flow_tool(tool, node_id)) + knowledge_function = self._knowledge_function(node_id) + if knowledge_function: + functions.append(knowledge_function) + for edge in self._engine.llm_edges(node_id): + functions.append(self._flow_edge(edge)) + return { + "name": node_id, + "role_message": self._agent_role_message(node_id), + "task_messages": [], + "functions": functions, + "context_strategy": ContextStrategyConfig(strategy=strategy), + "respond_immediately": True, + } + + def _terminal_config(self, node_id: str) -> NodeConfig: + return { + "name": node_id, + "role_message": self._store.render(self._engine.global_prompt()), + "task_messages": [], + "functions": [], + "context_strategy": ContextStrategyConfig(strategy=ContextStrategy.APPEND), + "respond_immediately": False, + } + + def _flow_tool(self, tool: RuntimeTool, node_id: str) -> FlowsFunctionSchema: + properties, required = self._tools.schema_parts(tool) + self._tools.register_secrets(tool) + + async def handler(args, _flow_manager): + try: + result = await self._tools.execute(tool, dict(args or {})) + except ToolExecutionError as exc: + return {"status": "error", "message": str(exc)} + if result.get("updated_variables"): + await self._refresh_agent_prompt(node_id) + edge = self._engine.deterministic_edge( + node_id, + self._store, + include_default=False, + ) + if edge: + return result, await self._follow_edge(edge) + return result + + return FlowsFunctionSchema( + name=tool.function_name, + description=tool.description or f"调用 {tool.name}", + properties=properties, + required=required, + handler=handler, + ) + + def _flow_edge(self, edge: dict) -> FlowsFunctionSchema: + async def handler(_args, _flow_manager): + return None, await self._follow_edge(edge) + + return FlowsFunctionSchema( + name=self._engine.edge_fn_name(edge), + description=self._engine.edge_description(edge), + properties={}, + required=[], + handler=handler, + ) + + def _knowledge_function(self, node_id: str) -> FlowsFunctionSchema | None: + data = self._engine.data(node_id) + knowledge_id = str(data.get("knowledgeBaseId") or "") + if not knowledge_id or data.get("knowledgeMode") != "on_demand": + return None + cfg = self._cfg or AssistantConfig(type="workflow") + knowledge = cfg.workflow_knowledge_bases.get(knowledge_id) + description = "在当前 Agent 绑定的知识库中检索资料。" + if knowledge: + description += f"知识库：{knowledge.name}。{knowledge.description}" + + async def handler(args, _flow_manager): + query = str((args or {}).get("query") or "").strip() + if not query: + return {"status": "error", "message": "检索问题为空"} + try: + async with SessionLocal() as session: + results = await search_knowledge( + session, + knowledge_id, + query, + top_k=int(data.get("knowledgeTopN") or 5), + score_threshold=float(data.get("knowledgeScoreThreshold") or 0.0), + ) + return {"status": "ok", "results": results} + except Exception as exc: # noqa: BLE001 - tool errors are returned to the LLM + logger.warning(f"Workflow 知识库检索失败:{exc}") + return {"status": "error", "message": "知识库检索暂时不可用"} + + return FlowsFunctionSchema( + name="search_knowledge_base", + description=description, + properties={ + "query": {"type": "string", "description": "完整问题或检索关键词"} + }, + required=["query"], + handler=handler, + ) + + async def _follow_edge(self, edge: dict) -> NodeConfig: + speech = self._engine.edge_transition_speech(edge) + if speech: + await self._require_runtime().queue_frame( + TTSSpeakFrame(self._store.render(speech), append_to_context=False) + ) + return await self._resolve_path(str(edge.get("target") or "")) + + async def _resolve_path(self, node_id: str) -> NodeConfig: + for _ in range(MAX_AUTOMATIC_HOPS): + node_type = self._engine.node_type(node_id) + if node_type == "agent": + await self._apply_agent_stage(node_id) + return self._agent_config(node_id) + if node_type == "end": + await self._enter_end(node_id) + return self._terminal_config(node_id) + if node_type == "action": + await self._enter_action(node_id) + elif node_type == "handoff": + await self._enter_handoff(node_id) + elif node_type == "start": + await self._emit_node_active(node_id) + else: + raise RuntimeError(f"工作流指向未知节点:{node_id}") + edge = self._engine.deterministic_edge( + node_id, + self._store, + include_default=True, + ) + if not edge: + raise RuntimeError(f"自动节点 {node_id} 没有命中的表达式边或默认边") + speech = self._engine.edge_transition_speech(edge) + if speech: + await self._require_runtime().queue_frame( + TTSSpeakFrame(self._store.render(speech), append_to_context=False) + ) + node_id = str(edge.get("target") or "") + raise RuntimeError("工作流连续自动跳转超过安全上限") + + async def _enter_action(self, node_id: str) -> None: + await self._emit_node_active(node_id) + data = self._engine.data(node_id) + tool_id = str(data.get("toolId") or "") + tool = self._tool_by_id.get(tool_id) + if not tool: + self._store.values["system__last_action_status"] = "error" + self._store.values["system__last_action_error"] = f"工具不存在:{tool_id}" + return + try: + arguments = self._store.render_data(data.get("arguments") or {}) + await self._tools.execute( + tool, + arguments, + result_assignments=data.get("resultAssignments") or {}, + ) + self._store.values["system__last_action_status"] = "ok" + self._store.values["system__last_action_error"] = "" + except (ToolExecutionError, ValueError) as exc: + self._store.values["system__last_action_status"] = "error" + self._store.values["system__last_action_error"] = str(exc)[:2048] + + async def _enter_handoff(self, node_id: str) -> None: + await self._emit_node_active(node_id) + data = self._engine.data(node_id) + message = self._store.render(str(data.get("message") or "")) + await self._require_runtime().queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "handoff-requested", + "nodeId": node_id, + "targetType": data.get("targetType", "human"), + "target": data.get("target", ""), + "message": message, + } + ) + ) + if message: + await self._require_runtime().queue_frame( + TTSSpeakFrame(message, append_to_context=False) + ) + self._store.values["system__handoff_status"] = "requested" + + async def _enter_end(self, node_id: str) -> None: + self._ended = True + await self._emit_node_active(node_id) + runtime = self._require_runtime() + if runtime.set_knowledge_scope: + runtime.set_knowledge_scope({"mode": "disabled"}) + if runtime.set_input_enabled: + runtime.set_input_enabled(False) + data = self._engine.data(node_id) + message = self._store.render(str(data.get("message") or "")) + scope = str(data.get("scope") or "session") + if scope == "flow": + await runtime.queue_frame( + OutputTransportMessageUrgentFrame( + message={"type": "flow-ended", "nodeId": node_id} + ) + ) + if message: + await runtime.queue_frame(TTSSpeakFrame(message, append_to_context=False)) + return + runtime.call_end.begin("workflow_completed") + if message: + runtime.call_end.arm_after_speech() + await runtime.queue_frame(TTSSpeakFrame(message, append_to_context=False)) + else: + await runtime.call_end.finish() async def _emit_node_active(self, node_id: str | None) -> None: if node_id: @@ -128,61 +406,12 @@ class WorkflowBrain(BaseBrain): ) ) - 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}") + def _require_manager(self) -> FlowManager: + if self._manager is None: + raise RuntimeError("Workflow FlowManager 尚未初始化") + return self._manager diff --git a/backend/services/config_resolver.py b/backend/services/config_resolver.py index 197ea19..c2362e6 100644 --- a/backend/services/config_resolver.py +++ b/backend/services/config_resolver.py @@ -12,7 +12,13 @@ from db.models import ( ModelResource, Tool, ) -from models import AssistantConfig, RuntimeTool +from models import ( + AssistantConfig, + RuntimeKnowledgeBase, + RuntimeModelResource, + RuntimeTool, +) +from services.node_specs import graph_references, normalize_graph from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -83,7 +89,7 @@ def _secret(resource: ModelResource | None, key: str, default: str = "") -> str: async def _tools_for(session: AsyncSession, assistant: Assistant) -> list[RuntimeTool]: - if assistant.type != "prompt": + if assistant.type not in {"prompt", "workflow"}: return [] tools = ( await session.execute( @@ -134,6 +140,43 @@ async def resolve_runtime_config( else None ) + graph = normalize_graph(assistant.graph or {}) if assistant.type == "workflow" else {} + refs = graph_references(graph) if graph else { + "model_resources": set(), + "knowledge_bases": set(), + } + workflow_resources: dict[str, RuntimeModelResource] = {} + if refs["model_resources"]: + resources = ( + await session.execute( + select(ModelResource).where(ModelResource.id.in_(refs["model_resources"])) + ) + ).scalars().all() + workflow_resources = { + resource.id: RuntimeModelResource( + id=resource.id, + name=resource.name, + capability=resource.capability, + interface_type=resource.interface_type, + values=resource.values or {}, + secrets=resource.secrets or {}, + ) + for resource in resources + if resource.enabled + } + workflow_knowledge: dict[str, RuntimeKnowledgeBase] = {} + if refs["knowledge_bases"]: + knowledge_rows = ( + await session.execute( + select(KnowledgeBase).where(KnowledgeBase.id.in_(refs["knowledge_bases"])) + ) + ).scalars().all() + workflow_knowledge = { + kb.id: RuntimeKnowledgeBase(id=kb.id, name=kb.name, description=kb.description) + for kb in knowledge_rows + if kb.status == "active" + } + return AssistantConfig( name=assistant.name, type=assistant.type, @@ -150,7 +193,9 @@ async def resolve_runtime_config( knowledge_base_description=knowledge_base.description if knowledge_base else "", knowledge_retrieval_config=assistant.knowledge_retrieval_config or {}, # workflow 图:仅 workflow 类型非空,引擎据此启用图驱动对话 - graph=(assistant.graph or {}) if assistant.type == "workflow" else {}, + graph=graph, + workflow_model_resources=workflow_resources, + workflow_knowledge_bases=workflow_knowledge, # 外部托管类型连接信息(DB 存真 key,直接注入) dify_api_url=str(_value(agent_resource, "apiUrl", assistant.api_url)), dify_api_key=_secret(agent_resource, "apiKey", assistant.api_key), diff --git a/backend/services/node_specs.py b/backend/services/node_specs.py index cd562c8..7c4b915 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -1,132 +1,108 @@ -"""工作流节点规格 + 图校验(对齐 dograh 的 node-spec / GraphConstraints 思路)。 - -当前实现 4 个核心节点:开始(startCall)/智能体(agentNode)/结束(endCall)/全局(globalNode)。 -本模块是「节点类型」的唯一事实源: - - /api/node-types 接口直接吐这里的规格; - - 助手保存时用这里的约束校验 workflow 图。 - -新增节点类型只需在 NODE_SPECS 里加一条并补充约束。前端 specs.ts 与此保持一致。 -""" +"""Workflow v3 node catalog, v2 compatibility normalization, and validation.""" from __future__ import annotations +from collections import defaultdict, deque +from copy import deepcopy from typing import Any -# 规格版本号:节点定义有破坏性变更时 +1,前端可据此判断是否需要刷新缓存。 -SPEC_VERSION = "2" -# 每个节点的图约束。None 表示不限制。 -# min_incoming / max_incoming:入边数量 -# min_outgoing / max_outgoing:出边数量 +SPEC_VERSION = "3" +NODE_TYPES = {"start", "agent", "action", "handoff", "end"} +EDGE_MODES = {"llm", "expression", "always"} +EXPRESSION_OPERATORS = { + "eq", + "neq", + "gt", + "gte", + "lt", + "lte", + "contains", + "in", + "exists", +} + NODE_SPECS: list[dict[str, Any]] = [ { - "name": "startCall", - "displayName": "开始", - "category": "call_node", - "description": "工作流入口,每个流程有且仅有一个。播放开场白,并用自己的提示词进行多轮对话,满足出边条件后流转。", + "name": "start", + "displayName": "Start", + "category": "control_node", + "description": "初始化会话、动态变量和全局观察器，可播放固定开场白。", "icon": "Play", "accent": "mint", "addable": False, "constraints": { "minIncoming": 0, "maxIncoming": 0, + "minOutgoing": 1, "minInstances": 1, "maxInstances": 1, }, "fields": [ - {"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, - }, + {"key": "name", "label": "节点名称", "type": "text", "default": "Start"}, + {"key": "greeting", "label": "固定开场白", "type": "textarea", "default": ""}, ], }, { - "name": "agentNode", - "displayName": "智能体节点", - "category": "call_node", - "description": "对话处理单元。按提示词与用户多轮交互,可有多个并通过条件边流转。", + "name": "agent", + "displayName": "Agent", + "category": "conversation_node", + "description": "阶段智能体：绑定上下文、工具、知识库及 ASR/TTS 资源。", "icon": "Bot", "accent": "sky", "addable": True, - "constraints": {"minIncoming": 1}, + "constraints": {"minIncoming": 1, "minOutgoing": 1}, "fields": [ - {"key": "name", "label": "节点名称", "type": "text", "default": "智能体节点"}, + {"key": "name", "label": "节点名称", "type": "text", "default": "Agent"}, { "key": "prompt", - "label": "节点提示词", + "label": "阶段提示词", "type": "textarea", "required": True, "default": "", }, - { - "key": "allowInterrupt", - "label": "允许用户打断", - "type": "switch", - "default": True, - }, - { - "key": "addGlobalPrompt", - "label": "应用全局提示词", - "type": "switch", - "default": True, - }, ], }, { - "name": "endCall", - "displayName": "结束", - "category": "call_node", - "description": "终止节点,礼貌结束对话。可有多个,均无出边。", + "name": "action", + "displayName": "Action", + "category": "execution_node", + "description": "确定性执行指定工具，并将结果字段写入会话动态变量。", + "icon": "Zap", + "accent": "peach", + "addable": True, + "constraints": {"minIncoming": 1, "minOutgoing": 1}, + "fields": [ + {"key": "name", "label": "节点名称", "type": "text", "default": "Action"}, + ], + }, + { + "name": "handoff", + "displayName": "Handoff", + "category": "execution_node", + "description": "转交其他 AI、人工、队列或电话；MVP 发送转交事件后继续路由。", + "icon": "PhoneForwarded", + "accent": "lavender", + "addable": True, + "constraints": {"minIncoming": 1, "minOutgoing": 1}, + "fields": [ + {"key": "name", "label": "节点名称", "type": "text", "default": "Handoff"}, + {"key": "target", "label": "转交目标", "type": "text", "default": ""}, + {"key": "message", "label": "转交提示", "type": "textarea", "default": ""}, + ], + }, + { + "name": "end", + "displayName": "End", + "category": "control_node", + "description": "结束 AI 流程或整个音视频会话。", "icon": "Flag", "accent": "rose", "addable": True, "constraints": {"minIncoming": 1, "minOutgoing": 0, "maxOutgoing": 0}, "fields": [ - {"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": "你是一个友好、专业的语音助手。请使用简短、自然、适合口语表达的句子。", - }, + {"key": "name", "label": "节点名称", "type": "text", "default": "End"}, + {"key": "message", "label": "固定结束语", "type": "textarea", "default": ""}, ], }, ] @@ -135,108 +111,303 @@ _SPEC_BY_NAME = {spec["name"]: spec for spec in NODE_SPECS} def node_types_response() -> dict[str, Any]: - """/api/node-types 的响应体(camelCase,直接喂前端)。""" return {"specVersion": SPEC_VERSION, "nodeTypes": NODE_SPECS} -def validate_graph(graph: dict[str, Any]) -> list[str]: - """校验 workflow 图,返回错误信息列表(空列表 = 通过)。 +def _edge_data_v3(edge: dict, source_type: str) -> dict: + data = deepcopy(edge.get("data") or {}) + if data.get("mode") in EDGE_MODES: + data.setdefault("priority", 10) + return data + condition = str(data.pop("condition", "") or "").strip() + transition = data.pop("transition_speech", data.get("transitionSpeech", "")) + data.update( + { + "mode": "llm" if condition and source_type == "agent" else "always", + "priority": 10, + "condition": condition, + "transitionSpeech": transition, + } + ) + return data - 基础规则(对齐 dograh 的核心不变量): - 1. 节点类型必须是已注册类型; - 2. 有且仅有一个 startCall; - 3. 至少有一个 endCall,全局节点最多一个; - 4. 边的 source/target 必须指向存在的节点; - 5. 入边/出边数量满足各节点类型的约束。 - 空图(无节点)视为草稿,直接放行,方便先存后编排。 - """ - errors: list[str] = [] - nodes = graph.get("nodes") or [] - edges = graph.get("edges") or [] +def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: + """Return a deep-copied v3 graph; preserve v3 IDs and migrate v2 semantics.""" + source = deepcopy(graph or {}) + if str(source.get("specVersion") or "") == SPEC_VERSION: + source.setdefault("settings", {}) + source.setdefault("nodes", []) + source.setdefault("edges", []) + return source - if not nodes: - return errors # 草稿:放行 - - node_ids: set[str] = set() - type_counts: dict[str, int] = {} - node_type_by_id: dict[str, str] = {} + nodes = source.get("nodes") or [] + edges = source.get("edges") or [] + global_prompt = "" + mapped_nodes: list[dict] = [] + type_by_id: dict[str, str] = {} + start_prompt_nodes: dict[str, str] = {} + type_map = { + "startCall": "start", + "agentNode": "agent", + "endCall": "end", + "start": "start", + "agent": "agent", + "action": "action", + "handoff": "handoff", + "end": "end", + } for node in nodes: - node_id = node.get("id") - node_type = node.get("type") - if not node_id: - errors.append("存在缺少 id 的节点") + old_type = str(node.get("type") or "") + data = deepcopy(node.get("data") or {}) + if old_type == "globalNode": + global_prompt = str(data.get("prompt") or "") continue - if node_id in node_ids: - errors.append(f"节点 id 重复:{node_id}") - node_ids.add(node_id) - if node_type not in _SPEC_BY_NAME: - errors.append(f"未知节点类型:{node_type}(节点 {node_id})") - continue - node_type_by_id[node_id] = node_type - type_counts[node_type] = type_counts.get(node_type, 0) + 1 + new_type = type_map.get(old_type, old_type) + migrated = deepcopy(node) + migrated["type"] = new_type + if new_type == "end": + data["message"] = data.pop("message", data.pop("prompt", "")) + data.setdefault("scope", "session") + elif new_type == "agent": + data.setdefault("contextPolicy", "inherit") + data.setdefault("toolIds", []) + data.setdefault("knowledgeMode", "disabled") + elif new_type == "start": + prompt = str(data.pop("prompt", "") or "").strip() + if prompt: + start_prompt_nodes[str(node.get("id"))] = prompt + for key in ("allowInterrupt", "addGlobalPrompt"): + data.pop(key, None) + migrated["data"] = data + mapped_nodes.append(migrated) + if migrated.get("id"): + type_by_id[str(migrated["id"])] = new_type - start_count = type_counts.get("startCall", 0) - if start_count == 0: - errors.append("工作流必须有一个「开始」节点") - elif start_count > 1: - errors.append("工作流只能有一个「开始」节点") + mapped_edges: list[dict] = [] + for edge in edges: + migrated = deepcopy(edge) + migrated["data"] = _edge_data_v3( + migrated, type_by_id.get(str(migrated.get("source")), "") + ) + mapped_edges.append(migrated) - 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, - "实例", + # A v2 Start was conversational. Insert a synthetic Agent so its prompt remains active. + for start_id, prompt in start_prompt_nodes.items(): + synthetic_id = f"{start_id}-migrated-agent" + start_node = next((n for n in mapped_nodes if n.get("id") == start_id), None) + position = (start_node or {}).get("position") or {"x": 100, "y": 120} + mapped_nodes.append( + { + "id": synthetic_id, + "type": "agent", + "position": {"x": position.get("x", 100) + 300, "y": position.get("y", 120)}, + "data": { + "name": "迁移的开场 Agent", + "prompt": prompt, + "contextPolicy": "inherit", + "toolIds": [], + "knowledgeMode": "disabled", + }, + } + ) + for edge in mapped_edges: + if edge.get("source") == start_id: + edge["source"] = synthetic_id + edge["data"] = _edge_data_v3(edge, "agent") + if str(edge["data"].get("condition") or "").strip(): + edge["data"]["mode"] = "llm" + mapped_edges.append( + { + "id": f"e-{start_id}-{synthetic_id}", + "source": start_id, + "target": synthetic_id, + "data": {"mode": "always", "priority": 0, "transitionSpeech": ""}, + } ) - # 统计入边/出边 - incoming: dict[str, int] = {nid: 0 for nid in node_ids} - outgoing: dict[str, int] = {nid: 0 for nid in node_ids} - for edge in edges: - source = edge.get("source") - target = edge.get("target") - if source not in node_ids: - errors.append(f"连线指向了不存在的源节点:{source}") - continue - if target not in node_ids: - errors.append(f"连线指向了不存在的目标节点:{target}") - continue - outgoing[source] += 1 - incoming[target] += 1 + settings = deepcopy(source.get("settings") or {}) + settings.setdefault("globalPrompt", global_prompt) + settings.setdefault("defaultAsrResourceId", "") + settings.setdefault("defaultTtsResourceId", "") + return { + "specVersion": 3, + "settings": settings, + "nodes": mapped_nodes, + "edges": mapped_edges, + **({"viewport": deepcopy(source["viewport"])} if source.get("viewport") else {}), + } - for node_id, node_type in node_type_by_id.items(): - constraints = _SPEC_BY_NAME[node_type]["constraints"] - name = node_type - _check_count(errors, incoming[node_id], constraints, "Incoming", name, "入边") - _check_count(errors, outgoing[node_id], constraints, "Outgoing", name, "出边") +def _validate_expression(expression: Any) -> list[str]: + if not isinstance(expression, dict): + return ["表达式条件不能为空"] + combinator = expression.get("combinator", "and") + if combinator not in {"and", "or"}: + return ["表达式组合方式必须是 and 或 or"] + rules = expression.get("rules") + if not isinstance(rules, list) or not rules: + return ["表达式至少需要一条规则"] + errors = [] + for rule in rules: + if not isinstance(rule, dict) or not rule.get("variable"): + errors.append("表达式规则缺少变量") + elif rule.get("operator") not in EXPRESSION_OPERATORS: + errors.append(f"不支持的表达式运算符:{rule.get('operator')}") return errors -def _check_count( - errors: list[str], - actual: int, - constraints: dict[str, int], - suffix: str, - node_type: str, - label: str, -) -> None: - lo = constraints.get(f"min{suffix}") - hi = constraints.get(f"max{suffix}") - display = _SPEC_BY_NAME[node_type]["displayName"] - if lo is not None and actual < lo: - errors.append(f"「{display}」节点{label}数量不能少于 {lo}(当前 {actual})") - if hi is not None and actual > hi: - errors.append(f"「{display}」节点{label}数量不能多于 {hi}(当前 {actual})") +def validate_graph(graph: dict[str, Any]) -> list[str]: + graph = normalize_graph(graph) + nodes = graph.get("nodes") or [] + edges = graph.get("edges") or [] + if not nodes: + return [] + + errors: list[str] = [] + node_by_id: dict[str, dict] = {} + counts: dict[str, int] = defaultdict(int) + for node in nodes: + node_id = str(node.get("id") or "") + node_type = str(node.get("type") or "") + if not node_id: + errors.append("存在缺少 id 的节点") + continue + if node_id in node_by_id: + errors.append(f"节点 id 重复:{node_id}") + if node_type not in NODE_TYPES: + errors.append(f"未知节点类型:{node_type}(节点 {node_id})") + node_by_id[node_id] = node + counts[node_type] += 1 + + if counts["start"] != 1: + errors.append("工作流必须有且仅有一个 Start 节点") + if counts["end"] < 1: + errors.append("工作流至少需要一个 End 节点") + + incoming: dict[str, int] = defaultdict(int) + outgoing: dict[str, int] = defaultdict(int) + adj: dict[str, list[str]] = defaultdict(list) + auto_adj: dict[str, list[str]] = defaultdict(list) + priorities: dict[str, set[int]] = defaultdict(set) + always_counts: dict[str, int] = defaultdict(int) + for edge in edges: + edge_id = str(edge.get("id") or "") + source_id = str(edge.get("source") or "") + target_id = str(edge.get("target") or "") + if source_id not in node_by_id: + errors.append(f"边 {edge_id} 指向不存在的源节点:{source_id}") + continue + if target_id not in node_by_id: + errors.append(f"边 {edge_id} 指向不存在的目标节点:{target_id}") + continue + if source_id == target_id and node_by_id[source_id].get("type") != "agent": + errors.append(f"自动节点不能自连:{source_id}") + data = edge.get("data") or {} + mode = data.get("mode") + if mode not in EDGE_MODES: + errors.append(f"边 {edge_id} 的判断模式无效:{mode}") + if mode == "llm" and node_by_id[source_id].get("type") != "agent": + errors.append(f"LLM 判断边只能从 Agent 发出:{edge_id}") + if mode == "llm" and not str(data.get("condition") or "").strip(): + errors.append(f"LLM 判断边缺少自然语言条件:{edge_id}") + if mode == "expression": + errors.extend(f"边 {edge_id}:{item}" for item in _validate_expression(data.get("expression"))) + try: + priority = int(data.get("priority", 10)) + except (TypeError, ValueError): + errors.append(f"边 {edge_id} 的优先级必须是整数") + priority = 10 + if priority in priorities[source_id]: + errors.append(f"节点 {source_id} 的出边优先级不能重复:{priority}") + priorities[source_id].add(priority) + if mode == "always": + always_counts[source_id] += 1 + if always_counts[source_id] > 1: + errors.append(f"节点 {source_id} 最多只能有一条默认边") + incoming[target_id] += 1 + outgoing[source_id] += 1 + adj[source_id].append(target_id) + if node_by_id[source_id].get("type") != "agent" and node_by_id[target_id].get("type") != "agent": + auto_adj[source_id].append(target_id) + + for node_id, node in node_by_id.items(): + spec = _SPEC_BY_NAME.get(str(node.get("type"))) + if not spec: + continue + constraints = spec["constraints"] + for actual, suffix, label in ( + (incoming[node_id], "Incoming", "入边"), + (outgoing[node_id], "Outgoing", "出边"), + ): + lo = constraints.get(f"min{suffix}") + hi = constraints.get(f"max{suffix}") + if lo is not None and actual < lo: + errors.append(f"节点 {node_id} 的{label}不能少于 {lo}") + if hi is not None and actual > hi: + errors.append(f"节点 {node_id} 的{label}不能多于 {hi}") + if node.get("type") in {"start", "action", "handoff"} and always_counts[node_id] != 1: + errors.append(f"自动节点 {node_id} 必须有且仅有一条默认边") + + start_id = next((nid for nid, n in node_by_id.items() if n.get("type") == "start"), None) + if start_id: + reached = {start_id} + queue = deque([start_id]) + while queue: + current = queue.popleft() + for target in adj[current]: + if target not in reached: + reached.add(target) + queue.append(target) + for node_id in node_by_id.keys() - reached: + errors.append(f"节点不可从 Start 到达:{node_id}") + + # Reject cycles made only of instantaneous nodes; Agent cycles are valid waits. + visiting: set[str] = set() + visited: set[str] = set() + + def visit(node_id: str) -> bool: + if node_id in visiting: + return True + if node_id in visited: + return False + visiting.add(node_id) + for target in auto_adj[node_id]: + if visit(target): + return True + visiting.remove(node_id) + visited.add(node_id) + return False + + if any(visit(node_id) for node_id, node in node_by_id.items() if node.get("type") != "agent"): + errors.append("Start/Action/Handoff/End 之间不能形成无等待循环") + return list(dict.fromkeys(errors)) + + +def graph_references(graph: dict[str, Any]) -> dict[str, set[str]]: + """Collect externally referenced IDs for save/runtime validation.""" + normalized = normalize_graph(graph) + settings = normalized.get("settings") or {} + resources = { + str(value) + for value in ( + settings.get("defaultAsrResourceId"), + settings.get("defaultTtsResourceId"), + ) + if value + } + tools: set[str] = set() + knowledge: set[str] = set() + for node in normalized.get("nodes") or []: + data = node.get("data") or {} + for resource_id in (data.get("asrResourceId"), data.get("ttsResourceId")): + if resource_id: + resources.add(str(resource_id)) + for tool_id in data.get("toolIds") or []: + tools.add(str(tool_id)) + if data.get("toolId"): + tools.add(str(data["toolId"])) + if data.get("knowledgeBaseId"): + knowledge.add(str(data["knowledgeBaseId"])) + return {"model_resources": resources, "tools": tools, "knowledge_bases": knowledge} diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 6394d20..b62eeca 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -23,6 +23,7 @@ from services.pipecat.call_lifecycle import ( EndCallAfterSpeechProcessor, ) from services.pipecat.service_factory import ( + config_with_resource, create_realtime_service, create_stt, create_tts, @@ -32,6 +33,7 @@ from services.knowledge import search as search_knowledge from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.flows import FlowsFunctionSchema from pipecat.frames.frames import ( EndFrame, InputTransportMessageFrame, @@ -40,6 +42,7 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMContextFrame, LLMTextFrame, + ManuallySwitchServiceFrame, LLMMessagesAppendFrame, OutputTransportMessageUrgentFrame, TextFrame, @@ -48,6 +51,7 @@ from pipecat.frames.frames import ( UserImageRequestFrame, ) from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.service_switcher import ServiceSwitcher from pipecat.pipeline.worker import PipelineParams, PipelineWorker from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import ( @@ -397,15 +401,59 @@ class KnowledgeRetrievalProcessor(FrameProcessor): self._knowledge_base_id = knowledge_base_id self._top_n = top_n self._score_threshold = score_threshold + self._mode = "automatic" if knowledge_base_id else "disabled" self._last_signature = "" + def set_scope(self, scope: dict) -> None: + self._knowledge_base_id = scope.get("knowledge_base_id") or None + self._mode = str(scope.get("mode") or "disabled") + self._top_n = int(scope.get("top_n") or 5) + self._score_threshold = float(scope.get("score_threshold") or 0.0) + self._last_signature = "" + + def _clear_context(self, messages: list[dict]) -> None: + # Remove the legacy Workflow knowledge message so an in-flight context + # created before this compatibility fix cannot keep sending that role. + messages[:] = [ + message + for message in messages + if not ( + message.get("role") == "developer" + and KNOWLEDGE_CONTEXT_MARKER in str(message.get("content") or "") + ) + ] + system_message = next( + (message for message in messages if message.get("role") == "system"), + None, + ) + if system_message is not None: + content = str(system_message.get("content") or "") + system_message["content"] = content.split(KNOWLEDGE_CONTEXT_MARKER, 1)[0].rstrip() + + def _set_context(self, messages: list[dict], block: str) -> None: + """Store retrieved knowledge in a provider-compatible system message.""" + self._clear_context(messages) + system_message = next( + (message for message in messages if message.get("role") == "system"), + None, + ) + if system_message is None: + messages.insert(0, {"role": "system", "content": block}) + return + content = str(system_message.get("content") or "").rstrip() + system_message["content"] = f"{content}\n\n{block}" if content else block + async def process_frame(self, frame, direction: FrameDirection): await super().process_frame(frame, direction) - if not self._knowledge_base_id or not isinstance(frame, LLMContextFrame): + if not isinstance(frame, LLMContextFrame): await self.push_frame(frame, direction) return messages = frame.context.get_messages() + if self._mode != "automatic" or not self._knowledge_base_id: + self._clear_context(messages) + await self.push_frame(frame, direction) + return user_messages = [message for message in messages if message.get("role") == "user"] if not user_messages: await self.push_frame(frame, direction) @@ -435,13 +483,7 @@ class KnowledgeRetrievalProcessor(FrameProcessor): for index, item in enumerate(results) ) or "未检索到相关资料。" block = f"{KNOWLEDGE_CONTEXT_MARKER}\n当前问题的知识库检索结果：\n{sources}" - system_message = next((message for message in messages if message.get("role") == "system"), None) - if system_message is None: - messages.insert(0, {"role": "system", "content": block}) - else: - content = str(system_message.get("content") or "") - base = content.split(KNOWLEDGE_CONTEXT_MARKER, 1)[0].rstrip() - system_message["content"] = f"{base}\n\n{block}" if base else block + self._set_context(messages, block) await self.push_frame(frame, direction) @@ -457,7 +499,6 @@ class PassthroughLLMAssistantAggregator(LLMAssistantAggregator): self._stream_turn_id: str | None = None self._stream_timestamp = "" self._stream_text = "" - async def process_frame(self, frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -505,6 +546,49 @@ class PassthroughLLMAssistantAggregator(LLMAssistantAggregator): self._stream_text = "" +class WorkflowAggregatorPair: + """Small public-shape adapter required by Pipecat FlowManager.""" + + def __init__(self, user_aggregator, assistant_aggregator): + self._user = user_aggregator + self._assistant = assistant_aggregator + + def user(self): + return self._user + + def assistant(self): + return self._assistant + + +def _workflow_service_switcher( + cfg: AssistantConfig, capability: str, base_service: FrameProcessor +): + """Build one switcher and an ID lookup for every referenced voice resource.""" + create = create_stt if capability == "ASR" else create_tts + settings = cfg.graph.get("settings") or {} + default_key = ( + "defaultAsrResourceId" if capability == "ASR" else "defaultTtsResourceId" + ) + default_id = str(settings.get(default_key) or "") + services_by_id = {} + for resource_id, resource in cfg.workflow_model_resources.items(): + if resource.capability != capability: + continue + services_by_id[resource_id] = ( + base_service + if resource_id == default_id + else create(config_with_resource(cfg, resource)) + ) + primary = services_by_id.get(default_id, base_service) + services = [primary] + services.extend( + service for service in services_by_id.values() if service is not primary + ) + if base_service is not primary: + services.append(base_service) + return ServiceSwitcher(services=services), services_by_id, primary + + async def run_pipeline( transport, cfg: AssistantConfig, @@ -545,8 +629,35 @@ async def run_pipeline( ) return - stt = create_stt(cfg) - tts = create_tts(cfg) + graph_settings = cfg.graph.get("settings") or {} + default_asr_resource = cfg.workflow_model_resources.get( + str(graph_settings.get("defaultAsrResourceId") or "") + ) + default_tts_resource = cfg.workflow_model_resources.get( + str(graph_settings.get("defaultTtsResourceId") or "") + ) + stt = create_stt( + config_with_resource(cfg, default_asr_resource) + if cfg.type == "workflow" and default_asr_resource + else cfg + ) + tts = create_tts( + config_with_resource(cfg, default_tts_resource) + if cfg.type == "workflow" and default_tts_resource + else cfg + ) + stt_processor = stt + tts_processor = tts + stt_services: dict[str, FrameProcessor] = {} + tts_services: dict[str, FrameProcessor] = {} + current_voice_services: dict[str, FrameProcessor] = {"asr": stt, "tts": tts} + if cfg.type == "workflow": + stt_processor, stt_services, current_voice_services["asr"] = ( + _workflow_service_switcher(cfg, "ASR", stt) + ) + tts_processor, tts_services, current_voice_services["tts"] = ( + _workflow_service_switcher(cfg, "TTS", tts) + ) greeting = await brain.greeting(cfg) system_content = brain.system_prompt(cfg) @@ -594,8 +705,13 @@ async def run_pipeline( return "\n\n".join(part for part in [text, *hints] if part) context = LLMContext( - messages=[{"role": "system", "content": with_vision_hint(system_content)}] + messages=( + [] + if cfg.type == "workflow" + else [{"role": "system", "content": with_vision_hint(system_content)}] + ) ) + input_state = {"enabled": True} # LLM 槽由大脑提供:本地模型或 Dify/FastGPT 外部托管适配器。 llm = brain.build_llm(cfg, context) user_aggregator = LLMUserAggregator( @@ -604,7 +720,9 @@ async def run_pipeline( vad_analyzer=create_vad_analyzer(cfg.turnConfig), user_mute_strategies=[ FunctionCallUserMuteStrategy(), - CallEndingUserMuteStrategy(lambda: call_end.ending), + CallEndingUserMuteStrategy( + lambda: call_end.ending or not input_state["enabled"] + ), ], user_turn_strategies=create_user_turn_strategies( cfg.turnConfig, @@ -613,7 +731,9 @@ async def run_pipeline( ), ) assistant_aggregator = PassthroughLLMAssistantAggregator(context) - text_input = TextInputProcessor(should_ignore_input=lambda: call_end.ending) + text_input = TextInputProcessor( + should_ignore_input=lambda: call_end.ending or not input_state["enabled"] + ) vision_capture = VisionCaptureProcessor() knowledge_retrieval = KnowledgeRetrievalProcessor( automatic_knowledge_id, @@ -723,14 +843,54 @@ async def run_pipeline( } ) - if vision_enabled: + flow_global_functions = [] + if cfg.type == "workflow" and vision_enabled: + async def flow_fetch_user_image(args, _flow_manager): + question = str((args or {}).get("question") or "请描述当前画面。") + user_id = vision_state.get("client_id") + if not user_id: + return { + "status": "no_video_client", + "message": "当前还没有可用的摄像头视频流。", + } + request = UserImageRequestFrame( + user_id=user_id, + text=question, + append_to_context=False, + function_name=VISION_TOOL_NAME, + ) + try: + frame = await vision_capture.request_image(llm, request) + observation = await _analyze_image_with_vision_model(cfg, frame, question) + return { + "status": "ok", + "question": question, + "observation": observation or "视觉模型没有返回有效观察结果。", + } + except asyncio.TimeoutError: + return {"status": "timeout", "message": "等待摄像头视频帧超时。"} + except Exception as exc: # noqa: BLE001 - return tool errors to the LLM + logger.warning(f"Workflow 视觉理解失败:{exc}") + return {"status": "error", "message": "视觉理解暂时不可用。"} + + flow_global_functions.append( + FlowsFunctionSchema( + name=VISION_TOOL_NAME, + description=vision_schema.description, + properties=vision_schema.properties, + required=vision_schema.required, + handler=flow_fetch_user_image, + ) + ) + + if vision_enabled and cfg.type != "workflow": llm.register_function(VISION_TOOL_NAME, fetch_user_image) if cfg.knowledge_base_id and knowledge_mode == "on_demand": llm.register_function(KNOWLEDGE_TOOL_NAME, search_bound_knowledge) def set_visible_tools(schemas: list[FunctionSchema] | None = None) -> None: tools = list(schemas or []) - if vision_enabled: + if vision_enabled and cfg.type != "workflow": tools.append(vision_schema) if cfg.knowledge_base_id and knowledge_mode == "on_demand": tools.append(knowledge_schema) @@ -751,7 +911,7 @@ async def run_pipeline( transport.input(), vision_capture, text_input, - stt, + stt_processor, user_aggregator, knowledge_retrieval, llm, @@ -759,7 +919,7 @@ async def run_pipeline( # Pipecat commits the generated prefix immediately instead of # waiting for a TTS provider to emit spoken-text/timestamp frames. assistant_aggregator, - tts, + tts_processor, EndCallAfterSpeechProcessor(call_end), ConversationHistoryProcessor(recorder), transport.output(), @@ -774,6 +934,33 @@ async def run_pipeline( enable_rtvi=False, ) worker_holder["worker"] = worker + default_voice_services = dict(current_voice_services) + + async def switch_workflow_services( + asr_resource_id: str | None, + tts_resource_id: str | None, + ) -> None: + requested = ( + ("asr", stt_services, asr_resource_id), + ("tts", tts_services, tts_resource_id), + ) + for kind, services, resource_id in requested: + target = services.get(resource_id) if resource_id else default_voice_services[kind] + if target is None: + raise ValueError(f"Workflow {kind.upper()} 资源未加载:{resource_id}") + if current_voice_services[kind] is target: + continue + await worker.queue_frame(ManuallySwitchServiceFrame(service=target)) + current_voice_services[kind] = target + await worker.queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "service-switched", + "capability": kind.upper(), + "resourceId": resource_id, + } + ) + ) async def queue_transcript(role: str, content: str, timestamp: str) -> None: if content: @@ -810,6 +997,16 @@ async def run_pipeline( set_system_prompt=set_system_prompt, set_tools=set_visible_tools, call_end=call_end, + worker=worker, + context_aggregator=WorkflowAggregatorPair( + user_aggregator, + assistant_aggregator, + ), + transport=transport, + switch_services=switch_workflow_services, + set_knowledge_scope=knowledge_retrieval.set_scope, + set_input_enabled=lambda enabled: input_state.__setitem__("enabled", enabled), + flow_global_functions=flow_global_functions, ), ) diff --git a/backend/services/pipecat/service_factory.py b/backend/services/pipecat/service_factory.py index ce2ff59..907bc68 100644 --- a/backend/services/pipecat/service_factory.py +++ b/backend/services/pipecat/service_factory.py @@ -5,7 +5,7 @@ """ from loguru import logger -from models import AssistantConfig +from models import AssistantConfig, RuntimeModelResource from pipecat.services.openai.llm import OpenAILLMService from pipecat.services.openai.stt import OpenAISTTService @@ -26,6 +26,35 @@ from services.pipecat.xfyun_tts import DEFAULT_XFYUN_TTS_URL, XfyunTTSService TTS_STOP_FRAME_TIMEOUT_S = 1.0 +def config_with_resource( + cfg: AssistantConfig, resource: RuntimeModelResource +) -> AssistantConfig: + """Return a call-local config view for one workflow ASR/TTS resource.""" + result = cfg.model_copy(deep=True) + values = resource.values or {} + secrets = resource.secrets or {} + if resource.capability == "ASR": + result.asr = str(values.get("modelId") or "") + result.stt_language = str(values.get("language") or "") + result.stt_interface_type = resource.interface_type + result.stt_values = values + result.stt_secrets = secrets + result.stt_api_key = str(secrets.get("apiKey") or "") + result.stt_base_url = str(values.get("apiUrl") or "") + elif resource.capability == "TTS": + result.tts_model = str(values.get("modelId") or "") + result.voice = str(values.get("voice") or "") + result.tts_speed = float(values.get("speed") or 1.0) + result.tts_interface_type = resource.interface_type + result.tts_values = values + result.tts_secrets = secrets + result.tts_api_key = str(secrets.get("apiKey") or "") + result.tts_base_url = str(values.get("apiUrl") or "") + else: + raise ValueError(f"工作流语音资源能力无效:{resource.capability}") + return result + + def _require(value: str, label: str) -> str: if value: return value diff --git a/backend/services/runtime_variables.py b/backend/services/runtime_variables.py index 84dc52f..ed1f14a 100644 --- a/backend/services/runtime_variables.py +++ b/backend/services/runtime_variables.py @@ -175,9 +175,9 @@ def prepare_dynamic_config( trusted_values: dict[str, Any] | None = None, ) -> AssistantConfig: """Validate and merge one call's values without mutating stored config.""" - if cfg.type != "prompt": + if cfg.type not in {"prompt", "workflow"}: if client_values: - raise DynamicVariableError("动态变量目前仅支持 prompt 助手") + raise DynamicVariableError("动态变量仅支持 prompt 和 workflow 助手") return cfg supplied = _validate_public(client_values) @@ -214,9 +214,13 @@ def prepare_dynamic_config( prepared = cfg.model_copy(deep=True) prepared.dynamic_variables = merged prepared.conversation_id = conversation_id - # Validate prompt and greeting before media resources are allocated. + # Validate top-level prompt/greeting before media resources are allocated. + # Workflow node templates are rendered lazily as their nodes become active. store = DynamicVariableStore.from_config(prepared) - store.render(prepared.prompt) + if prepared.type == "prompt": + store.render(prepared.prompt) + elif prepared.type == "workflow": + store.render_data(prepared.graph) store.render(prepared.greeting) return prepared diff --git a/backend/services/tool_executor.py b/backend/services/tool_executor.py new file mode 100644 index 0000000..532292d --- /dev/null +++ b/backend/services/tool_executor.py @@ -0,0 +1,154 @@ +"""Reusable deterministic tool execution shared by Prompt, Agent, and Action.""" + +from __future__ import annotations + +from copy import deepcopy +from typing import Any +from urllib.parse import quote + +import httpx +from models import RuntimeTool + +from services.runtime_variables import ( + DynamicVariableError, + DynamicVariableStore, + value_at_path, +) + + +class ToolExecutionError(RuntimeError): + pass + + +class ToolExecutor: + def __init__(self, store: DynamicVariableStore): + self.store = store + + def register_secrets(self, tool: RuntimeTool) -> None: + dynamic = (tool.secrets or {}).get("dynamic_variables") or {} + for name, value in dynamic.items(): + if not str(name).startswith("secret__"): + raise DynamicVariableError(f"工具密钥变量必须以 secret__ 开头: {name}") + self.store.secrets[str(name)] = str(value) + + @staticmethod + def schema_parts(tool: RuntimeTool) -> tuple[dict[str, Any], list[str]]: + config = (tool.definition or {}).get("config") or {} + parameters = list(config.get("parameters") or []) + properties = { + str(parameter.get("name")): { + "type": str(parameter.get("type") or "string"), + "description": str(parameter.get("description") or ""), + } + for parameter in parameters + if parameter.get("name") + } + required = [ + str(parameter["name"]) + for parameter in parameters + if parameter.get("name") and parameter.get("required", True) + ] + return properties, required + + async def execute( + self, + tool: RuntimeTool, + arguments: dict[str, Any] | None = None, + *, + result_assignments: dict[str, str] | None = None, + ) -> dict[str, Any]: + self.register_secrets(tool) + if tool.type != "http": + raise ToolExecutionError(f"Action 暂不支持工具类型: {tool.type}") + return await self._execute_http( + tool, + dict(arguments or {}), + result_assignments=result_assignments, + ) + + async def _execute_http( + self, + tool: RuntimeTool, + arguments: dict[str, Any], + *, + result_assignments: dict[str, str] | None, + ) -> dict[str, Any]: + config = (tool.definition or {}).get("config") or {} + parameters = list(config.get("parameters") or []) + url = self.store.render(str(config.get("url") or "")) + configured_headers = self.store.render_data( + deepcopy(config.get("headers") or {}), allow_secrets=True + ) + secret_headers = self.store.render_data( + deepcopy((tool.secrets or {}).get("headers") or {}), allow_secrets=True + ) + headers: dict[str, str] = {} + query: dict[str, object] = {} + body = self.store.render_data(deepcopy(config.get("body") or {})) + + for parameter in parameters: + name = str(parameter.get("name") or "") + if not name or name not in arguments: + continue + value = arguments[name] + location = str(parameter.get("location") or "body") + if location == "path": + url = url.replace(f"{{{name}}}", quote(str(value), safe="")) + elif location == "query": + query[name] = value + elif location == "header": + headers[name] = str(value) + else: + body[name] = value + + headers.update({str(key): str(value) for key, value in configured_headers.items()}) + headers.update({str(key): str(value) for key, value in secret_headers.items()}) + try: + async with httpx.AsyncClient( + timeout=float(config.get("timeout_seconds") or 15), + follow_redirects=False, + ) as client: + response = await client.request( + str(config.get("method") or "GET"), + url, + headers=headers, + params=query, + json=body if body else None, + ) + response.raise_for_status() + except httpx.TimeoutException as exc: + raise ToolExecutionError("HTTP 工具调用超时") from exc + except httpx.HTTPStatusError as exc: + raise ToolExecutionError(f"HTTP 工具返回错误状态:{exc.response.status_code}") from exc + except httpx.RequestError as exc: + raise ToolExecutionError(f"HTTP 工具调用失败:{exc}") from exc + + if len(response.content) > 1_000_000: + raise ToolExecutionError("HTTP 工具响应超过 1 MB 限制") + try: + payload: Any = response.json() + except ValueError: + payload = {"text": response.text[:8000]} + + assignments = ( + result_assignments + if result_assignments is not None + else config.get("dynamic_variable_assignments") or {} + ) + updated: list[str] = [] + for variable_name, path in assignments.items(): + try: + value = value_at_path(payload, str(path)) + except KeyError: + try: + value = value_at_path({"response": payload}, str(path)) + except KeyError: + continue + self.store.assign(str(variable_name), value) + updated.append(str(variable_name)) + return { + "status": "ok", + "status_code": response.status_code, + "data": payload, + "updated_variables": updated, + } diff --git a/backend/services/workflow_engine.py b/backend/services/workflow_engine.py index 29e13af..94b94f4 100644 --- a/backend/services/workflow_engine.py +++ b/backend/services/workflow_engine.py @@ -1,170 +1,140 @@ -"""工作流图引擎(第一版)。 - -对应 dograh 的 pipecat_engine.py,极简实现: - - 单个 startCall 入口,开场白来自该节点; - - agentNode 用各自的 prompt 驱动多轮对话; - - globalNode 不参与连线,按节点开关向会话节点注入统一提示词; - - 每轮助手回复后,用一次轻量 LLM「路由」判断是否满足某条出边的 condition, - 满足则切换当前节点(linear = 单边;branching = 多边按条件分流); - - 到达 endCall 播放结束语并停止路由。 - -只读图结构,不持有对话状态(当前节点由 pipeline 维护),便于单测。 -""" +"""Pure Workflow v3 graph queries and deterministic edge evaluation.""" from __future__ import annotations import re from typing import Any -from loguru import logger +from services.node_specs import normalize_graph +from services.runtime_variables import DynamicVariableStore class WorkflowEngine: def __init__(self, graph: dict[str, Any]): - nodes = graph.get("nodes") or [] - self.nodes: dict[str, dict] = {n["id"]: n for n in nodes if n.get("id")} - self.edges: list[dict] = graph.get("edges") or [] - self.start_id: str | None = next( - (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"), + self.graph = normalize_graph(graph) + self.settings = self.graph.get("settings") or {} + self.nodes: dict[str, dict] = { + str(node["id"]): node + for node in self.graph.get("nodes") or [] + if node.get("id") + } + self.edges: list[dict] = list(self.graph.get("edges") or []) + self.start_id = next( + (node_id for node_id, node in self.nodes.items() if node.get("type") == "start"), None, ) - # ---- 结构查询 ---- - def node_type(self, nid: str | None) -> str | None: - return self.nodes.get(nid or "", {}).get("type") + def has_graph(self) -> bool: + return bool(self.start_id) - def data(self, nid: str | None) -> dict: - return self.nodes.get(nid or "", {}).get("data") or {} + def node_type(self, node_id: str | None) -> str | None: + return self.nodes.get(node_id or "", {}).get("type") - def name(self, nid: str | None) -> str: - return self.data(nid).get("name") or (self.node_type(nid) or "") + def data(self, node_id: str | None) -> dict: + return self.nodes.get(node_id or "", {}).get("data") or {} - def outgoing(self, nid: str | None) -> list[dict]: - return [e for e in self.edges if e.get("source") == nid] + def name(self, node_id: str | None) -> str: + return str(self.data(node_id).get("name") or self.node_type(node_id) or "") + + def outgoing(self, node_id: str | None) -> list[dict]: + result = [edge for edge in self.edges if edge.get("source") == node_id] + return sorted(result, key=lambda edge: int((edge.get("data") or {}).get("priority", 10))) + + def edge_mode(self, edge: dict) -> str: + return str((edge.get("data") or {}).get("mode") or "always") def edge_fn_name(self, edge: dict) -> str: - """每条边对应一个 LLM 函数名(稳定、合法标识符)。""" raw = edge.get("id") or f"{edge.get('source')}_{edge.get('target')}" slug = re.sub(r"[^a-z0-9]+", "_", str(raw).lower()).strip("_") return f"goto_{slug or 'next'}" - def edge_condition(self, edge: dict) -> str: - return (edge.get("data") or {}).get("condition") or "" + def edge_description(self, edge: dict) -> str: + data = edge.get("data") or {} + target = self.name(str(edge.get("target") or "")) + condition = str(data.get("condition") or "").strip() + if condition: + return f"当满足以下条件时转到「{target}」：{condition}" + return f"当当前阶段任务完成时转到「{target}」。" def edge_transition_speech(self, edge: dict | None) -> str: - """命中该边、切换节点瞬间播报的过渡语(可选,掩盖延迟,不写入上下文)。""" if not edge: return "" - return (edge.get("data") or {}).get("transition_speech") or "" + data = edge.get("data") or {} + return str(data.get("transitionSpeech") or data.get("transition_speech") or "") - def find_edge(self, source: str | None, target: str | None) -> dict | None: - for edge in self.edges: - if edge.get("source") == source and edge.get("target") == target: - return edge - return None + def global_prompt(self) -> str: + return str(self.settings.get("globalPrompt") or "").strip() - def edge_description(self, edge: dict) -> str: - """作为转移函数的 description 交给 LLM:满足该条件时模型应调用此函数。""" - cond = self.edge_condition(edge) - target = self.name(edge.get("target")) - if cond: - return f"当满足以下条件时调用以转到节点「{target}」:{cond}" - return f"当当前节点任务完成、应继续推进对话时调用以转到节点「{target}」。" - - def is_end(self, nid: str | None) -> bool: - return self.node_type(nid) == "endCall" - - def has_graph(self) -> bool: - return self.start_id is not None - - def greeting(self) -> str: - return self.data(self.start_id).get("greeting") or "" - - def system_prompt_for(self, nid: str | None) -> str: - """组合当前节点提示与可选的全局提示(开始节点也是会话节点)。""" - header = f"[当前节点:{self.name(nid)}]" - 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}") + def prompt_for(self, node_id: str, store: DynamicVariableStore) -> str: + prompt = store.render(str(self.data(node_id).get("prompt") or "").strip()) + sections = [f"[当前阶段：{self.name(node_id)}]"] + if self.global_prompt(): + sections.append(f"[全局规则]\n{store.render(self.global_prompt())}") if prompt: - sections.append(f"[当前节点任务]\n{prompt}") + sections.append(f"[当前阶段任务]\n{prompt}") return "\n\n".join(sections) - # ---- 路由:决定下一节点 ---- - async def route( + def greeting(self, store: DynamicVariableStore) -> str: + return store.render(str(self.data(self.start_id).get("greeting") or "")) + + def expression_matches(self, expression: dict, values: dict[str, Any]) -> bool: + results = [] + for rule in expression.get("rules") or []: + name = str(rule.get("variable") or "") + operator = str(rule.get("operator") or "") + expected = rule.get("value") + exists = name in values + actual = values.get(name) + try: + if operator == "exists": + matched = exists if expected is not False else not exists + elif operator == "eq": + matched = actual == expected + elif operator == "neq": + matched = actual != expected + elif operator == "gt": + matched = actual > expected + elif operator == "gte": + matched = actual >= expected + elif operator == "lt": + matched = actual < expected + elif operator == "lte": + matched = actual <= expected + elif operator == "contains": + matched = expected in actual + elif operator == "in": + matched = actual in expected + else: + matched = False + except (TypeError, ValueError): + matched = False + results.append(matched) + if not results: + return False + return all(results) if expression.get("combinator", "and") == "and" else any(results) + + def deterministic_edge( self, - nid: str | None, - history: list[dict], + node_id: str, + store: DynamicVariableStore, *, - api_key: str, - base_url: str, - model: str, - ) -> str | None: - """根据对话历史判断当前节点是否应转移。返回目标节点 id,或 None 表示停留。""" - outs = self.outgoing(nid) - if not outs: - return None + include_default: bool, + ) -> dict | None: + default = None + for edge in self.outgoing(node_id): + data = edge.get("data") or {} + mode = data.get("mode") + if mode == "expression" and self.expression_matches( + data.get("expression") or {}, store.values + ): + return edge + if mode == "always": + default = edge + return default if include_default else None - options = [] - for i, edge in enumerate(outs, 1): - edata = edge.get("data") or {} - cond = edata.get("condition") or "(无明确条件,作为默认后继)" - tgt_name = self.name(edge.get("target")) - options.append(f"{i}. 条件:{cond} → 目标节点:{tgt_name}") - - convo = "\n".join( - f"{m['role']}: {m['content']}" for m in history[-8:] if m.get("content") - ) - system = ( - "你是语音对话的流程路由器。根据最近的对话,判断是否已满足某条转移条件。\n" - "规则:仅当某条件被明确满足时,返回其编号;若都不满足或不确定,返回 0 " - "(停留在当前节点继续对话)。只输出一个数字,不要任何解释。" - ) - user = ( - "可选转移:\n" - + "\n".join(options) - + f"\n\n对话记录:\n{convo}\n\n请只返回编号(0 表示停留):" - ) - - try: - from openai import AsyncOpenAI - - client = AsyncOpenAI(api_key=api_key, base_url=base_url or None) - resp = await client.chat.completions.create( - model=model, - messages=[ - {"role": "system", "content": system}, - {"role": "user", "content": user}, - ], - temperature=0, - max_tokens=5, - ) - text = (resp.choices[0].message.content or "").strip() - except Exception as exc: # noqa: BLE001 - 路由失败不应中断通话 - logger.warning(f"工作流路由调用失败,停留当前节点: {exc}") - return None - - match = re.search(r"\d+", text) - if not match: - return None - idx = int(match.group()) - if idx < 1 or idx > len(outs): - return None - target = outs[idx - 1].get("target") - logger.info(f"工作流路由: {self.name(nid)} → {self.name(target)} (edge {idx})") - return target + def llm_edges(self, node_id: str) -> list[dict]: + return [ + edge + for edge in self.outgoing(node_id) + if self.edge_mode(edge) in {"llm", "always"} + ] diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index f0990aa..3ac44fd 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -111,6 +111,19 @@ class BrainRegistryTests(unittest.TestCase): ) self.assertIn("user_name", assistant.dynamic_variable_definitions) + def test_workflow_keeps_dynamic_variables_and_tool_bindings(self): + assistant = AssistantUpsert( + name="workflow", + type="workflow", + toolIds=["tool_a"], + dynamicVariableDefinitions={ + "customer": {"type": "string", "required": False, "default": "王先生"} + }, + graph={}, + ) + self.assertEqual(assistant.tool_ids, ["tool_a"]) + self.assertIn("customer", assistant.dynamic_variable_definitions) + class DifyHelpersTests(unittest.TestCase): def test_normalize_api_base(self): @@ -363,7 +376,7 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): params = FakeFunctionParams( {"order_id": "A/1", "Authorization": "attacker-value"} ) - with patch("services.brains.prompt_brain.httpx.AsyncClient", FakeClient): + with patch("services.tool_executor.httpx.AsyncClient", FakeClient): await llm.functions["lookup_order"](params) self.assertEqual(requests[0][1], "https://example.test/orders/A%2F1") @@ -377,35 +390,91 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_transition_and_end_are_owned_by_workflow_brain(self): graph = { + "specVersion": 3, + "settings": {"globalPrompt": "全局规则"}, "nodes": [ { "id": "start", - "type": "startCall", - "data": {"name": "开始", "prompt": "收集需求"}, + "type": "start", + "data": {"name": "Start"}, + }, + { + "id": "agent", + "type": "agent", + "data": { + "name": "收集需求", + "prompt": "服务 {{user_name}}", + "contextPolicy": "fresh", + }, }, { "id": "end", - "type": "endCall", - "data": {"name": "结束", "prompt": "礼貌结束"}, + "type": "end", + "data": {"name": "End", "message": "感谢来电", "scope": "session"}, }, ], "edges": [ { - "id": "finish", + "id": "begin", "source": "start", + "target": "agent", + "data": {"mode": "always", "priority": 0}, + }, + { + "id": "finish", + "source": "agent", "target": "end", - "data": {"condition": "需求已收集"}, + "data": { + "mode": "llm", + "priority": 10, + "condition": "需求已收集", + }, } ], } - brain = WorkflowBrain(graph) + cfg = prepare_dynamic_config( + AssistantConfig( + type="workflow", + graph=graph, + dynamic_variable_definitions={ + "user_name": {"type": "string", "required": True} + }, + ), + {"user_name": "王先生"}, + assistant_id="asst_workflow", + ) + brain = WorkflowBrain(cfg) llm = FakeLLM() context = LLMContext(messages=[]) queued = [] - prompts = [] - visible_tools = [] call_end = FakeCallEnd() + class FakeWorker: + def __init__(self): + self.frames = [] + self.handlers = {} + + def set_reached_downstream_filter(self, *_args): + pass + + def event_handler(self, name): + def decorator(fn): + self.handlers[name] = fn + return fn + return decorator + + async def queue_frame(self, frame): + self.frames.append(frame) + + async def queue_frames(self, frames): + self.frames.extend(frames) + + worker = FakeWorker() + pair = SimpleNamespace( + user=lambda: SimpleNamespace(_context=context), + assistant=lambda: SimpleNamespace(has_function_calls_in_progress=False), + ) + async def queue_frame(frame): queued.append(frame) @@ -413,30 +482,34 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): context=context, llm=llm, queue_frame=queue_frame, - set_system_prompt=prompts.append, - set_tools=lambda tools: visible_tools.append(tools or []), + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, call_end=call_end, + worker=worker, + context_aggregator=pair, ) - 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, + await brain.setup(cfg, runtime) + await brain.on_connected() + self.assertEqual(brain._manager.current_node, "agent") + agent_config = brain._agent_config("agent") + self.assertIn("王先生", agent_config["role_message"]) + self.assertIn("完成当前阶段任务", agent_config["role_message"]) + self.assertEqual(agent_config["task_messages"], []) + self.assertEqual( + agent_config["context_strategy"].strategy.value, + "reset", ) + + edge_function = next( + function + for function in brain._agent_config("agent")["functions"] + if function.name == "goto_finish" + ) + _, terminal = await edge_function.handler({}, brain._manager) + self.assertEqual(terminal["name"], "end") self.assertTrue(call_end.ending) self.assertTrue(call_end.armed) + self.assertTrue(any(getattr(frame, "text", "") == "感谢来电" for frame in queued)) if __name__ == "__main__": diff --git a/backend/tests/test_pipeline_knowledge.py b/backend/tests/test_pipeline_knowledge.py index a71babf..9b8f328 100644 --- a/backend/tests/test_pipeline_knowledge.py +++ b/backend/tests/test_pipeline_knowledge.py @@ -1,7 +1,11 @@ import unittest from models import AssistantConfig -from services.pipecat.pipeline import _knowledge_tool_description +from services.pipecat.pipeline import ( + KNOWLEDGE_CONTEXT_MARKER, + KnowledgeRetrievalProcessor, + _knowledge_tool_description, +) class KnowledgeToolDescriptionTest(unittest.TestCase): @@ -33,6 +37,25 @@ class KnowledgeToolDescriptionTest(unittest.TestCase): self.assertNotIn("\n ", description) self.assertLess(len(description), 1000) + def test_workflow_knowledge_uses_system_role(self): + processor = KnowledgeRetrievalProcessor(None) + messages = [ + {"role": "assistant", "content": "你好"}, + { + "role": "developer", + "content": f"{KNOWLEDGE_CONTEXT_MARKER}\n旧检索结果", + }, + ] + + processor._set_context( + messages, + f"{KNOWLEDGE_CONTEXT_MARKER}\n新检索结果", + ) + + self.assertEqual(messages[0]["role"], "system") + self.assertIn("新检索结果", messages[0]["content"]) + self.assertFalse(any(message["role"] == "developer" for message in messages)) + if __name__ == "__main__": unittest.main() diff --git a/backend/tests/test_workflow_v3.py b/backend/tests/test_workflow_v3.py new file mode 100644 index 0000000..cb6dfac --- /dev/null +++ b/backend/tests/test_workflow_v3.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +import unittest + +from models import AssistantConfig, RuntimeModelResource +from services.pipecat.service_factory import config_with_resource +from services.node_specs import normalize_graph, validate_graph +from services.runtime_variables import DynamicVariableStore, prepare_dynamic_config +from services.workflow_engine import WorkflowEngine + + +def valid_graph(): + return { + "specVersion": 3, + "settings": {"globalPrompt": "服务 {{customer}}"}, + "nodes": [ + {"id": "start", "type": "start", "data": {"name": "Start"}}, + { + "id": "agent", + "type": "agent", + "data": {"name": "Agent", "prompt": "处理订单", "contextPolicy": "fresh"}, + }, + {"id": "end", "type": "end", "data": {"name": "End", "scope": "flow"}}, + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "agent", + "data": {"mode": "always", "priority": 0}, + }, + { + "id": "paid", + "source": "agent", + "target": "end", + "data": { + "mode": "expression", + "priority": 10, + "expression": { + "combinator": "and", + "rules": [ + {"variable": "order_status", "operator": "eq", "value": "paid"} + ], + }, + }, + }, + ], + } + + +class WorkflowGraphTests(unittest.TestCase): + def test_voice_resource_creates_isolated_runtime_config(self): + base = AssistantConfig(type="workflow", asr="default", voice="default") + asr = RuntimeModelResource( + id="asr_1", + capability="ASR", + interface_type="openai-asr", + values={"modelId": "sensevoice", "language": "zh", "apiUrl": "https://asr.test"}, + secrets={"apiKey": "secret"}, + ) + resolved = config_with_resource(base, asr) + self.assertEqual(resolved.asr, "sensevoice") + self.assertEqual(resolved.stt_api_key, "secret") + self.assertEqual(base.asr, "default") + + def test_v2_start_prompt_is_preserved_in_synthetic_agent(self): + graph = normalize_graph( + { + "nodes": [ + {"id": "s", "type": "startCall", "data": {"prompt": "询问需求"}}, + {"id": "e", "type": "endCall", "data": {"prompt": "再见"}}, + ], + "edges": [ + {"id": "done", "source": "s", "target": "e", "data": {"condition": "完成"}} + ], + } + ) + migrated = next(node for node in graph["nodes"] if node["type"] == "agent") + self.assertEqual(migrated["data"]["prompt"], "询问需求") + self.assertEqual(validate_graph(graph), []) + + def test_rejects_automatic_cycle_and_duplicate_priorities(self): + graph = valid_graph() + graph["nodes"].insert(1, {"id": "action", "type": "action", "data": {}}) + graph["edges"] = [ + {"id": "a", "source": "start", "target": "action", "data": {"mode": "always", "priority": 0}}, + {"id": "b", "source": "action", "target": "start", "data": {"mode": "always", "priority": 0}}, + {"id": "c", "source": "agent", "target": "end", "data": {"mode": "always", "priority": 10}}, + ] + errors = validate_graph(graph) + self.assertTrue(any("无等待循环" in error for error in errors)) + + def test_expression_routes_using_session_variables(self): + cfg = prepare_dynamic_config( + AssistantConfig( + type="workflow", + graph=valid_graph(), + dynamic_variable_definitions={ + "customer": {"type": "string", "default": "王先生"}, + "order_status": {"type": "string", "default": "pending"}, + }, + ), + {}, + assistant_id="asst_test", + ) + store = DynamicVariableStore.from_config(cfg) + engine = WorkflowEngine(cfg.graph) + self.assertIsNone(engine.deterministic_edge("agent", store, include_default=False)) + store.assign("order_status", "paid") + self.assertEqual( + engine.deterministic_edge("agent", store, include_default=False)["target"], + "end", + ) + self.assertIn("王先生", engine.prompt_for("agent", store)) + + +if __name__ == "__main__": + unittest.main() diff --git a/frontend/src/components/pages/AssistantPage.tsx b/frontend/src/components/pages/AssistantPage.tsx index 7545f5d..1e647ba 100644 --- a/frontend/src/components/pages/AssistantPage.tsx +++ b/frontend/src/components/pages/AssistantPage.tsx @@ -53,13 +53,6 @@ import { DialogHeader, DialogTitle, } from "@/components/ui/dialog"; -import { - Sheet, - SheetContent, - SheetDescription, - SheetHeader, - SheetTitle, -} from "@/components/ui/sheet"; import { DropdownMenu, DropdownMenuContent, @@ -478,10 +471,15 @@ export function AssistantPage(props: AssistantPageProps) { defaultGraph(), ); const [workflowSettings, setWorkflowSettings] = useState({ + globalPrompt: defaultGraph().settings.globalPrompt, allowInterrupt: true, turnConfig: defaultTurnConfig(), }); - const [debugOpen, setDebugOpen] = useState(false); + const [workflowDynamicVariableDefinitions, setWorkflowDynamicVariableDefinitions] = + useState>({}); + const [workflowDebugOpen, setWorkflowDebugOpen] = useState(false); + const [workflowEditingNodeId, setWorkflowEditingNodeId] = useState(null); + const [workflowEditingEdgeId, setWorkflowEditingEdgeId] = useState(null); const [activeNodeId, setActiveNodeId] = useState(null); const [dynamicVariablesOpen, setDynamicVariablesOpen] = useState(false); @@ -708,6 +706,7 @@ export function AssistantPage(props: AssistantPageProps) { name: workflowName, graph: workflowGraph, settings: workflowSettings, + dynamicVariableDefinitions: workflowDynamicVariableDefinitions, }), ); } @@ -849,19 +848,24 @@ export function AssistantPage(props: AssistantPageProps) { : defaultGraph(); const wfSettings: WorkflowSettings = { llm: assistant.modelResourceIds.LLM, - asr: assistant.modelResourceIds.ASR, - tts: assistant.modelResourceIds.TTS, + asr: graph.settings?.defaultAsrResourceId || assistant.modelResourceIds.ASR, + tts: graph.settings?.defaultTtsResourceId || assistant.modelResourceIds.TTS, + globalPrompt: graph.settings?.globalPrompt ?? "", allowInterrupt: assistant.enableInterrupt, turnConfig: assistant.turnConfig, }; setWorkflowName(assistant.name); setWorkflowGraph(graph); setWorkflowSettings(wfSettings); + setWorkflowDynamicVariableDefinitions( + assistant.dynamicVariableDefinitions ?? {}, + ); setSavedSnapshot( JSON.stringify({ name: assistant.name, graph, settings: wfSettings, + dynamicVariableDefinitions: assistant.dynamicVariableDefinitions ?? {}, }), ); } @@ -913,6 +917,7 @@ export function AssistantPage(props: AssistantPageProps) { ...(workflowSettings.tts ? { TTS: workflowSettings.tts } : {}), }, graph: workflowGraph as unknown as Record, + dynamicVariableDefinitions: workflowDynamicVariableDefinitions, }), ); } @@ -932,6 +937,7 @@ export function AssistantPage(props: AssistantPageProps) { name: workflowName, graph: workflowGraph, settings: workflowSettings, + dynamicVariableDefinitions: workflowDynamicVariableDefinitions, }) : null; const dirty = @@ -1377,7 +1383,11 @@ export function AssistantPage(props: AssistantPageProps) {
{saveError && ( - + {saveError} )} @@ -1385,7 +1395,11 @@ export function AssistantPage(props: AssistantPageProps) { variant="outline" className="gap-2 border-hairline-strong text-foreground hover:bg-surface-strong" disabled={!editingId} - onClick={() => setDebugOpen(true)} + onClick={() => { + setWorkflowEditingNodeId(null); + setWorkflowEditingEdgeId(null); + setWorkflowDebugOpen(true); + }} > 调试 @@ -1411,43 +1425,49 @@ export function AssistantPage(props: AssistantPageProps) { onChange={setWorkflowGraph} settings={workflowSettings} onSettingsChange={setWorkflowSettings} + onOpenDynamicVariables={() => setDynamicVariablesOpen(true)} + editingNodeId={workflowEditingNodeId} + onEditingNodeIdChange={setWorkflowEditingNodeId} + editingEdgeId={workflowEditingEdgeId} + onEditingEdgeIdChange={setWorkflowEditingEdgeId} + debugOpen={workflowDebugOpen} + onDebugOpenChange={(open) => { + setWorkflowDebugOpen(open); + if (!open) setActiveNodeId(null); + }} + debugPanel={ + { + setWorkflowDebugOpen(false); + setActiveNodeId(null); + }} + hasUnsavedChanges={dirty} + onNodeActive={setActiveNodeId} + dynamicVariablesEnabled + dynamicVariableDefinitions={workflowDynamicVariableDefinitions} + /> + } activeNodeId={activeNodeId} modelOptions={{ llm: credOptions("LLM"), asr: credOptions("ASR"), tts: credOptions("TTS"), }} + toolOptions={tools + .filter((tool) => tool.status === "active") + .map((tool) => ({ value: tool.id, label: tool.name }))} + knowledgeOptions={kbOptions} />
- { - setDebugOpen(open); - if (!open) setActiveNodeId(null); - }} - modal={false} - > - e.preventDefault()} - className="w-[440px] gap-0 border-l border-hairline bg-card p-0 sm:max-w-[440px]" - > - - 语音调试 - - 与当前助手进行语音对话调试,画布会高亮正在激活的节点。 - - - - - + ); } @@ -2132,7 +2152,8 @@ function SegmentedIconButton({ function DebugDrawer({ assistantId, - asSheet = false, + overlay = false, + onClose, hasUnsavedChanges = false, onNodeActive, vision = false, @@ -2140,7 +2161,8 @@ function DebugDrawer({ dynamicVariableDefinitions = {}, }: { assistantId: string | null; - asSheet?: boolean; + overlay?: boolean; + onClose?: () => void; hasUnsavedChanges?: boolean; onNodeActive?: (nodeId: string | null) => void; vision?: boolean; @@ -2179,14 +2201,25 @@ function DebugDrawer({ [camera, preview], ); - const containerClass = asSheet - ? "flex h-full min-w-0 flex-1 flex-col overflow-hidden" - : "hidden min-w-0 flex-1 flex-col overflow-hidden rounded-2xl border border-hairline bg-card shadow-sm lg:flex"; - return ( -