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..abc07c2 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,54 @@ 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 ( + ("defaultLlmResourceId", "LLM"), + ("defaultAsrResourceId", "ASR"), + ("defaultTtsResourceId", "TTS"), + ): + if settings.get(key): + resource_expectations[str(settings[key])] = capability + knowledge_ids: set[str] = ( + {str(settings["knowledgeBaseId"])} + if settings.get("knowledgeBaseId") + else set() + ) + for node in graph.get("nodes") or []: + data = node.get("data") or {} + if node.get("type") == "agent" and data.get("inheritGlobalConfig", True): + continue + if data.get("llmResourceId"): + resource_expectations[str(data["llmResourceId"])] = "LLM" + 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 +146,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 +207,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 +229,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 +302,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..bd738ff 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,15 @@ 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, 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: @@ -77,6 +86,15 @@ class BaseBrain: def record_user_message(self, content: str) -> None: """Observe a committed user message for brain-owned routing state.""" + async def on_user_turn_end(self, content: str) -> bool: + """Handle a complete user turn before the conversational LLM runs. + + Return True when the brain scheduled the next action itself and the + in-flight context frame must not reach the previous Agent's LLM. + """ + self.record_user_message(content) + return False + async def on_assistant_text_start(self, turn_id: str) -> None: """Observe the start of a generated assistant turn.""" @@ -107,6 +125,8 @@ class Brain(Protocol): def record_user_message(self, content: str) -> None: ... + async def on_user_turn_end(self, content: str) -> bool: ... + async def on_assistant_text_start(self, turn_id: str) -> None: ... async def on_assistant_text_end( 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..5676a7a 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -1,27 +1,43 @@ -"""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 ( + LLMRunFrame, + 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 pipecat.utils.time import time_now_iso8601 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 +from services.workflow_router import STAY_ON_CURRENT_AGENT, WorkflowLLMRouter -@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 +46,29 @@ 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._router = WorkflowLLMRouter(cfg or AssistantConfig(type="workflow")) + 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 +76,422 @@ 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._router = WorkflowLLMRouter(cfg) + 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 and self._engine.has_outgoing(self._engine.start_id): + raise RuntimeError("Start 初始化后没有命中的表达式边或默认边") + node_config = ( + await self._follow_edge(edge) + if edge + else self._passive_node_config(self._engine.start_id) + ) + 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}) + if content and not self._ended: + self._store.record("user", content) - async def on_assistant_text_start(self, turn_id: str) -> None: - if self._state.ended and self._state.end_turn_id is None: - self._state.end_turn_id = turn_id + async def on_user_turn_end(self, content: str) -> bool: + """Route a complete user turn before any Agent is allowed to reply.""" + if not content or self._ended: + return True + self.record_user_message(content) + manager = self._require_manager() + current = manager.current_node + if not current or self._engine.node_type(current) != "agent": + return True + + edge = self._engine.deterministic_edge( + current, + self._store, + include_default=False, + ) + outgoing = self._engine.outgoing(current) + llm_edges = [ + candidate + for candidate in outgoing + if self._engine.edge_mode(candidate) == "llm" + ] + default_edge = next( + ( + candidate + for candidate in outgoing + if self._engine.edge_mode(candidate) == "always" + ), + None, + ) + + if edge is None and llm_edges: + selected = await self._router_for_node(current).select_edge( + node_name=self._engine.name(current), + node_prompt=self._engine.prompt_for(current, self._store), + edges=llm_edges, + history=self._store.history, + variables={ + key: value + for key, value in self._store.values.items() + if not key.startswith("system__") + }, + edge_name=self._engine.edge_fn_name, + edge_description=self._engine.edge_description, + ) + if selected and selected != STAY_ON_CURRENT_AGENT: + edge = next( + ( + candidate + for candidate in llm_edges + if self._engine.edge_fn_name(candidate) == selected + ), + None, + ) + elif selected == STAY_ON_CURRENT_AGENT: + edge = default_edge + elif edge is None and not llm_edges: + edge = default_edge + + if edge and manager.current_node == current: + next_config = await self._follow_edge(edge) + await manager.set_node_from_config(next_config) + return True + + # The incoming LLMContextFrame is intentionally suppressed by the + # pipeline router. Queue prompt refresh + inference in this order so + # this user turn is answered with the current Agent's latest variables. + await self._refresh_agent_prompt(current) + await self._require_runtime().queue_frame(LLMRunFrame()) + return True 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) - 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}" + + def _router_for_node(self, node_id: str) -> WorkflowLLMRouter: + stage = self._engine.agent_stage_config(node_id) + resource_id = stage.llm_resource_id + cfg = self._cfg + resource = cfg.workflow_model_resources.get(resource_id) if cfg else None + if not cfg or not resource: + return self._router + from services.pipecat.service_factory import config_with_resource + + return WorkflowLLMRouter(config_with_resource(cfg, resource)) + + async def _apply_agent_stage(self, node_id: str) -> None: + stage = self._engine.agent_stage_config(node_id) + await self._emit_node_active(node_id) + if self._runtime and self._runtime.set_input_enabled: + self._runtime.set_input_enabled(True) + runtime = self._require_runtime() + if runtime.switch_services: + await runtime.switch_services( + stage.llm_resource_id or None, + stage.asr_resource_id or None, + stage.tts_resource_id or None, + ) + if runtime.set_knowledge_scope: + runtime.set_knowledge_scope( + { + "knowledge_base_id": stage.knowledge_base_id, + "mode": stage.knowledge_mode, + "top_n": stage.knowledge_top_n, + "score_threshold": stage.knowledge_score_threshold, + } + ) + + def _agent_config(self, node_id: str) -> NodeConfig: + data = self._engine.data(node_id) + entry_mode = str(data.get("entryMode") or "wait_user") + entry_speech = self._store.render(str(data.get("entrySpeech") or "")) + strategy = ( + ContextStrategy.RESET + if data.get("contextPolicy") == "fresh" + else ContextStrategy.APPEND + ) + stage = self._engine.agent_stage_config(node_id) + functions: list[FlowsFunctionSchema] = [] + for tool_id in stage.tool_ids: + 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) + config: NodeConfig = { + "name": node_id, + "role_message": self._agent_role_message(node_id), + "task_messages": ( + [{"role": "assistant", "content": entry_speech}] + if entry_mode == "fixed_speech" + else [] + ), + "functions": functions, + "context_strategy": ContextStrategyConfig(strategy=strategy), + "respond_immediately": entry_mode == "generate", + } + if entry_mode == "fixed_speech": + config["pre_actions"] = [ + { + "type": "workflow_fixed_speech", + "text": entry_speech, + "handler": self._play_fixed_speech, + } + ] + return config + + async def _play_fixed_speech(self, action: dict, _flow_manager: FlowManager) -> None: + """Play and persist Agent entry speech without creating an LLM turn.""" + await self._queue_visible_speech(str(action.get("text") or "")) + + async def _queue_visible_speech(self, text: str) -> None: + """Show and persist fixed workflow speech before sending it to TTS.""" + content = text.strip() + if not content: + return + self._store.record("agent", content) + runtime = self._require_runtime() + await runtime.queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "transcript", + "role": "assistant", + "content": content, + "timestamp": time_now_iso8601(), + } + ) + ) + await runtime.queue_frame(TTSSpeakFrame(content, append_to_context=False)) + + def _passive_node_config(self, node_id: str) -> NodeConfig: + """Keep a non-conversational terminal node active without ending the call.""" + 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, + cancel_on_interruption=True, + ) + + def _knowledge_function(self, node_id: str) -> FlowsFunctionSchema | None: + stage = self._engine.agent_stage_config(node_id) + knowledge_id = str(stage.knowledge_base_id or "") + if not knowledge_id or stage.knowledge_mode != "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=stage.knowledge_top_n, + score_threshold=stage.knowledge_score_threshold, + ) + 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, + cancel_on_interruption=True, + ) + + async def _follow_edge(self, edge: dict) -> NodeConfig: + speech = self._engine.edge_transition_speech(edge) + if speech: + await self._queue_visible_speech(self._store.render(speech)) + 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._passive_node_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}") + if not self._engine.has_outgoing(node_id): + return self._passive_node_config(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._queue_visible_speech(self._store.render(speech)) + 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._queue_visible_speech(message) + 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 self._queue_visible_speech(message) + return + runtime.call_end.begin("workflow_completed") + if message: + runtime.call_end.arm_after_speech() + await self._queue_visible_speech(message) + else: + await runtime.call_end.finish() async def _emit_node_active(self, node_id: str | None) -> None: if node_id: @@ -128,61 +501,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..02ebace 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -1,132 +1,110 @@ -"""工作流节点规格 + 图校验(对齐 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"} +AGENT_ENTRY_MODES = {"wait_user", "generate", "fixed_speech"} +AUTOMATIC_NODE_TYPES = {"start", "action", "handoff"} +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": 0, "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": 0}, "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": 0}, + "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 +113,372 @@ _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_agent_data(data: dict[str, Any]) -> None: + """Add v3 Agent defaults without changing existing node-level behavior.""" + data.setdefault("contextPolicy", "inherit") + data.setdefault("entryMode", "wait_user") + data.setdefault("entrySpeech", "") + if "inheritGlobalConfig" not in data: + has_node_overrides = any( + ( + data.get("llmResourceId"), + data.get("asrResourceId"), + data.get("ttsResourceId"), + data.get("knowledgeBaseId"), + data.get("toolIds"), + ) + ) + data["inheritGlobalConfig"] = not has_node_overrides - if not nodes: - return errors # 草稿:放行 - node_ids: set[str] = set() - type_counts: dict[str, int] = {} - node_type_by_id: dict[str, str] = {} +def _normalize_settings(settings: dict[str, Any], *, global_prompt: str = "") -> None: + settings.setdefault("globalPrompt", global_prompt) + settings.setdefault("defaultLlmResourceId", "") + settings.setdefault("defaultAsrResourceId", "") + settings.setdefault("defaultTtsResourceId", "") + settings.setdefault("toolIds", []) + settings.setdefault("knowledgeBaseId", "") + settings.setdefault("knowledgeMode", "automatic") + settings.setdefault("knowledgeTopN", 5) + settings.setdefault("knowledgeScoreThreshold", 0.0) + +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: + settings = source.setdefault("settings", {}) + _normalize_settings(settings) + source.setdefault("nodes", []) + source.setdefault("edges", []) + for node in source["nodes"]: + if node.get("type") != "agent": + continue + data = node.setdefault("data", {}) + _normalize_agent_data(data) + return source + + 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": + _normalize_agent_data(data) + 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", + "inheritGlobalConfig": True, + "entryMode": "wait_user", + "entrySpeech": "", + }, + } + ) + 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 {}) + _normalize_settings(settings, global_prompt=global_prompt) + 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 node_type == "agent": + data = node.get("data") or {} + entry_mode = data.get("entryMode", "wait_user") + if entry_mode not in AGENT_ENTRY_MODES: + errors.append(f"Agent 节点 {node_id} 的进入模式无效:{entry_mode}") + elif entry_mode == "fixed_speech" and not str( + data.get("entrySpeech") or "" + ).strip(): + errors.append(f"Agent 节点 {node_id} 的固定进入语不能为空") + + if counts["start"] != 1: + errors.append("工作流必须有且仅有一个 Start 节点") + 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": + expression_errors = _validate_expression(data.get("expression")) + errors.extend(f"边 {edge_id}:{item}" for item in expression_errors) + 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) + source_is_automatic = node_by_id[source_id].get("type") != "agent" + target_is_automatic = node_by_id[target_id].get("type") != "agent" + if source_is_automatic and target_is_automatic: + 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}") + node_type = node.get("type") + if ( + node_type in AUTOMATIC_NODE_TYPES + and outgoing[node_id] > 0 + and always_counts[node_id] != 1 + ): + errors.append(f"自动节点 {node_id} 存在出边时必须有且仅有一条默认边") + + start_id = next( + (node_id for node_id, node in node_by_id.items() if node.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 + + automatic_node_ids = ( + node_id + for node_id, node in node_by_id.items() + if node.get("type") != "agent" + ) + if any(visit(node_id) for node_id in automatic_node_ids): + 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("defaultLlmResourceId"), + settings.get("defaultAsrResourceId"), + settings.get("defaultTtsResourceId"), + ) + if value + } + tools: set[str] = {str(tool_id) for tool_id in settings.get("toolIds") or []} + knowledge: set[str] = ( + {str(settings["knowledgeBaseId"])} + if settings.get("knowledgeBaseId") + else set() + ) + for node in normalized.get("nodes") or []: + data = node.get("data") or {} + inherits_global = ( + node.get("type") == "agent" and data.get("inheritGlobalConfig", True) + ) + if not inherits_global: + for resource_id in ( + data.get("llmResourceId"), + 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("knowledgeBaseId"): + knowledge.add(str(data["knowledgeBaseId"])) + if data.get("toolId"): + tools.add(str(data["toolId"])) + 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..8896465 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -23,6 +23,8 @@ from services.pipecat.call_lifecycle import ( EndCallAfterSpeechProcessor, ) from services.pipecat.service_factory import ( + config_with_resource, + create_llm, create_realtime_service, create_stt, create_tts, @@ -32,6 +34,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 +43,7 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMContextFrame, LLMTextFrame, + ManuallySwitchServiceFrame, LLMMessagesAppendFrame, OutputTransportMessageUrgentFrame, TextFrame, @@ -48,6 +52,8 @@ from pipecat.frames.frames import ( UserImageRequestFrame, ) from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.llm_switcher import LLMSwitcher +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 +403,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,16 +485,53 @@ 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) +class UserTurnRoutingProcessor(FrameProcessor): + """Give a brain first right of refusal before a new user turn reaches the LLM.""" + + def __init__(self, brain: Brain): + super().__init__() + self._brain = brain + self._last_user_message: dict | None = None + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if direction != FrameDirection.DOWNSTREAM or not isinstance( + frame, LLMContextFrame + ): + await self.push_frame(frame, direction) + return + + user_message = next( + ( + message + for message in reversed(frame.context.get_messages()) + if message.get("role") == "user" + and isinstance(message.get("content"), str) + and str(message.get("content") or "").strip() + ), + None, + ) + if user_message is None: + await self.push_frame(frame, direction) + return + + if user_message is self._last_user_message: + # Programmatic LLMRunFrame after a node transition reuses the same + # user message. It is a response run, not another routing event. + await self.push_frame(frame, direction) + return + self._last_user_message = user_message + + content = str(user_message.get("content") or "").strip() + handled = await self._brain.on_user_turn_end(content) + if not handled: + await self.push_frame(frame, direction) + + class PassthroughLLMAssistantAggregator(LLMAssistantAggregator): """聚合 LLM 回复进上下文,同时继续把回复帧交给下游 TTS。""" @@ -457,7 +544,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 +591,72 @@ 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 + + +def _workflow_llm_switcher(cfg: AssistantConfig, base_service): + """Build an LLM switcher for the global model and Agent overrides.""" + settings = cfg.graph.get("settings") or {} + default_id = str(settings.get("defaultLlmResourceId") or "") + services_by_id = {} + for resource_id, resource in cfg.workflow_model_resources.items(): + if resource.capability != "LLM": + continue + services_by_id[resource_id] = ( + base_service + if resource_id == default_id + else create_llm(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 LLMSwitcher(llms=services), services_by_id, primary + + async def run_pipeline( transport, cfg: AssistantConfig, @@ -545,8 +697,38 @@ async def run_pipeline( ) return - stt = create_stt(cfg) - tts = create_tts(cfg) + graph_settings = cfg.graph.get("settings") or {} + default_llm_resource = cfg.workflow_model_resources.get( + str(graph_settings.get("defaultLlmResourceId") 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,17 +776,33 @@ 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) + llm = brain.build_llm( + config_with_resource(cfg, default_llm_resource) + if cfg.type == "workflow" and default_llm_resource + else cfg, + context, + ) + llm_services: dict[str, FrameProcessor] = {} + current_llm_service = llm + if cfg.type == "workflow": + llm, llm_services, current_llm_service = _workflow_llm_switcher(cfg, llm) user_aggregator = LLMUserAggregator( context, params=LLMUserAggregatorParams( 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, @@ -612,8 +810,11 @@ async def run_pipeline( ), ), ) + user_turn_router = UserTurnRoutingProcessor(brain) 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 +924,55 @@ 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, + cancel_on_interruption=True, + ) + ) + + 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,15 +993,16 @@ async def run_pipeline( transport.input(), vision_capture, text_input, - stt, + stt_processor, user_aggregator, + user_turn_router, knowledge_retrieval, llm, # Aggregate the streamed LLM text before TTS. On interruption, # 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 +1017,51 @@ async def run_pipeline( enable_rtvi=False, ) worker_holder["worker"] = worker + default_workflow_services = { + "llm": current_llm_service, + **current_voice_services, + } + + async def switch_workflow_services( + llm_resource_id: str | None, + asr_resource_id: str | None, + tts_resource_id: str | None, + ) -> None: + nonlocal current_llm_service + requested = ( + ("llm", llm_services, llm_resource_id), + ("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_workflow_services[kind] + ) + if target is None: + raise ValueError(f"Workflow {kind.upper()} 资源未加载:{resource_id}") + current = ( + current_llm_service + if kind == "llm" + else current_voice_services[kind] + ) + if current is target: + continue + await worker.queue_frame(ManuallySwitchServiceFrame(service=target)) + if kind == "llm": + current_llm_service = target + else: + 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 +1098,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, ), ) @@ -823,8 +1121,6 @@ async def run_pipeline( @user_aggregator.event_handler("on_user_turn_stopped") async def on_user_turn_stopped(_aggregator, _strategy, message): - if message.content: - brain.record_user_message(message.content) await queue_transcript("user", message.content, message.timestamp) @assistant_aggregator.event_handler("on_assistant_text_start") @@ -869,7 +1165,6 @@ async def run_pipeline( @text_input.event_handler("on_text_input") async def on_text_input(_processor, text): pending_text_inputs.append(text) - brain.record_user_message(text) # 前端显示不依赖 interruption 后续事件,必须在打断前先排入发送队列。 await queue_transcript("user", text, time_now_iso8601()) diff --git a/backend/services/pipecat/service_factory.py b/backend/services/pipecat/service_factory.py index ce2ff59..f575be7 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,42 @@ 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 model resource.""" + result = cfg.model_copy(deep=True) + values = resource.values or {} + secrets = resource.secrets or {} + if resource.capability == "LLM": + result.model = str(values.get("modelId") or "") + result.llm_interface_type = resource.interface_type + result.llm_values = values + result.llm_secrets = secrets + result.llm_api_key = str(secrets.get("apiKey") or "") + result.llm_base_url = str(values.get("apiUrl") or "") + elif 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..98078e9 100644 --- a/backend/services/workflow_engine.py +++ b/backend/services/workflow_engine.py @@ -1,170 +1,204 @@ -"""工作流图引擎(第一版)。 - -对应 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 dataclasses import dataclass from typing import Any -from loguru import logger +from services.node_specs import normalize_graph +from services.runtime_variables import DynamicVariableStore + + +@dataclass(frozen=True) +class AgentStageConfig: + """The complete assistant configuration active inside one Agent node.""" + + inherits_global: bool + llm_resource_id: str + asr_resource_id: str + tts_resource_id: str + tool_ids: tuple[str, ...] + knowledge_base_id: str | None + knowledge_mode: str + knowledge_top_n: int + knowledge_score_threshold: float 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 has_outgoing(self, node_id: str | None) -> bool: + return any(edge.get("source") == node_id for edge in self.edges) + + 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 "" - - 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 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 "" + data = edge.get("data") or {} + return str( + data.get("transitionSpeech") or data.get("transition_speech") or "" ) - sections = [header] - if global_prompt: - sections.append(f"[全局规则]\n{global_prompt}") + def global_prompt(self) -> str: + return str(self.settings.get("globalPrompt") or "").strip() + + def inherits_global_config(self, node_id: str) -> bool: + """Return the Agent's explicit configuration scope, defaulting to global.""" + return bool(self.data(node_id).get("inheritGlobalConfig", True)) + + def agent_stage_config(self, node_id: str) -> AgentStageConfig: + """Resolve either Workflow defaults or one Agent's complete override.""" + data = self.data(node_id) + inherits_global = self.inherits_global_config(node_id) + source = self.settings if inherits_global else data + llm_key = "defaultLlmResourceId" if inherits_global else "llmResourceId" + asr_key = "defaultAsrResourceId" if inherits_global else "asrResourceId" + tts_key = "defaultTtsResourceId" if inherits_global else "ttsResourceId" + knowledge_base_id = str(source.get("knowledgeBaseId") or "") + return AgentStageConfig( + inherits_global=inherits_global, + llm_resource_id=str(source.get(llm_key) or ""), + asr_resource_id=str(source.get(asr_key) or ""), + tts_resource_id=str(source.get(tts_key) or ""), + tool_ids=tuple(str(tool_id) for tool_id in source.get("toolIds") or []), + knowledge_base_id=knowledge_base_id or None, + knowledge_mode=( + str(source.get("knowledgeMode") or "automatic") + if knowledge_base_id + else "disabled" + ), + knowledge_top_n=int(source.get("knowledgeTopN") or 5), + knowledge_score_threshold=float( + source.get("knowledgeScoreThreshold") or 0.0 + ), + ) + + def prompt_for(self, node_id: str, store: DynamicVariableStore) -> str: + """Build the Agent system prompt according to its inheritance setting.""" + prompt = store.render(str(self.data(node_id).get("prompt") or "").strip()) + sections = [f"[当前阶段：{self.name(node_id)}]"] + if self.inherits_global_config(node_id) and 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/services/workflow_router.py b/backend/services/workflow_router.py new file mode 100644 index 0000000..8170a3f --- /dev/null +++ b/backend/services/workflow_router.py @@ -0,0 +1,129 @@ +"""Pre-response LLM routing for Workflow Agent edges. + +The router deliberately uses a separate, short completion. Its only output is +a required function choice, so the current Agent cannot speak before the graph +has decided whether the user turn belongs to another node. +""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from typing import Any + +from loguru import logger +from models import AssistantConfig +from openai import AsyncOpenAI + + +STAY_ON_CURRENT_AGENT = "workflow_stay_on_current_agent" +MAX_ROUTING_HISTORY_ENTRIES = 20 + + +class WorkflowLLMRouter: + """Select one LLM edge before the conversational LLM is allowed to reply.""" + + def __init__(self, cfg: AssistantConfig): + self._cfg = cfg + + async def select_edge( + self, + *, + node_name: str, + node_prompt: str, + edges: list[dict[str, Any]], + history: list[dict[str, str]], + variables: dict[str, Any], + edge_name: Callable[[dict[str, Any]], str], + edge_description: Callable[[dict[str, Any]], str], + ) -> str | None: + """Return an edge function name, STAY, or None when routing failed.""" + if not edges: + return STAY_ON_CURRENT_AGENT + + names = {edge_name(edge) for edge in edges} + stay_name = STAY_ON_CURRENT_AGENT + while stay_name in names: + stay_name = f"_{stay_name}" + + tools = [ + { + "type": "function", + "function": { + "name": edge_name(edge), + "description": edge_description(edge), + "parameters": {"type": "object", "properties": {}}, + }, + } + for edge in edges + ] + tools.append( + { + "type": "function", + "function": { + "name": stay_name, + "description": "所有转移条件都不满足，继续由当前 Agent 处理用户消息。", + "parameters": {"type": "object", "properties": {}}, + }, + } + ) + + ordered_conditions = "\n".join( + f"{index + 1}. {edge_description(edge)}" + for index, edge in enumerate(edges) + ) + router_prompt = ( + "你是工作流路由器，不是对话助手。收到一轮完整用户输入后，" + "必须且只能调用一个提供的函数，禁止输出任何口头回复。\n" + "按给出的顺序判断转移条件；选择第一个明确满足的转移函数。" + "如果没有条件满足，调用留在当前 Agent 的函数。\n\n" + f"当前节点：{node_name}\n" + f"当前节点任务：{node_prompt or '未配置'}\n" + f"转移条件：\n{ordered_conditions}" + ) + recent_history = history[-MAX_ROUTING_HISTORY_ENTRIES:] + routing_input = json.dumps( + { + "conversation": recent_history, + "session_variables": variables, + }, + ensure_ascii=False, + separators=(",", ":"), + ) + extra_body = self._cfg.llm_values.get("extraBody") + request_extra = ( + {"extra_body": extra_body} if isinstance(extra_body, dict) else {} + ) + client = AsyncOpenAI( + api_key=self._cfg.llm_api_key, + base_url=self._cfg.llm_base_url, + timeout=15.0, + ) + try: + response = await client.chat.completions.create( + model=self._cfg.model, + messages=[ + {"role": "system", "content": router_prompt}, + {"role": "user", "content": routing_input}, + ], + tools=tools, + tool_choice="required", + temperature=0, + **request_extra, + ) + tool_calls = response.choices[0].message.tool_calls or [] + if not tool_calls: + logger.warning("Workflow 路由 LLM 未返回函数调用，留在当前 Agent") + return STAY_ON_CURRENT_AGENT + selected = str(tool_calls[0].function.name or "") + if selected == stay_name: + return STAY_ON_CURRENT_AGENT + if selected not in names: + logger.warning(f"Workflow 路由 LLM 返回未知函数:{selected}") + return STAY_ON_CURRENT_AGENT + return selected + except Exception as exc: # noqa: BLE001 - routing failure must not end the call + logger.warning(f"Workflow LLM 边判断失败，留在当前 Agent:{exc}") + return None + finally: + await client.close() diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index f0990aa..b2e2e43 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -9,7 +9,10 @@ from pipecat.frames.frames import ( LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, + LLMRunFrame, LLMTextFrame, + OutputTransportMessageUrgentFrame, + TTSSpeakFrame, ) from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameDirection @@ -111,6 +114,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 +379,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") @@ -375,68 +391,315 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): - async def test_transition_and_end_are_owned_by_workflow_brain(self): - graph = { - "nodes": [ - { - "id": "start", - "type": "startCall", - "data": {"name": "开始", "prompt": "收集需求"}, - }, - { - "id": "end", - "type": "endCall", - "data": {"name": "结束", "prompt": "礼貌结束"}, - }, - ], - "edges": [ - { - "id": "finish", - "source": "start", - "target": "end", - "data": {"condition": "需求已收集"}, - } - ], - } - brain = WorkflowBrain(graph) - llm = FakeLLM() - context = LLMContext(messages=[]) + async def test_nodes_without_outgoing_edges_remain_active(self): queued = [] - prompts = [] - visible_tools = [] - call_end = FakeCallEnd() async def queue_frame(frame): queued.append(frame) + runtime = BrainRuntime( + context=LLMContext(messages=[]), + llm=FakeLLM(), + queue_frame=queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + ) + + class FakeManager: + def __init__(self, current_node=None): + self.current_node = current_node + + async def initialize(self, config): + self.current_node = config["name"] + + start_brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {}, + "nodes": [{"id": "start", "type": "start", "data": {}}], + "edges": [], + } + ) + start_brain._runtime = runtime + start_brain._manager = FakeManager() + await start_brain.on_connected() + self.assertEqual(start_brain._manager.current_node, "start") + + agent_brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {"globalPrompt": "全局规则"}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "agent", + "type": "agent", + "data": {"prompt": "持续回答"}, + }, + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "agent", + "data": {"mode": "always", "priority": 0}, + } + ], + } + ) + agent_brain._runtime = runtime + agent_brain._manager = FakeManager("agent") + queued.clear() + handled = await agent_brain.on_user_turn_end("请继续回答") + self.assertTrue(handled) + self.assertEqual(agent_brain._manager.current_node, "agent") + self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in queued)) + + handoff_brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "handoff", + "type": "handoff", + "data": {"targetType": "human"}, + }, + ], + "edges": [], + } + ) + handoff_brain._runtime = runtime + handoff_config = await handoff_brain._resolve_path("handoff") + self.assertEqual(handoff_config["name"], "handoff") + self.assertTrue( + any( + isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("type") == "handoff-requested" + for frame in queued + ) + ) + + async def test_transition_and_end_are_owned_by_workflow_brain(self): + graph = { + "specVersion": 3, + "settings": { + "globalPrompt": "全局规则", + "defaultLlmResourceId": "llm_global", + "defaultAsrResourceId": "asr_global", + "defaultTtsResourceId": "tts_global", + "knowledgeBaseId": "kb_global", + "knowledgeMode": "automatic", + }, + "nodes": [ + { + "id": "start", + "type": "start", + "data": {"name": "Start"}, + }, + { + "id": "agent", + "type": "agent", + "data": { + "name": "收集需求", + "prompt": "服务 {{user_name}}", + "contextPolicy": "fresh", + }, + }, + { + "id": "end", + "type": "end", + "data": {"name": "End", "message": "感谢来电", "scope": "session"}, + }, + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "agent", + "data": {"mode": "always", "priority": 0}, + }, + { + "id": "finish", + "source": "agent", + "target": "end", + "data": { + "mode": "llm", + "priority": 10, + "condition": "需求已收集", + "transitionSpeech": "正在为你结束流程", + }, + } + ], + } + 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 = [] + service_switches = [] + knowledge_scopes = [] + 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) + + async def switch_services(llm_id, asr_id, tts_id): + service_switches.append((llm_id, asr_id, tts_id)) + runtime = BrainRuntime( 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, + switch_services=switch_services, + set_knowledge_scope=knowledge_scopes.append, ) - 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") + self.assertEqual( + service_switches, + [("llm_global", "asr_global", "tts_global")], ) + self.assertEqual(knowledge_scopes[-1]["knowledge_base_id"], "kb_global") + + brain._engine.data("agent").update( + { + "inheritGlobalConfig": False, + "llmResourceId": "llm_agent", + "asrResourceId": "asr_agent", + "ttsResourceId": "tts_agent", + "knowledgeBaseId": "kb_agent", + "knowledgeMode": "on_demand", + } + ) + await brain._apply_agent_stage("agent") + self.assertEqual( + service_switches[-1], + ("llm_agent", "asr_agent", "tts_agent"), + ) + self.assertEqual(knowledge_scopes[-1]["knowledge_base_id"], "kb_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.assertFalse(agent_config["respond_immediately"]) + self.assertFalse(any(isinstance(frame, LLMRunFrame) for frame in worker.frames)) + self.assertEqual( + agent_config["context_strategy"].strategy.value, + "reset", + ) + + brain._engine.data("agent")["entryMode"] = "generate" + generate_config = brain._agent_config("agent") + self.assertTrue(generate_config["respond_immediately"]) + worker.frames.clear() + await brain._manager.set_node_from_config(generate_config) + self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in worker.frames)) + + brain._engine.data("agent").update( + {"entryMode": "fixed_speech", "entrySpeech": "您好，{{user_name}}"} + ) + fixed_config = brain._agent_config("agent") + self.assertFalse(fixed_config["respond_immediately"]) + self.assertEqual( + fixed_config["pre_actions"][0]["type"], + "workflow_fixed_speech", + ) + self.assertEqual(fixed_config["pre_actions"][0]["text"], "您好，王先生") + self.assertEqual( + fixed_config["task_messages"], + [{"role": "assistant", "content": "您好，王先生"}], + ) + worker.frames.clear() + queued.clear() + await brain._manager.set_node_from_config(fixed_config) + self.assertTrue(any(isinstance(frame, TTSSpeakFrame) for frame in queued)) + self.assertFalse(any(isinstance(frame, LLMRunFrame) for frame in worker.frames)) + + self.assertFalse( + any( + function.name == "goto_finish" + for function in brain._agent_config("agent")["functions"] + ) + ) + await brain.on_assistant_text_end("old-turn", "需求已收集", False) + self.assertEqual(brain._manager.current_node, "agent") + + class FakeRouter: + async def select_edge(self, **_kwargs): + return "goto_finish" + + brain._router = FakeRouter() + handled = await brain.on_user_turn_end("我的需求已经说完了") + self.assertTrue(handled) + self.assertEqual(brain._manager.current_node, "end") + self.assertIn("我的需求已经说完了", brain._store.values["system__conversation_history"]) self.assertTrue(call_end.ending) self.assertTrue(call_end.armed) + self.assertTrue(any(getattr(frame, "text", "") == "感谢来电" for frame in queued)) + assistant_transcripts = [ + frame.message.get("content") + for frame in queued + if isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("type") == "transcript" + and frame.message.get("role") == "assistant" + ] + self.assertEqual( + assistant_transcripts, + ["您好，王先生", "正在为你结束流程", "感谢来电"], + ) + self.assertIn( + "正在为你结束流程", + brain._store.values["system__conversation_history"], + ) + self.assertIn( + "感谢来电", + brain._store.values["system__conversation_history"], + ) if __name__ == "__main__": diff --git a/backend/tests/test_pipeline_knowledge.py b/backend/tests/test_pipeline_knowledge.py index a71babf..0e80113 100644 --- a/backend/tests/test_pipeline_knowledge.py +++ b/backend/tests/test_pipeline_knowledge.py @@ -1,7 +1,15 @@ import unittest from models import AssistantConfig -from services.pipecat.pipeline import _knowledge_tool_description +from pipecat.frames.frames import LLMContextFrame +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.frame_processor import FrameDirection +from services.pipecat.pipeline import ( + KNOWLEDGE_CONTEXT_MARKER, + KnowledgeRetrievalProcessor, + UserTurnRoutingProcessor, + _knowledge_tool_description, +) class KnowledgeToolDescriptionTest(unittest.TestCase): @@ -33,6 +41,57 @@ 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)) + + +class UserTurnRoutingProcessorTest(unittest.IsolatedAsyncioTestCase): + async def test_routes_each_user_message_once_before_response_run(self): + class FakeBrain: + def __init__(self): + self.turns = [] + + async def on_user_turn_end(self, content): + self.turns.append(content) + return True + + brain = FakeBrain() + processor = UserTurnRoutingProcessor(brain) + forwarded = [] + + async def push_frame(frame, direction): + forwarded.append((frame, direction)) + + processor.push_frame = push_frame + context = LLMContext(messages=[{"role": "user", "content": "我叫李白"}]) + frame = LLMContextFrame(context) + + await processor.process_frame(frame, FrameDirection.DOWNSTREAM) + self.assertEqual(brain.turns, ["我叫李白"]) + self.assertEqual(forwarded, []) + + # A queued LLMRunFrame after the transition uses the same context. It + # must reach the target Agent without invoking routing a second time. + await processor.process_frame(frame, FrameDirection.DOWNSTREAM) + self.assertEqual(brain.turns, ["我叫李白"]) + self.assertEqual(forwarded, [(frame, FrameDirection.DOWNSTREAM)]) + if __name__ == "__main__": unittest.main() diff --git a/backend/tests/test_workflow_router.py b/backend/tests/test_workflow_router.py new file mode 100644 index 0000000..67d7766 --- /dev/null +++ b/backend/tests/test_workflow_router.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from models import AssistantConfig +from services.workflow_router import WorkflowLLMRouter + + +class WorkflowLLMRouterTest(unittest.IsolatedAsyncioTestCase): + async def test_uses_required_tool_choice_without_developer_messages(self): + requests = [] + + class FakeCompletions: + async def create(self, **kwargs): + requests.append(kwargs) + return SimpleNamespace( + choices=[ + SimpleNamespace( + message=SimpleNamespace( + tool_calls=[ + SimpleNamespace( + function=SimpleNamespace(name="goto_age", arguments="{}") + ) + ] + ) + ) + ] + ) + + class FakeClient: + def __init__(self, **_kwargs): + self.chat = SimpleNamespace(completions=FakeCompletions()) + self.closed = False + + async def close(self): + self.closed = True + + cfg = AssistantConfig( + type="workflow", + model="deepseek-chat", + llm_api_key="secret", + llm_base_url="https://llm.test/v1", + ) + router = WorkflowLLMRouter(cfg) + edges = [ + { + "id": "age", + "data": {"condition": "用户已经回答姓名", "priority": 10}, + } + ] + + with patch("services.workflow_router.AsyncOpenAI", FakeClient): + selected = await router.select_edge( + node_name="询问姓名", + node_prompt="询问用户姓名", + edges=edges, + history=[{"role": "user", "message": "我叫李白"}], + variables={"customer_type": "new"}, + edge_name=lambda _edge: "goto_age", + edge_description=lambda _edge: "用户已经回答姓名", + ) + + self.assertEqual(selected, "goto_age") + self.assertEqual(requests[0]["tool_choice"], "required") + self.assertEqual( + [message["role"] for message in requests[0]["messages"]], + ["system", "user"], + ) + self.assertNotIn("developer", str(requests[0]["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..1cfa553 --- /dev/null +++ b/backend/tests/test_workflow_v3.py @@ -0,0 +1,328 @@ +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 graph_references, 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_agent_entry_mode_defaults_and_validation(self): + graph = valid_graph() + normalized = normalize_graph(graph) + agent = next(node for node in normalized["nodes"] if node["type"] == "agent") + self.assertEqual(agent["data"]["entryMode"], "wait_user") + self.assertEqual(agent["data"]["entrySpeech"], "") + self.assertTrue(agent["data"]["inheritGlobalConfig"]) + self.assertEqual(agent["data"]["contextPolicy"], "fresh") + + agent["data"]["entryMode"] = "fixed_speech" + self.assertTrue( + any("固定进入语不能为空" in error for error in validate_graph(normalized)) + ) + agent["data"]["entrySpeech"] = "您好，{{customer}}" + self.assertEqual(validate_graph(normalized), []) + + 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") + + llm = RuntimeModelResource( + id="llm_1", + capability="LLM", + interface_type="openai-llm", + values={"modelId": "deepseek-chat", "apiUrl": "https://llm.test/v1"}, + secrets={"apiKey": "llm-secret"}, + ) + llm_resolved = config_with_resource(base, llm) + self.assertEqual(llm_resolved.model, "deepseek-chat") + self.assertEqual(llm_resolved.llm_api_key, "llm-secret") + + def test_global_and_custom_agent_references_are_preserved(self): + graph = valid_graph() + graph["settings"].update( + { + "defaultLlmResourceId": "llm_global", + "defaultAsrResourceId": "asr_global", + "defaultTtsResourceId": "tts_global", + "toolIds": ["tool_global"], + "knowledgeBaseId": "kb_global", + } + ) + agent = next(node for node in graph["nodes"] if node["type"] == "agent") + agent["data"].update( + { + "inheritGlobalConfig": False, + "llmResourceId": "llm_agent", + "asrResourceId": "asr_agent", + "ttsResourceId": "tts_agent", + "toolIds": ["tool_agent"], + "knowledgeBaseId": "kb_agent", + } + ) + + refs = graph_references(graph) + self.assertEqual( + refs["model_resources"], + { + "llm_global", + "asr_global", + "tts_global", + "llm_agent", + "asr_agent", + "tts_agent", + }, + ) + self.assertEqual(refs["tools"], {"tool_global", "tool_agent"}) + self.assertEqual(refs["knowledge_bases"], {"kb_global", "kb_agent"}) + + def test_existing_agent_override_disables_implicit_inheritance(self): + graph = valid_graph() + agent = next(node for node in graph["nodes"] if node["type"] == "agent") + agent["data"]["toolIds"] = ["legacy_tool"] + normalized = normalize_graph(graph) + normalized_agent = next( + node for node in normalized["nodes"] if node["type"] == "agent" + ) + self.assertFalse(normalized_agent["data"]["inheritGlobalConfig"]) + + def test_inherited_agent_ignores_stale_custom_references(self): + graph = valid_graph() + agent = next(node for node in graph["nodes"] if node["type"] == "agent") + agent["data"].update( + { + "inheritGlobalConfig": True, + "llmResourceId": "stale_llm", + "asrResourceId": "stale_asr", + "ttsResourceId": "stale_tts", + "toolIds": ["stale_tool"], + "knowledgeBaseId": "stale_kb", + } + ) + + refs = graph_references(graph) + + self.assertNotIn("stale_llm", refs["model_resources"]) + self.assertNotIn("stale_tool", refs["tools"]) + self.assertNotIn("stale_kb", refs["knowledge_bases"]) + + def test_agent_effective_config_inherits_then_switches_to_override(self): + graph = valid_graph() + graph["settings"].update( + { + "defaultLlmResourceId": "llm_global", + "defaultAsrResourceId": "asr_global", + "defaultTtsResourceId": "tts_global", + "toolIds": ["tool_global"], + "knowledgeBaseId": "kb_global", + "knowledgeMode": "on_demand", + "knowledgeTopN": 8, + "knowledgeScoreThreshold": 0.4, + } + ) + engine = WorkflowEngine(graph) + inherited = engine.agent_stage_config("agent") + self.assertEqual(inherited.llm_resource_id, "llm_global") + self.assertEqual(inherited.tool_ids, ("tool_global",)) + self.assertEqual(inherited.knowledge_mode, "on_demand") + + engine.data("agent").update( + { + "inheritGlobalConfig": False, + "llmResourceId": "llm_agent", + "toolIds": ["tool_agent"], + "knowledgeBaseId": "", + } + ) + custom = engine.agent_stage_config("agent") + self.assertEqual(custom.llm_resource_id, "llm_agent") + self.assertEqual(custom.tool_ids, ("tool_agent",)) + self.assertEqual(custom.knowledge_mode, "disabled") + + def test_start_agent_and_handoff_may_have_no_outgoing_edge(self): + terminal_graphs = [ + { + "specVersion": 3, + "settings": {}, + "nodes": [{"id": "start", "type": "start", "data": {}}], + "edges": [], + }, + { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "agent", + "type": "agent", + "data": {"prompt": "持续处理用户问题"}, + }, + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "agent", + "data": {"mode": "always", "priority": 0}, + } + ], + }, + { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "handoff", "type": "handoff", "data": {}}, + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "handoff", + "data": {"mode": "always", "priority": 0}, + } + ], + }, + ] + + for graph in terminal_graphs: + with self.subTest(node=graph["nodes"][-1]["type"]): + self.assertEqual(validate_graph(graph), []) + + action_without_exit = { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "action", "type": "action", "data": {}}, + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "action", + "data": {"mode": "always", "priority": 0}, + } + ], + } + self.assertTrue( + any( + "action 的出边不能少于 1" in error + for error in validate_graph(action_without_exit) + ) + ) + + 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)) + + inherited_prompt = engine.prompt_for("agent", store) + self.assertIn("服务 王先生", inherited_prompt) + self.assertIn("处理订单", inherited_prompt) + + engine.data("agent")["inheritGlobalConfig"] = False + custom_prompt = engine.prompt_for("agent", store) + self.assertNotIn("服务 王先生", custom_prompt) + self.assertIn("处理订单", custom_prompt) + + +if __name__ == "__main__": + unittest.main() diff --git a/frontend/src/components/editor/knowledge-retrieval-config-dialog.tsx b/frontend/src/components/editor/knowledge-retrieval-config-dialog.tsx new file mode 100644 index 0000000..b41412f --- /dev/null +++ b/frontend/src/components/editor/knowledge-retrieval-config-dialog.tsx @@ -0,0 +1,168 @@ +"use client"; + +import { Settings2 } from "lucide-react"; +import { useState } from "react"; + +import { Button } from "@/components/ui/button"; +import { + Dialog, + DialogContent, + DialogDescription, + DialogFooter, + DialogHeader, + DialogTitle, +} from "@/components/ui/dialog"; +import { Input } from "@/components/ui/input"; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from "@/components/ui/select"; +import type { KnowledgeRetrievalConfig } from "@/lib/api"; + +export const DEFAULT_KNOWLEDGE_RETRIEVAL_CONFIG: KnowledgeRetrievalConfig = { + mode: "automatic", + topN: 5, + scoreThreshold: 0, +}; + +export function KnowledgeRetrievalConfigDialog({ + disabled, + value, + onChange, +}: { + disabled: boolean; + value: KnowledgeRetrievalConfig; + onChange: (config: KnowledgeRetrievalConfig) => void; +}) { + const [open, setOpen] = useState(false); + const [draft, setDraft] = useState(value); + const [error, setError] = useState(null); + + function openDialog() { + setDraft(value); + setError(null); + setOpen(true); + } + + function saveDraft() { + if (draft.topN === 0 || draft.topN < -1 || !Number.isInteger(draft.topN)) { + setError("Top N 必须为 -1 或大于 0 的整数"); + return; + } + if (draft.scoreThreshold < 0 || draft.scoreThreshold > 1) { + setError("最低相关度必须在 0 到 1 之间"); + return; + } + onChange(draft); + setOpen(false); + } + + return ( + <> + + + + + + 知识库高级配置 + + 设置检索触发方式、返回数量和相关度过滤条件。 + + + +
+
+
检索方式
+ +

+ {draft.mode === "automatic" + ? "每轮用户提问后自动检索，响应行为更稳定。" + : "由大模型判断是否调用知识库，依赖模型的工具调用能力。"} +

+
+ + + + + + {error &&

{error}

} +
+ + + + + + + + + ); +} diff --git a/frontend/src/components/editor/section-card.tsx b/frontend/src/components/editor/section-card.tsx new file mode 100644 index 0000000..8f0954e --- /dev/null +++ b/frontend/src/components/editor/section-card.tsx @@ -0,0 +1,92 @@ +"use client"; + +/** + * Compact section chrome shared by assistant editors and workflow node panels. + * Density matches the debug preview drawer (text-sm titles, tight padding). + */ + +import { HelpCircle } from "lucide-react"; +import type { ReactNode } from "react"; + +import { + Card, + CardContent, + CardHeader, + CardTitle, +} from "@/components/ui/card"; +import { + Popover, + PopoverContent, + PopoverTrigger, +} from "@/components/ui/popover"; +import { cn } from "@/lib/utils"; + +export function SectionCard({ + icon, + title, + description, + children, + className, +}: { + icon?: ReactNode; + title?: string; + description?: string; + children: ReactNode; + className?: string; +}) { + const hasHeader = Boolean(title); + + return ( + + {hasHeader && ( + +
+ {icon && ( +
+ {icon} +
+ )} +
+ + {title} + + {description && } +
+
+ + )} + + {children} + + + ); +} + +export function HelpHint({ text }: { text: string }) { + return ( + + + + + + {text} + + + ); +} diff --git a/frontend/src/components/pages/AssistantPage.tsx b/frontend/src/components/pages/AssistantPage.tsx index 7545f5d..a171382 100644 --- a/frontend/src/components/pages/AssistantPage.tsx +++ b/frontend/src/components/pages/AssistantPage.tsx @@ -21,7 +21,6 @@ import { Save, Mic, Send, - HelpCircle, Waypoints, AudioLines, Terminal, @@ -53,13 +52,6 @@ import { DialogHeader, DialogTitle, } from "@/components/ui/dialog"; -import { - Sheet, - SheetContent, - SheetDescription, - SheetHeader, - SheetTitle, -} from "@/components/ui/sheet"; import { DropdownMenu, DropdownMenuContent, @@ -94,12 +86,6 @@ import { PageHeader } from "@/components/ui/page-header"; import { FilterPills } from "@/components/ui/filter-pills"; import { SearchInput } from "@/components/ui/search-input"; import { ListToolbar } from "@/components/ui/list-toolbar"; -import { - Card, - CardContent, - CardHeader, - CardTitle, -} from "@/components/ui/card"; import { useCallback, useEffect, useRef, useState } from "react"; import { useRouter } from "next/navigation"; import { @@ -131,6 +117,7 @@ import { WorkflowEditor, type WorkflowSettings, } from "@/components/workflow/WorkflowEditor"; +import { HelpHint, SectionCard } from "@/components/editor/section-card"; import { defaultGraph, type WorkflowGraph, @@ -369,7 +356,7 @@ type AssistantTypeOption = { label: string; description: string; icon: React.ReactNode; - /** 提示词、Dify、FastGPT 类型已落地，工作流暂时显示占位页 */ + /** 提示词、工作流、Dify、FastGPT 已落地；OpenCode 暂时显示即将上线 */ available: boolean; }; @@ -386,7 +373,7 @@ const assistantTypeOptions: AssistantTypeOption[] = [ label: "使用工作流构建", description: "用可视化编排串联多个节点，适合多步骤、带分支的复杂流程。", icon: , - available: false, + available: true, }, { type: "Dify", @@ -407,7 +394,7 @@ const assistantTypeOptions: AssistantTypeOption[] = [ label: "使用 OpenCode 构建", description: "对接 OpenCode 服务，通过提示词驱动代码助手并支持实时语音对话。", icon: , - available: true, + available: false, }, ]; @@ -478,10 +465,26 @@ export function AssistantPage(props: AssistantPageProps) { defaultGraph(), ); const [workflowSettings, setWorkflowSettings] = useState({ + globalPrompt: defaultGraph().settings.globalPrompt, + llm: defaultGraph().settings.defaultLlmResourceId, + asr: defaultGraph().settings.defaultAsrResourceId, + tts: defaultGraph().settings.defaultTtsResourceId, + toolIds: defaultGraph().settings.toolIds, + knowledgeBaseId: defaultGraph().settings.knowledgeBaseId, + knowledgeRetrievalConfig: { + mode: defaultGraph().settings.knowledgeMode, + topN: defaultGraph().settings.knowledgeTopN, + scoreThreshold: defaultGraph().settings.knowledgeScoreThreshold, + }, allowInterrupt: true, turnConfig: defaultTurnConfig(), }); - const [debugOpen, setDebugOpen] = useState(false); + const [workflowDynamicVariableDefinitions, setWorkflowDynamicVariableDefinitions] = + useState>({}); + const [workflowDebugOpen, setWorkflowDebugOpen] = useState(false); + const [workflowSettingsOpen, setWorkflowSettingsOpen] = useState(false); + const [workflowEditingNodeId, setWorkflowEditingNodeId] = useState(null); + const [workflowEditingEdgeId, setWorkflowEditingEdgeId] = useState(null); const [activeNodeId, setActiveNodeId] = useState(null); const [dynamicVariablesOpen, setDynamicVariablesOpen] = useState(false); @@ -708,6 +711,7 @@ export function AssistantPage(props: AssistantPageProps) { name: workflowName, graph: workflowGraph, settings: workflowSettings, + dynamicVariableDefinitions: workflowDynamicVariableDefinitions, }), ); } @@ -848,20 +852,41 @@ export function AssistantPage(props: AssistantPageProps) { ? (assistant.graph as WorkflowGraph) : defaultGraph(); const wfSettings: WorkflowSettings = { - llm: assistant.modelResourceIds.LLM, - asr: assistant.modelResourceIds.ASR, - tts: assistant.modelResourceIds.TTS, + llm: + graph.settings?.defaultLlmResourceId || + assistant.modelResourceIds.LLM, + asr: graph.settings?.defaultAsrResourceId || assistant.modelResourceIds.ASR, + tts: graph.settings?.defaultTtsResourceId || assistant.modelResourceIds.TTS, + toolIds: graph.settings?.toolIds ?? [], + knowledgeBaseId: + graph.settings?.knowledgeBaseId || assistant.knowledgeBaseId || "", + knowledgeRetrievalConfig: { + mode: + graph.settings?.knowledgeMode || + assistant.knowledgeRetrievalConfig.mode, + topN: + graph.settings?.knowledgeTopN ?? + assistant.knowledgeRetrievalConfig.topN, + scoreThreshold: + graph.settings?.knowledgeScoreThreshold ?? + assistant.knowledgeRetrievalConfig.scoreThreshold, + }, + 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 ?? {}, }), ); } @@ -912,7 +937,11 @@ export function AssistantPage(props: AssistantPageProps) { ...(workflowSettings.asr ? { ASR: workflowSettings.asr } : {}), ...(workflowSettings.tts ? { TTS: workflowSettings.tts } : {}), }, + knowledgeBaseId: workflowSettings.knowledgeBaseId || null, + knowledgeRetrievalConfig: workflowSettings.knowledgeRetrievalConfig, + toolIds: workflowSettings.toolIds, graph: workflowGraph as unknown as Record, + dynamicVariableDefinitions: workflowDynamicVariableDefinitions, }), ); } @@ -932,6 +961,7 @@ export function AssistantPage(props: AssistantPageProps) { name: workflowName, graph: workflowGraph, settings: workflowSettings, + dynamicVariableDefinitions: workflowDynamicVariableDefinitions, }) : null; const dirty = @@ -1377,7 +1407,11 @@ export function AssistantPage(props: AssistantPageProps) {
{saveError && ( - + {saveError} )} @@ -1385,7 +1419,12 @@ 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={() => { + setWorkflowSettingsOpen(false); + setWorkflowEditingNodeId(null); + setWorkflowEditingEdgeId(null); + setWorkflowDebugOpen(true); + }} > 调试 @@ -1411,43 +1450,51 @@ export function AssistantPage(props: AssistantPageProps) { onChange={setWorkflowGraph} settings={workflowSettings} onSettingsChange={setWorkflowSettings} + onOpenDynamicVariables={() => setDynamicVariablesOpen(true)} + editingNodeId={workflowEditingNodeId} + onEditingNodeIdChange={setWorkflowEditingNodeId} + editingEdgeId={workflowEditingEdgeId} + onEditingEdgeIdChange={setWorkflowEditingEdgeId} + settingsOpen={workflowSettingsOpen} + onSettingsOpenChange={setWorkflowSettingsOpen} + 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]" - > - - 语音调试 - - 与当前助手进行语音对话调试,画布会高亮正在激活的节点。 - - - - - + ); } @@ -1488,10 +1535,10 @@ export function AssistantPage(props: AssistantPageProps) { -
-
+
+
} + icon={} title="Dify 应用配置" description="从「模型资源」中选择 Dify 应用。开场白、知识库、提示词等对话编排请在 Dify 平台配置，本页不重复设置。" > @@ -1505,7 +1552,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="语音配置" description="从「模型资源」中选择语音识别与语音合成。大模型、知识库与开场白由 Dify 应用提供，请前往 Dify 平台配置。" > @@ -1526,7 +1573,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="交互策略" description="设置实时视频对话时的交互体验" > @@ -1584,10 +1631,10 @@ export function AssistantPage(props: AssistantPageProps) {
-
-
+
+
} + icon={} title="FastGPT 应用配置" description="从「模型资源」中选择 FastGPT 应用。开场白、知识库、提示词等对话编排请在 FastGPT 平台配置，本页不重复设置。" > @@ -1601,7 +1648,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="语音配置" description="从「模型资源」中选择语音识别与语音合成。大模型、知识库与开场白由 FastGPT 应用提供，请前往 FastGPT 平台配置。" > @@ -1622,7 +1669,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="交互策略" description="设置实时视频对话时的交互体验" > @@ -1684,10 +1731,10 @@ export function AssistantPage(props: AssistantPageProps) {
-
-
+
+
} + icon={} title="OpenCode 服务配置" description="从「模型资源」中选择 OpenCode 服务资源。" > @@ -1701,7 +1748,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="提示词" description="描述助手的角色、能力和回答要求" > @@ -1714,7 +1761,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="模型与语音配置" description="配置 OpenCode 使用的大语言模型、语音识别与语音合成资源。" > @@ -1759,7 +1806,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="交互策略" description="设置实时视频对话时的交互体验" > @@ -1823,10 +1870,10 @@ export function AssistantPage(props: AssistantPageProps) { } /> -
-
+
+
-
+
-
-
- +
+
+
- Pipeline 模式 + Pipeline 模式
{form.runtimeMode === "pipeline" && ( - - + + )}
@@ -1873,25 +1920,25 @@ export function AssistantPage(props: AssistantPageProps) { } }} className={[ - "cursor-pointer rounded-2xl border p-5 text-left transition-colors", + "cursor-pointer rounded-xl border p-3.5 text-left transition-colors", form.runtimeMode === "realtime" ? "border-primary bg-primary/5 ring-1 ring-primary" : "border-hairline bg-canvas-soft hover:border-hairline-strong", ].join(" ")} >
-
-
- +
+
+
- Realtime 模式 + Realtime 模式
{form.runtimeMode === "realtime" && ( - - + + )}
@@ -1900,7 +1947,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="提示词" description="描述助手的角色、能力和回答要求" > @@ -1918,7 +1965,7 @@ export function AssistantPage(props: AssistantPageProps) { {form.runtimeMode === "pipeline" ? ( } + icon={} title="模型配置" description="从「模型资源」中选择大语言模型、语音识别与语音合成" > @@ -1963,7 +2010,7 @@ export function AssistantPage(props: AssistantPageProps) { ) : ( } + icon={} title="模型配置" description="当前模式下 ASR 与 TTS 由 Realtime 模型内置完成" > @@ -1978,7 +2025,7 @@ export function AssistantPage(props: AssistantPageProps) { )} } + icon={} title="开场白" description="助手与用户首次对话时的开场语" > @@ -1995,7 +2042,7 @@ export function AssistantPage(props: AssistantPageProps) { {form.runtimeMode === "pipeline" && ( } + icon={} title="知识库配置" description="选择助手回答时可检索的业务知识来源" > @@ -2021,7 +2068,7 @@ export function AssistantPage(props: AssistantPageProps) { )} } + icon={} title="工具" description="配置该提示词助手可以调用的工具" > @@ -2033,7 +2080,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="交互策略" description="设置实时视频对话时的交互体验" > @@ -2132,7 +2179,8 @@ function SegmentedIconButton({ function DebugDrawer({ assistantId, - asSheet = false, + overlay = false, + onClose, hasUnsavedChanges = false, onNodeActive, vision = false, @@ -2140,7 +2188,8 @@ function DebugDrawer({ dynamicVariableDefinitions = {}, }: { assistantId: string | null; - asSheet?: boolean; + overlay?: boolean; + onClose?: () => void; hasUnsavedChanges?: boolean; onNodeActive?: (nodeId: string | null) => void; vision?: boolean; @@ -2179,14 +2228,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 ( -