wx44wx/ai-video-fullstack
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: support vision prompt modes

Browse Source
This commit is contained in:
Xin Wang
2026-08-12 19:13:04 +08:00
parent 09b07751b6
commit 94a02d7e2b
29 changed files with 676 additions and 439 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_brains.py
+9 -4
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
import asyncio
import unittest
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock, MagicMock, patch
from models import AssistantConfig, RuntimeTool
from pipecat.flows import FlowManager
@@ -1782,7 +1782,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
}
)
scopes = []
vision_function = object()
vision_function = MagicMock(side_effect=lambda _stage: object())
async def queue_frame(_frame):
pass
@@ -1800,13 +1800,18 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
await brain._apply_agent_stage("agent")
inherited_config = brain._agent_config("agent")
self.assertIn(vision_function, inherited_config["functions"])
self.assertEqual(len(inherited_config["functions"]), 1)
vision_function.assert_called_once()
selected_stage = vision_function.call_args.args[0]
self.assertEqual(selected_stage.vision_prompt_mode, "runtime")
self.assertIn("fetch_user_image", inherited_config["role_message"])
self.assertEqual(
scopes[-1],
{
"enabled": True,
"vision_model_resource_id": "vision_global",
"vision_prompt_mode": "runtime",
"vision_prompt": "",
"llm_resource_id": "llm_global",
},
)
@@ -1821,7 +1826,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
)
await brain._apply_agent_stage("agent")
custom_config = brain._agent_config("agent")
self.assertNotIn(vision_function, custom_config["functions"])
self.assertEqual(custom_config["functions"], [])
self.assertNotIn("fetch_user_image", custom_config["role_message"])
self.assertFalse(scopes[-1]["enabled"])
backend/tests/test_vision_focus.py
-126
View File
@@ -1,126 +0,0 @@
from __future__ import annotations
import unittest
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
from models import RuntimeModelResource
from services.vision import (
VISION_ANALYSIS_SYSTEM_PROMPT,
VISION_SYSTEM_HINT,
generate_vision_focus_prompt,
vision_analysis_system_prompt,
vision_system_hint,
vision_tool_description,
)
from services.workflow_engine import WorkflowEngine
class VisionFocusPromptTest(unittest.TestCase):
def test_focus_extends_but_does_not_replace_safety_prompts(self):
focus = "优先识别车辆数量和人员状态，只报告可确认的信息。"
system_hint = vision_system_hint(focus)
analysis_prompt = vision_analysis_system_prompt(focus)
self.assertIn(VISION_SYSTEM_HINT, system_hint)
self.assertIn(focus, system_hint)
self.assertIn(VISION_ANALYSIS_SYSTEM_PROMPT, analysis_prompt)
self.assertIn(focus, analysis_prompt)
self.assertIn("优先回答当前用户问题", analysis_prompt)
def test_empty_focus_preserves_previous_runtime_prompts(self):
self.assertEqual(vision_system_hint(), VISION_SYSTEM_HINT)
self.assertEqual(
vision_analysis_system_prompt(),
VISION_ANALYSIS_SYSTEM_PROMPT,
)
def test_tool_description_compacts_and_limits_focus(self):
description = vision_tool_description("车辆\n 数量 " * 300)
self.assertIn("当前业务重点", description)
self.assertNotIn("\n", description)
self.assertLess(len(description), 700)
def test_workflow_resolves_global_and_node_focus(self):
graph = {
"specVersion": 3,
"settings": {
"visionEnabled": True,
"visionFocusPrompt": "全局关注目标",
},
"nodes": [
{
"id": "global-agent",
"type": "agent",
"data": {"name": "全局 Agent", "prompt": "处理任务"},
},
{
"id": "custom-agent",
"type": "agent",
"data": {
"name": "独立 Agent",
"prompt": "检查车辆",
"inheritGlobalConfig": False,
"visionEnabled": True,
"visionFocusPrompt": "节点关注目标",
},
},
],
"edges": [],
}
engine = WorkflowEngine(graph)
self.assertEqual(
engine.agent_stage_config("global-agent").vision_focus_prompt,
"全局关注目标",
)
self.assertEqual(
engine.agent_stage_config("custom-agent").vision_focus_prompt,
"节点关注目标",
)
class VisionFocusGenerationTest(unittest.IsolatedAsyncioTestCase):
async def test_generates_editable_focus_with_selected_resource(self):
resource = RuntimeModelResource(
id="llm_1",
name="生成模型",
capability="LLM",
interface_type="openai-llm",
values={
"apiUrl": "https://llm.example/v1",
"modelId": "focus-model",
},
secrets={"apiKey": "secret"},
)
client = MagicMock()
client.chat.completions.create = AsyncMock(
return_value=SimpleNamespace(
choices=[
SimpleNamespace(
message=SimpleNamespace(
content="优先识别车辆数量、损伤位置和人员状态。"
)
)
]
)
)
client.close = AsyncMock()
with patch("services.vision.AsyncOpenAI", return_value=client):
result = await generate_vision_focus_prompt(
resource,
"你是交通事故处理助手，需要了解车辆和人员情况。",
)
self.assertEqual(result, "优先识别车辆数量、损伤位置和人员状态。")
request = client.chat.completions.create.await_args.kwargs
self.assertEqual(request["model"], "focus-model")
self.assertIn("交通事故处理助手", request["messages"][1]["content"])
client.close.assert_awaited_once()
if __name__ == "__main__":
unittest.main()
backend/tests/test_vision_prompt_modes.py
+227
View File
@@ -0,0 +1,227 @@
from __future__ import annotations
import unittest
from types import SimpleNamespace
from models import AssistantConfig
from services.brains.base import RealtimeBrainRuntime
from services.node_specs import normalize_graph, validate_graph
from services.vision import (
VISION_SYSTEM_HINT,
resolve_vision_question,
vision_system_hint,
vision_tool_schema_parts,
)
from services.workflow_engine import WorkflowEngine
from services.workflow.realtime import WorkflowRealtimeController
class VisionPromptModeTest(unittest.TestCase):
def test_fixed_mode_has_no_runtime_question_parameter(self):
description, properties, required = vision_tool_schema_parts(
"fixed",
"识别车辆数量和人员状态",
)
self.assertIn("识别车辆数量和人员状态", description)
self.assertEqual(properties, {})
self.assertEqual(required, [])
self.assertEqual(
resolve_vision_question("fixed", " 识别车辆数量 ", {}),
"识别车辆数量",
)
def test_runtime_mode_requires_model_generated_question(self):
description, properties, required = vision_tool_schema_parts("runtime")
self.assertNotIn("固定引导", description)
self.assertIn("question", properties)
self.assertEqual(required, ["question"])
self.assertEqual(
resolve_vision_question(
"runtime",
"不会使用这个配置",
{"question": "用户手里拿着什么？"},
),
"用户手里拿着什么？",
)
def test_guided_mode_exposes_guide_but_uses_runtime_question(self):
guide = "只询问画面中可直接确认的车辆损伤"
description, properties, required = vision_tool_schema_parts(
"guided",
guide,
)
self.assertIn(guide, description)
self.assertIn(guide, properties["question"]["description"])
self.assertEqual(required, ["question"])
self.assertEqual(
resolve_vision_question(
"guided",
guide,
{"question": "右侧车门是否有明显凹陷？"},
),
"右侧车门是否有明显凹陷？",
)
def test_mode_specific_system_hints(self):
self.assertIn(VISION_SYSTEM_HINT, vision_system_hint("runtime"))
self.assertIn("不需要生成", vision_system_hint("fixed", "固定问题"))
self.assertIn("固定引导", vision_system_hint("guided", "业务引导"))
def test_missing_required_config_or_question_is_rejected(self):
with self.assertRaisesRegex(ValueError, "固定视觉提示词为空"):
resolve_vision_question("fixed", "", {})
with self.assertRaisesRegex(ValueError, "视觉问题为空"):
resolve_vision_question("runtime", "", {})
class WorkflowVisionPromptModeTest(unittest.TestCase):
def test_workflow_resolves_global_and_node_modes(self):
graph = {
"specVersion": 3,
"settings": {
"visionEnabled": True,
"visionPromptMode": "fixed",
"visionPrompt": "全局固定问题",
},
"nodes": [
{
"id": "global-agent",
"type": "agent",
"data": {"name": "全局 Agent", "prompt": "处理任务"},
},
{
"id": "custom-agent",
"type": "agent",
"data": {
"name": "独立 Agent",
"prompt": "检查车辆",
"inheritGlobalConfig": False,
"visionEnabled": True,
"visionPromptMode": "guided",
"visionPrompt": "节点引导",
},
},
],
"edges": [],
}
engine = WorkflowEngine(graph)
global_stage = engine.agent_stage_config("global-agent")
self.assertEqual(global_stage.vision_prompt_mode, "fixed")
self.assertEqual(global_stage.vision_prompt, "全局固定问题")
custom_stage = engine.agent_stage_config("custom-agent")
self.assertEqual(custom_stage.vision_prompt_mode, "guided")
self.assertEqual(custom_stage.vision_prompt, "节点引导")
def test_legacy_focus_is_migrated_to_guided_mode(self):
normalized = normalize_graph(
{
"specVersion": 3,
"settings": {"visionFocusPrompt": "旧全局关注目标"},
"nodes": [],
"edges": [],
}
)
self.assertEqual(normalized["settings"]["visionPromptMode"], "guided")
self.assertEqual(normalized["settings"]["visionPrompt"], "旧全局关注目标")
self.assertNotIn("visionFocusPrompt", normalized["settings"])
def test_fixed_or_guided_mode_requires_prompt_when_vision_is_enabled(self):
graph = {
"specVersion": 3,
"settings": {
"visionEnabled": True,
"visionPromptMode": "fixed",
"visionPrompt": "",
},
"nodes": [
{
"id": "agent",
"type": "agent",
"data": {"name": "Agent", "prompt": "处理任务"},
}
],
"edges": [],
}
self.assertTrue(
any("必须填写视觉提示词" in error for error in validate_graph(graph))
)
def test_invalid_mode_is_reported_instead_of_silently_rewritten(self):
graph = {
"specVersion": 3,
"settings": {"visionPromptMode": "unexpected"},
"nodes": [
{
"id": "agent",
"type": "agent",
"data": {"name": "Agent", "prompt": "处理任务"},
}
],
"edges": [],
}
self.assertTrue(
any("视觉提示词模式无效" in error for error in validate_graph(graph))
)
def test_realtime_tool_contract_changes_with_mode(self):
graph = {
"specVersion": 3,
"settings": {},
"nodes": [
{
"id": "fixed-agent",
"type": "agent",
"data": {
"name": "固定",
"prompt": "处理任务",
"inheritGlobalConfig": False,
"visionEnabled": True,
"visionPromptMode": "fixed",
"visionPrompt": "检查安全帽",
},
},
{
"id": "guided-agent",
"type": "agent",
"data": {
"name": "引导",
"prompt": "处理任务",
"inheritGlobalConfig": False,
"visionEnabled": True,
"visionPromptMode": "guided",
"visionPrompt": "只询问安全设备",
},
},
],
"edges": [],
}
engine = WorkflowEngine(graph)
controller = WorkflowRealtimeController(
cfg=AssistantConfig(type="workflow"),
engine=engine,
store=SimpleNamespace(),
runtime=RealtimeBrainRuntime(
realtime=SimpleNamespace(),
queue_frame=lambda _frame: None,
call_end=SimpleNamespace(ending=False),
),
)
fixed_tool, _fixed_handler = controller._vision_tool("fixed-agent", 0)
guided_tool, _guided_handler = controller._vision_tool("guided-agent", 0)
self.assertEqual(fixed_tool.properties, {})
self.assertEqual(fixed_tool.required, ())
self.assertIn("只询问安全设备", guided_tool.description)
self.assertEqual(guided_tool.required, ("question",))
if __name__ == "__main__":
unittest.main()