diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index bc92a5c..ae170b6 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -166,11 +166,18 @@ 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: + async def on_user_turn_end( + self, + content: str, + user_message: dict[str, Any] | None = None, + ) -> 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. + + ``user_message`` preserves the current universal-context message for + brains that need multimodal routing. Text-only brains can ignore it. """ self.record_user_message(content) return False @@ -222,7 +229,11 @@ class Brain(Protocol): def record_user_message(self, content: str) -> None: ... - async def on_user_turn_end(self, content: str) -> bool: ... + async def on_user_turn_end( + self, + content: str, + user_message: dict[str, Any] | None = None, + ) -> bool: ... async def on_assistant_text_start(self, turn_id: str) -> None: ... diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 064da76..cbeb412 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -69,8 +69,9 @@ class _MessageContinuation: token: int node_id: str - context_messages: list[dict[str, str]] + context_messages: list[dict[str, Any]] triggering_user_text: str + triggering_user_message: dict[str, Any] | None task: asyncio.Task[None] | None = None @@ -88,6 +89,26 @@ class ConfiguredFlowManager(FlowManager): async def _create_transition_func(self, name, handler): transition = await super()._create_transition_func(name, handler) + native_vision_handler = getattr( + handler, + "_workflow_native_vision_handler", + None, + ) + native_vision_enabled = getattr( + handler, + "_workflow_native_vision_enabled", + None, + ) + if callable(native_vision_handler) and callable(native_vision_enabled): + fallback_transition = transition + + async def vision_transition(params: FunctionCallParams) -> None: + if native_vision_enabled(): + await native_vision_handler(params) + return + await fallback_transition(params) + + transition = vision_transition if not getattr(handler, "_suppress_followup_llm", False): return transition @@ -290,14 +311,26 @@ class WorkflowBrain(BaseBrain): if content and not self._ended: self._store.record("user", content) - async def on_user_turn_end(self, content: str) -> bool: + async def on_user_turn_end( + self, + content: str, + user_message: dict[str, Any] | None = None, + ) -> bool: """Route a complete user turn before the active stage may reply.""" if not content or self._ended: return True async with self._turn_lock: - return await self._handle_user_turn_end(content) + return await self._handle_user_turn_end( + content, + user_message=user_message, + ) - async def _handle_user_turn_end(self, content: str) -> bool: + async def _handle_user_turn_end( + self, + content: str, + *, + user_message: dict[str, Any] | None = None, + ) -> bool: """Serialized implementation so one user turn cannot transition twice.""" self.record_user_message(content) self._state.begin_user_turn(content) @@ -307,7 +340,10 @@ class WorkflowBrain(BaseBrain): return True self._state.status = WorkflowStatus.ROUTING - decision = await self._edge_evaluator.evaluate(current) + decision = await self._edge_evaluator.evaluate( + current, + current_user_message=user_message, + ) if decision.status == RouteStatus.ERROR: await self._require_output().emit_error( decision.error or "工作流路由失败", @@ -320,6 +356,7 @@ class WorkflowBrain(BaseBrain): next_config = await self._follow_edge( decision.edge, triggering_user_text=content, + triggering_user_message=user_message, ) await self._activate_node_config( next_config, @@ -387,7 +424,7 @@ class WorkflowBrain(BaseBrain): def _agent_config( self, node_id: str, - leading_messages: list[dict[str, str]] | None = None, + leading_messages: list[dict[str, Any]] | None = None, ) -> NodeConfig: stage = self._engine.agent_stage_config(node_id) functions: list[FlowsFunctionSchema] = [] @@ -468,7 +505,7 @@ class WorkflowBrain(BaseBrain): def _passive_node_config( self, node_id: str, - task_messages: list[dict[str, str]] | None = None, + task_messages: list[dict[str, Any]] | None = None, ) -> NodeConfig: """Keep a non-conversational terminal node active without ending the call.""" return { @@ -643,8 +680,9 @@ class WorkflowBrain(BaseBrain): self, edge: dict, *, - leading_messages: list[dict[str, str]] | None = None, + leading_messages: list[dict[str, Any]] | None = None, triggering_user_text: str = "", + triggering_user_message: dict[str, Any] | None = None, ) -> NodeConfig: await self._begin_edge_transition(edge) context_messages = list(leading_messages or []) @@ -664,14 +702,16 @@ class WorkflowBrain(BaseBrain): str(edge.get("target") or ""), leading_messages=context_messages, triggering_user_text=triggering_user_text, + triggering_user_message=triggering_user_message, ) async def _resolve_path( self, node_id: str, *, - leading_messages: list[dict[str, str]] | None = None, + leading_messages: list[dict[str, Any]] | None = None, triggering_user_text: str = "", + triggering_user_message: dict[str, Any] | None = None, ) -> NodeConfig: context_messages = list(leading_messages or []) for hop in range(MAX_AUTOMATIC_HOPS): @@ -684,8 +724,13 @@ class WorkflowBrain(BaseBrain): triggering_user_text and self._engine.data(node_id).get("contextPolicy") == "fresh" ): + current_user_message = ( + deepcopy(triggering_user_message) + if triggering_user_message + else {"role": "user", "content": triggering_user_text} + ) agent_messages = [ - {"role": "user", "content": triggering_user_text}, + current_user_message, *context_messages, ] return self._agent_config(node_id, agent_messages) @@ -701,6 +746,7 @@ class WorkflowBrain(BaseBrain): node_id, context_messages=context_messages, triggering_user_text=triggering_user_text, + triggering_user_message=triggering_user_message, ) return self._passive_node_config(node_id, context_messages) elif node_type == "handoff": @@ -736,8 +782,9 @@ class WorkflowBrain(BaseBrain): self, node_id: str, *, - context_messages: list[dict[str, str]], + context_messages: list[dict[str, Any]], triggering_user_text: str, + triggering_user_message: dict[str, Any] | None, ) -> None: """Save the path state without waiting inside the pipeline call stack.""" token = self._next_message_token @@ -747,6 +794,7 @@ class WorkflowBrain(BaseBrain): node_id=node_id, context_messages=[dict(message) for message in context_messages], triggering_user_text=triggering_user_text, + triggering_user_message=deepcopy(triggering_user_message), ) self._state.enter(node_id, WorkflowStatus.RUNNING_MESSAGE) @@ -847,6 +895,7 @@ class WorkflowBrain(BaseBrain): edge, leading_messages=context_messages, triggering_user_text=continuation.triggering_user_text, + triggering_user_message=continuation.triggering_user_message, ) await self._activate_node_config( next_config, diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 76d6058..2179e7b 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -141,6 +141,25 @@ def _vision_uses_main_llm(cfg: AssistantConfig) -> bool: return not cfg.vision_model_resource_id and cfg.llm_support_image_input +def _workflow_vision_uses_main_llm( + cfg: AssistantConfig, + scope: dict[str, Any], +) -> bool: + """Resolve native versus auxiliary vision for the active Workflow Agent.""" + if not scope.get("enabled"): + raise ValueError("当前 Workflow Agent 节点未启用视觉能力") + if scope.get("vision_model_resource_id"): + return False + + llm_resource_id = str(scope.get("llm_resource_id") or "") + resource = cfg.workflow_model_resources.get(llm_resource_id) + if not resource: + raise ValueError(f"当前 Workflow Agent 的 LLM 资源未加载:{llm_resource_id}") + if not resource.support_image_input: + raise ValueError("当前 Workflow Agent 的 LLM 不支持图片输入") + return True + + def _image_data_uri(frame: UserImageRawFrame) -> str: if not frame.format: raise ValueError("摄像头图片帧缺少 format,无法编码给视觉模型") @@ -388,6 +407,11 @@ async def run_pipeline( "vision_model_resource_id": None, "llm_resource_id": None, } + + def active_vision_uses_main_llm() -> bool: + if cfg.type == "workflow": + return _workflow_vision_uses_main_llm(cfg, workflow_vision_scope) + return vision_native_mode vision_schema = FunctionSchema( name=VISION_TOOL_NAME, description=( @@ -510,6 +534,27 @@ async def run_pipeline( raise ValueError(f"视觉模型资源未加载:{vision_resource_id}") if cfg.type == "workflow" and vision_enabled: + async def native_flow_fetch_user_image(params: FunctionCallParams) -> None: + question = str(params.arguments.get("question") or "请描述当前画面。") + user_id = vision_state.get("client_id") + if not user_id: + await params.result_callback( + { + "status": "no_video_client", + "message": "当前还没有可用的摄像头视频流。", + } + ) + return + request = UserImageRequestFrame( + user_id=user_id, + text=question, + append_to_context=True, + function_name=params.function_name, + tool_call_id=params.tool_call_id, + result_callback=params.result_callback, + ) + await params.llm.push_frame(request, FrameDirection.UPSTREAM) + async def flow_fetch_user_image(args, _flow_manager): if not workflow_vision_scope.get("enabled"): return { @@ -547,6 +592,20 @@ async def run_pipeline( logger.warning(f"Workflow 视觉理解失败:{exc}") return {"status": "error", "message": "视觉理解暂时不可用。"} + # ConfiguredFlowManager keeps the Flows handler for auxiliary models, + # but uses Pipecat's native function-call image path when the active + # Agent selected its own visual-capable LLM. + setattr( + flow_fetch_user_image, + "_workflow_native_vision_handler", + native_flow_fetch_user_image, + ) + setattr( + flow_fetch_user_image, + "_workflow_native_vision_enabled", + active_vision_uses_main_llm, + ) + workflow_vision_function = FlowsFunctionSchema( name=VISION_TOOL_NAME, description=vision_schema.description, @@ -710,7 +769,8 @@ async def run_pipeline( user_id = vision_state.get("client_id") if not user_id: raise ValueError("当前没有可用的摄像头视频流") - analysis_cfg = None if vision_native_mode else active_vision_config() + native_vision = active_vision_uses_main_llm() + analysis_cfg = None if native_vision else active_vision_config() request = UserImageRequestFrame( user_id=user_id, @@ -722,7 +782,7 @@ async def run_pipeline( except asyncio.TimeoutError as exc: raise ValueError("等待摄像头视频帧超时") from exc - if vision_native_mode: + if native_vision: image_frame.text = value.prompt_text image_frame.append_to_context = True image_frame.request = None diff --git a/backend/services/pipecat/processors.py b/backend/services/pipecat/processors.py index b7bf2aa..8039382 100644 --- a/backend/services/pipecat/processors.py +++ b/backend/services/pipecat/processors.py @@ -713,7 +713,10 @@ class UserTurnRoutingProcessor(FrameProcessor): self._last_user_message = user_message content = message_text(user_message) - handled = await self._brain.on_user_turn_end(content) + handled = await self._brain.on_user_turn_end( + content, + user_message=user_message, + ) if not handled: await self.push_frame(frame, direction) diff --git a/backend/services/workflow/agent.py b/backend/services/workflow/agent.py index adbd2a6..09df9b1 100644 --- a/backend/services/workflow/agent.py +++ b/backend/services/workflow/agent.py @@ -2,6 +2,8 @@ from __future__ import annotations +from typing import Any + from models import AssistantConfig from pipecat.flows import ContextStrategy, ContextStrategyConfig, NodeConfig from pipecat.frames.frames import LLMUpdateSettingsFrame @@ -106,7 +108,7 @@ class WorkflowAgentStage: node_id: str, *, functions: list, - leading_messages: list[dict[str, str]] | None = None, + leading_messages: list[dict[str, Any]] | None = None, ) -> NodeConfig: data = self._engine.data(node_id) strategy = ( diff --git a/backend/services/workflow/routing.py b/backend/services/workflow/routing.py index 5e9a5f7..a4e300e 100644 --- a/backend/services/workflow/routing.py +++ b/backend/services/workflow/routing.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections.abc import Callable +from typing import Any from services.runtime_variables import DynamicVariableStore from services.workflow.models import EdgeEvaluation, RouteStatus @@ -23,7 +24,12 @@ class WorkflowEdgeEvaluator: self._store = store self._router_for_node = router_for_node - async def evaluate(self, node_id: str) -> EdgeEvaluation: + async def evaluate( + self, + node_id: str, + *, + current_user_message: dict[str, Any] | None = None, + ) -> EdgeEvaluation: """Select the first matching conditional path, then the default path.""" outgoing = self._engine.outgoing(node_id) expression_edge = self._engine.deterministic_edge( @@ -61,6 +67,7 @@ class WorkflowEdgeEvaluator: node_prompt=self._engine.routing_prompt(node_id, self._store), edges=llm_edges, history=self._store.history, + current_user_message=current_user_message, variables={ key: value for key, value in self._store.values.items() @@ -94,4 +101,3 @@ class WorkflowEdgeEvaluator: if edge is None: return EdgeEvaluation(status=RouteStatus.NO_MATCH) return EdgeEvaluation(status=RouteStatus.MATCHED, edge=edge) - diff --git a/backend/services/workflow_router.py b/backend/services/workflow_router.py index 11a09c1..93585bd 100644 --- a/backend/services/workflow_router.py +++ b/backend/services/workflow_router.py @@ -9,6 +9,7 @@ from __future__ import annotations import json from collections.abc import Callable +from copy import deepcopy from typing import Any from loguru import logger @@ -24,6 +25,32 @@ STAY_ON_CURRENT_AGENT = STAY_ON_CURRENT_NODE MAX_ROUTING_HISTORY_ENTRIES = 20 +def _routing_user_message( + routing_input: str, + current_user_message: dict[str, Any] | None, +) -> dict[str, Any]: + """Combine routing metadata with the current text or multimodal turn.""" + if not current_user_message: + return {"role": "user", "content": routing_input} + + content = current_user_message.get("content") + if not isinstance(content, list): + current_text = str(content or "").strip() + suffix = f"\n\n[当前用户输入]\n{current_text}" if current_text else "" + return {"role": "user", "content": f"{routing_input}{suffix}"} + + return { + "role": "user", + "content": [ + { + "type": "text", + "text": f"{routing_input}\n\n[当前用户输入如下]", + }, + *deepcopy(content), + ], + } + + class WorkflowLLMRouter: """Select one LLM edge without allowing the router to speak.""" @@ -40,6 +67,7 @@ class WorkflowLLMRouter: variables: dict[str, Any], edge_name: Callable[[dict[str, Any]], str], edge_description: Callable[[dict[str, Any]], str], + current_user_message: dict[str, Any] | None = None, ) -> LLMRouteResult: """Return a typed match, no-match or technical error.""" if not edges: @@ -85,7 +113,13 @@ class WorkflowLLMRouter: f"当前节点任务：{node_prompt or '未配置'}\n" f"转移条件：\n{ordered_conditions}" ) - recent_history = history[-MAX_ROUTING_HISTORY_ENTRIES:] + # WorkflowBrain records the current turn before routing. When the full + # current message is supplied separately, keep only earlier history so + # the text is not duplicated and the image remains attached to its turn. + routing_history = ( + history[:-1] if current_user_message and history else history + ) + recent_history = routing_history[-MAX_ROUTING_HISTORY_ENTRIES:] routing_input = json.dumps( { "conversation": recent_history, @@ -108,7 +142,7 @@ class WorkflowLLMRouter: model=self._cfg.model, messages=[ {"role": "system", "content": router_prompt}, - {"role": "user", "content": routing_input}, + _routing_user_message(routing_input, current_user_message), ], tools=tools, tool_choice="required", diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index dba1b8f..659fe42 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -6,6 +6,7 @@ from types import SimpleNamespace from unittest.mock import AsyncMock, patch from models import AssistantConfig, RuntimeTool +from pipecat.flows import FlowManager from pipecat.frames.frames import ( LLMContextFrame, LLMFullResponseEndFrame, @@ -28,7 +29,7 @@ from services.brains.dify_llm import ( last_user_text, normalize_api_base, ) -from services.brains.workflow_brain import WorkflowBrain +from services.brains.workflow_brain import ConfiguredFlowManager, WorkflowBrain from services.runtime_variables import prepare_dynamic_config from services.action_runtime import ActionOutcome, ActionStatus from services.workflow.models import ( @@ -789,6 +790,45 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): + async def test_flow_manager_dispatches_native_vision_without_auxiliary_handler(self): + manager = object.__new__(ConfiguredFlowManager) + fallback_transition = AsyncMock() + native_handler = AsyncMock() + native_enabled = {"value": True} + + async def flow_handler(_args, _manager): + return {"status": "ok"} + + setattr( + flow_handler, + "_workflow_native_vision_handler", + native_handler, + ) + setattr( + flow_handler, + "_workflow_native_vision_enabled", + lambda: native_enabled["value"], + ) + + with patch.object( + FlowManager, + "_create_transition_func", + new=AsyncMock(return_value=fallback_transition), + ): + transition = await manager._create_transition_func( + "fetch_user_image", + flow_handler, + ) + + params = SimpleNamespace() + await transition(params) + native_handler.assert_awaited_once_with(params) + fallback_transition.assert_not_awaited() + + native_enabled["value"] = False + await transition(params) + fallback_transition.assert_awaited_once_with(params) + def test_client_tool_session_wait_disables_flow_timeout(self): brain = WorkflowBrain( { @@ -1922,9 +1962,11 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): class FakeRouter: def __init__(self): self.calls = 0 + self.current_user_message = None - async def select_edge(self, **_kwargs): + async def select_edge(self, **kwargs): self.calls += 1 + self.current_user_message = kwargs.get("current_user_message") return LLMRouteResult( status=RouteStatus.MATCHED, function_name="goto_eat", @@ -1939,13 +1981,27 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(manager.current_node, "start") self.assertEqual(router.calls, 0) - handled = await brain.on_user_turn_end("我想吃饭") + image_message = { + "role": "user", + "content": [ + {"type": "text", "text": "我想吃饭"}, + { + "type": "image_url", + "image_url": {"url": "data:image/jpeg;base64,AA=="}, + }, + ], + } + handled = await brain.on_user_turn_end( + "我想吃饭", + user_message=image_message, + ) self.assertTrue(handled) self.assertEqual(router.calls, 1) + self.assertEqual(router.current_user_message, image_message) self.assertEqual(manager.current_node, "eat") self.assertIn( - {"role": "user", "content": "我想吃饭"}, + image_message, manager.config["task_messages"], ) self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in queued)) diff --git a/backend/tests/test_pipeline_knowledge.py b/backend/tests/test_pipeline_knowledge.py index 3eeadad..4092cf4 100644 --- a/backend/tests/test_pipeline_knowledge.py +++ b/backend/tests/test_pipeline_knowledge.py @@ -1,6 +1,6 @@ import unittest -from models import AssistantConfig +from models import AssistantConfig, RuntimeModelResource from pipecat.frames.frames import LLMContextFrame from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameDirection @@ -9,6 +9,7 @@ from services.pipecat.pipeline import ( KnowledgeRetrievalProcessor, UserTurnRoutingProcessor, _knowledge_tool_description, + _workflow_vision_uses_main_llm, ) @@ -67,8 +68,8 @@ class UserTurnRoutingProcessorTest(unittest.IsolatedAsyncioTestCase): def __init__(self): self.turns = [] - async def on_user_turn_end(self, content): - self.turns.append(content) + async def on_user_turn_end(self, content, user_message=None): + self.turns.append((content, user_message)) return True brain = FakeBrain() @@ -83,13 +84,19 @@ class UserTurnRoutingProcessorTest(unittest.IsolatedAsyncioTestCase): frame = LLMContextFrame(context) await processor.process_frame(frame, FrameDirection.DOWNSTREAM) - self.assertEqual(brain.turns, ["我叫李白"]) + self.assertEqual( + brain.turns, + [("我叫李白", {"role": "user", "content": "我叫李白"})], + ) 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( + brain.turns, + [("我叫李白", {"role": "user", "content": "我叫李白"})], + ) self.assertEqual(forwarded, [(frame, FrameDirection.DOWNSTREAM)]) async def test_routes_multimodal_user_message_by_its_text_part(self): @@ -97,8 +104,8 @@ class UserTurnRoutingProcessorTest(unittest.IsolatedAsyncioTestCase): def __init__(self): self.turns = [] - async def on_user_turn_end(self, content): - self.turns.append(content) + async def on_user_turn_end(self, content, user_message=None): + self.turns.append((content, user_message)) return False brain = FakeBrain() @@ -124,7 +131,87 @@ class UserTurnRoutingProcessorTest(unittest.IsolatedAsyncioTestCase): FrameDirection.DOWNSTREAM, ) - self.assertEqual(brain.turns, ["看看这张照片"]) + self.assertEqual( + brain.turns, + [ + ( + "看看这张照片", + { + "role": "user", + "content": [ + {"type": "text", "text": "看看这张照片"}, + { + "type": "image_url", + "image_url": { + "url": "data:image/jpeg;base64,AA==" + }, + }, + ], + }, + ) + ], + ) + + +class WorkflowVisionModeTest(unittest.TestCase): + def test_uses_active_agent_llm_only_without_auxiliary_model(self): + cfg = AssistantConfig( + type="workflow", + workflow_model_resources={ + "agent_llm": RuntimeModelResource( + id="agent_llm", + name="视觉 Agent", + capability="LLM", + interface_type="openai-llm", + support_image_input=True, + ) + }, + ) + + self.assertTrue( + _workflow_vision_uses_main_llm( + cfg, + { + "enabled": True, + "llm_resource_id": "agent_llm", + "vision_model_resource_id": None, + }, + ) + ) + self.assertFalse( + _workflow_vision_uses_main_llm( + cfg, + { + "enabled": True, + "llm_resource_id": "agent_llm", + "vision_model_resource_id": "auxiliary_vision", + }, + ) + ) + + def test_rejects_a_non_visual_active_agent_llm(self): + cfg = AssistantConfig( + type="workflow", + workflow_model_resources={ + "text_llm": RuntimeModelResource( + id="text_llm", + name="文本 Agent", + capability="LLM", + interface_type="openai-llm", + support_image_input=False, + ) + }, + ) + + with self.assertRaisesRegex(ValueError, "不支持图片输入"): + _workflow_vision_uses_main_llm( + cfg, + { + "enabled": True, + "llm_resource_id": "text_llm", + "vision_model_resource_id": None, + }, + ) async def _async_none(): diff --git a/backend/tests/test_workflow_router.py b/backend/tests/test_workflow_router.py index ae84e6c..4e51c98 100644 --- a/backend/tests/test_workflow_router.py +++ b/backend/tests/test_workflow_router.py @@ -72,6 +72,77 @@ class WorkflowLLMRouterTest(unittest.IsolatedAsyncioTestCase): ) self.assertNotIn("developer", str(requests[0]["messages"])) + async def test_routes_with_the_current_multimodal_user_message(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_confirm", + arguments="{}", + ) + ) + ] + ) + ) + ] + ) + + class FakeClient: + def __init__(self, **_kwargs): + self.chat = SimpleNamespace(completions=FakeCompletions()) + + async def close(self): + return None + + router = WorkflowLLMRouter( + AssistantConfig( + type="workflow", + model="visual-model", + llm_api_key="secret", + llm_base_url="https://llm.test/v1", + ) + ) + image_message = { + "role": "user", + "content": [ + {"type": "text", "text": "请检查车牌照片"}, + { + "type": "image_url", + "image_url": {"url": "data:image/jpeg;base64,AA=="}, + }, + ], + } + + with patch("services.workflow_router.AsyncOpenAI", FakeClient): + selected = await router.select_edge( + node_name="采集车牌", + node_prompt="确认车牌照片是否清晰", + edges=[{"id": "confirm", "data": {"condition": "车牌清晰"}}], + history=[ + {"role": "user", "message": "之前的消息"}, + {"role": "user", "message": "请检查车牌照片"}, + ], + variables={}, + edge_name=lambda _edge: "goto_confirm", + edge_description=lambda _edge: "车牌清晰", + current_user_message=image_message, + ) + + self.assertEqual(selected.status, RouteStatus.MATCHED) + content = requests[0]["messages"][1]["content"] + self.assertIsInstance(content, list) + self.assertEqual(content[-1], image_message["content"][-1]) + self.assertIn("之前的消息", content[0]["text"]) + self.assertEqual(content[0]["text"].count("请检查车牌照片"), 0) + if __name__ == "__main__": unittest.main()