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

merge: add workflow vision configuration

Browse Source
This commit is contained in:
Xin Wang
2026-07-18 00:10:28 +08:00
parent e39bb48ba8 28c6380d8a
commit cd319d27b9
23 changed files with 553 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
backend/models.py
View File

@@ -47,6 +47,7 @@ class RuntimeModelResource(BaseModel):
interface_type: str
values: dict = Field(default_factory=dict)
secrets: dict = Field(default_factory=dict)
support_image_input: bool = False
class RuntimeKnowledgeBase(BaseModel):

32
backend/routes/assistants.py
View File

@@ -16,6 +16,7 @@ from schemas import AssistantOut, AssistantUpsert
from services.auth import require_admin
from services.masking import mask, resolve_incoming_key
from services.node_specs import graph_references, normalize_graph, validate_graph
from services.workflow_engine import WorkflowEngine
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
@@ -35,6 +36,10 @@ def _validate_workflow(body: AssistantUpsert) -> None:
errors = validate_graph(body.graph)
if errors:
raise HTTPException(400, "工作流校验失败:" + ";".join(errors))
# Graph settings are the source of truth. The flat flag is only a session
# capability summary used by clients to decide whether to open the camera.
body.vision_enabled = WorkflowEngine(body.graph).uses_vision()
body.vision_model_resource_id = None
refs = graph_references(body.graph)
body.tool_ids = list(dict.fromkeys([*body.tool_ids, *sorted(refs["tools"])]))
@@ -45,8 +50,10 @@ async def _validate_workflow_references(
if body.type != "workflow" or not body.graph.get("nodes"):
return
graph = body.graph
engine = WorkflowEngine(graph)
settings = graph.get("settings") or {}
resource_expectations: dict[str, str] = {}
vision_resource_ids: set[str] = set()
for key, capability in (
("defaultLlmResourceId", "LLM"),
("defaultAsrResourceId", "ASR"),
@@ -71,10 +78,32 @@ async def _validate_workflow_references(
resource_expectations[str(data["ttsResourceId"])] = "TTS"
if data.get("knowledgeBaseId"):
knowledge_ids.add(str(data["knowledgeBaseId"]))
for node_id, node in engine.nodes.items():
if node.get("type") != "agent":
continue
stage = engine.agent_stage_config(node_id)
if not stage.vision_enabled:
continue
resource_id = (
stage.vision_model_resource_id or stage.llm_resource_id
)
if not resource_id:
raise HTTPException(
400,
f"Agent 节点 {node_id} 开启视觉理解时必须选择"
"支持图片输入的大语言模型或视觉模型",
)
resource_expectations[resource_id] = "LLM"
vision_resource_ids.add(resource_id)
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}")
if resource_id in vision_resource_ids and not resource.support_image_input:
raise HTTPException(
400,
f"Workflow 视觉模型必须支持图片输入:{resource_id}",
)
for knowledge_id in knowledge_ids:
knowledge = await session.get(KnowledgeBase, knowledge_id)
if not knowledge or knowledge.status != "active":
@@ -84,6 +113,9 @@ async def _validate_workflow_references(
async def _validate_vision_model(
session: AsyncSession, body: AssistantUpsert
) -> None:
# Workflow 的视觉模型属于图中的全局/Agent 配置，由上面的图引用校验处理。
if body.type == "workflow":
return
if body.vision_enabled:
if body.vision_model_resource_id:
resource = await session.get(ModelResource, body.vision_model_resource_id)

9
backend/routes/voice_webrtc.py
View File

@@ -163,8 +163,13 @@ async def _handle_offer_payload(payload, peers):
else:
cfg = await _resolve_config(offer) # 解析放在建连前,配置错就别建连
# 服务端助手配置是视觉理解的唯一授权来源；客户端 offer 只负责携带媒体轨。
vision_enabled = cfg.vision_enabled
if vision_enabled:
if cfg.type == "workflow":
from services.workflow_engine import WorkflowEngine
vision_enabled = WorkflowEngine(cfg.graph).uses_vision()
else:
vision_enabled = cfg.vision_enabled
if vision_enabled and cfg.type != "workflow":
has_native_vision = (
not cfg.vision_model_resource_id and cfg.llm_support_image_input
)

2
backend/services/brains/base.py
View File

@@ -81,6 +81,8 @@ class BrainRuntime:
Callable[[str | None, str | None, str | None], Awaitable[None]] | None
) = None
set_knowledge_scope: Callable[[dict[str, Any]], None] | None = None
set_vision_scope: Callable[[dict[str, Any]], None] | None = None
vision_function: Any = None
set_input_enabled: Callable[[bool], None] | None = None
apply_turn_config: (
Callable[[bool, dict[str, Any]], Awaitable[None]] | None

4
backend/services/brains/workflow_brain.py
View File

@@ -333,6 +333,8 @@ class WorkflowBrain(BaseBrain):
knowledge_function = self._knowledge_function(node_id)
if knowledge_function:
functions.append(knowledge_function)
if stage.vision_enabled and self._require_runtime().vision_function:
functions.append(self._require_runtime().vision_function)
return self._require_agent_stage().node_config(
node_id,
functions=functions,
@@ -703,6 +705,8 @@ class WorkflowBrain(BaseBrain):
runtime = self._require_runtime()
if runtime.set_knowledge_scope:
runtime.set_knowledge_scope({"mode": "disabled"})
if runtime.set_vision_scope:
runtime.set_vision_scope({"enabled": False})
if runtime.set_input_enabled:
runtime.set_input_enabled(False)
data = self._engine.data(node_id)

1
backend/services/config_resolver.py
View File

@@ -199,6 +199,7 @@ async def resolve_runtime_config(
interface_type=resource.interface_type,
values=resource.values or {},
secrets=resource.secrets or {},
support_image_input=bool(resource.support_image_input),
)
for resource in resources
if resource.enabled

16
backend/services/node_specs.py
View File

@@ -49,7 +49,7 @@ NODE_SPECS: list[dict[str, Any]] = [
"name": "agent",
"displayName": "Agent",
"category": "conversation_node",
"description": "阶段智能体：绑定上下文、工具、知识库及 ASR/TTS 资源。",
"description": "阶段智能体：绑定上下文、工具、知识库、视觉及语音资源。",
"icon": "Bot",
"accent": "sky",
"addable": True,
@@ -145,6 +145,8 @@ def _normalize_agent_data(data: dict[str, Any]) -> None:
data.get("llmResourceId"),
data.get("asrResourceId"),
data.get("ttsResourceId"),
data.get("visionEnabled"),
data.get("visionModelResourceId"),
data.get("knowledgeBaseId"),
data.get("toolIds"),
)
@@ -157,6 +159,8 @@ def _normalize_settings(settings: dict[str, Any], *, global_prompt: str = "") ->
settings.setdefault("defaultLlmResourceId", "")
settings.setdefault("defaultAsrResourceId", "")
settings.setdefault("defaultTtsResourceId", "")
settings.setdefault("visionEnabled", False)
settings.setdefault("visionModelResourceId", "")
settings.setdefault("toolIds", [])
settings.setdefault("knowledgeBaseId", "")
settings.setdefault("knowledgeMode", "automatic")
@@ -450,6 +454,11 @@ def graph_references(graph: dict[str, Any]) -> dict[str, set[str]]:
settings.get("defaultLlmResourceId"),
settings.get("defaultAsrResourceId"),
settings.get("defaultTtsResourceId"),
(
settings.get("visionModelResourceId")
if settings.get("visionEnabled")
else None
),
)
if value
}
@@ -469,6 +478,11 @@ def graph_references(graph: dict[str, Any]) -> dict[str, set[str]]:
data.get("llmResourceId"),
data.get("asrResourceId"),
data.get("ttsResourceId"),
(
data.get("visionModelResourceId")
if data.get("visionEnabled")
else None
),
):
if resource_id:
resources.add(str(resource_id))

77
backend/services/pipecat/pipeline.py
View File

@@ -29,6 +29,14 @@ from services.pipecat.service_factory import (
)
from db.session import SessionLocal
from services.knowledge import search as search_knowledge
from services.vision import (
VISION_ANALYSIS_SYSTEM_PROMPT,
VISION_SYSTEM_HINT,
VISION_TOOL_NAME,
config_with_main_llm_as_vision,
config_with_vision_resource,
)
from services.workflow_engine import WorkflowEngine
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
@@ -82,15 +90,6 @@ from services.pipecat.pipeline_events import (
from pipecat.workers.runner import WorkerRunner
VISION_TOOL_NAME = "fetch_user_image"
VISION_SYSTEM_HINT = (
"当前会话打开了视觉理解。用户询问当前画面、摄像头里有什么、人物/物品/"
"环境状态或需要你看一眼时,调用 fetch_user_image 获取当前视频帧,再基于画面回答。"
)
VISION_ANALYSIS_SYSTEM_PROMPT = (
"你是一个视觉理解模型。请只根据图片内容和用户问题给出准确、简洁的中文观察结果。"
"如果画面不足以判断,请明确说明不确定。"
)
KNOWLEDGE_TOOL_NAME = "search_knowledge_base"
AUTOMATIC_KNOWLEDGE_SYSTEM_HINT = (
"你已连接内部知识库。系统会在每轮用户问题前自动提供相关资料；"
@@ -205,6 +204,9 @@ async def run_pipeline(
只要有 .input() / .output() / event_handler 即可。
cfg: 助手配置(随请求内联传入)。
"""
if cfg.type == "workflow":
vision_enabled = WorkflowEngine(cfg.graph).uses_vision()
logger.info(
f"启动管线: assistant={cfg.name} type={cfg.type} "
f"mode={cfg.runtimeMode} vision={vision_enabled}"
@@ -298,7 +300,7 @@ async def run_pipeline(
def with_vision_hint(text: str) -> str:
hints = []
if vision_enabled:
if vision_enabled and cfg.type != "workflow":
hints.append(VISION_SYSTEM_HINT)
if cfg.knowledge_base_id:
hints.append(
@@ -354,8 +356,17 @@ async def run_pipeline(
top_n=knowledge_top_n,
score_threshold=knowledge_score_threshold,
)
vision_native_mode = vision_enabled and _vision_uses_main_llm(cfg)
vision_native_mode = (
vision_enabled
and cfg.type != "workflow"
and _vision_uses_main_llm(cfg)
)
vision_state: dict[str, str | None] = {"client_id": None}
workflow_vision_scope: dict[str, Any] = {
"enabled": False,
"vision_model_resource_id": None,
"llm_resource_id": None,
}
vision_schema = FunctionSchema(
name=VISION_TOOL_NAME,
description=(
@@ -458,8 +469,14 @@ async def run_pipeline(
)
flow_global_functions = []
workflow_vision_function = None
if cfg.type == "workflow" and vision_enabled:
async def flow_fetch_user_image(args, _flow_manager):
if not workflow_vision_scope.get("enabled"):
return {
"status": "disabled",
"message": "当前 Agent 节点未开启视觉理解。",
}
question = str((args or {}).get("question") or "请描述当前画面。")
user_id = vision_state.get("client_id")
if not user_id:
@@ -474,8 +491,26 @@ async def run_pipeline(
function_name=VISION_TOOL_NAME,
)
try:
vision_resource_id = str(
workflow_vision_scope.get("vision_model_resource_id") or ""
)
llm_resource_id = str(
workflow_vision_scope.get("llm_resource_id") or ""
)
resource_id = vision_resource_id or llm_resource_id
resource = cfg.workflow_model_resources.get(resource_id)
if resource:
vision_cfg = config_with_vision_resource(cfg, resource)
elif not vision_resource_id:
vision_cfg = config_with_main_llm_as_vision(cfg)
else:
raise ValueError(f"视觉模型资源未加载:{vision_resource_id}")
frame = await vision_capture.request_image(llm, request)
observation = await _analyze_image_with_vision_model(cfg, frame, question)
observation = await _analyze_image_with_vision_model(
vision_cfg,
frame,
question,
)
return {
"status": "ok",
"question": question,
@@ -487,15 +522,13 @@ async def run_pipeline(
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,
)
workflow_vision_function = 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":
@@ -612,6 +645,8 @@ async def run_pipeline(
transport=transport,
switch_services=service_controller.switch,
set_knowledge_scope=knowledge_retrieval.set_scope,
set_vision_scope=lambda scope: workflow_vision_scope.update(scope),
vision_function=workflow_vision_function,
set_input_enabled=lambda enabled: input_state.__setitem__("enabled", enabled),
apply_turn_config=apply_workflow_turn_config,
flow_global_functions=flow_global_functions,

58
backend/services/vision.py Normal file
View File

@@ -0,0 +1,58 @@
"""Shared vision capability metadata and Workflow model resolution."""
from __future__ import annotations
from models import AssistantConfig, RuntimeModelResource
VISION_TOOL_NAME = "fetch_user_image"
VISION_SYSTEM_HINT = (
"当前阶段打开了视觉理解。用户询问当前画面、摄像头里有什么、人物/物品/"
"环境状态或需要你看一眼时，调用 fetch_user_image 获取当前视频帧，"
"再基于画面回答。"
)
VISION_ANALYSIS_SYSTEM_PROMPT = (
"你是一个视觉理解模型。请只根据图片内容和用户问题给出准确、简洁的中文观察结果。"
"如果画面不足以判断，请明确说明不确定。"
)
def config_with_vision_resource(
cfg: AssistantConfig,
resource: RuntimeModelResource,
) -> AssistantConfig:
"""Create the config view used to analyze one Workflow camera frame."""
if resource.capability != "LLM":
raise ValueError(f"视觉模型资源能力无效:{resource.capability}")
if not resource.support_image_input:
raise ValueError(f"视觉模型不支持图片输入:{resource.id}")
result = cfg.model_copy(deep=True)
values = resource.values or {}
secrets = resource.secrets or {}
result.vision_model_resource_id = resource.id
result.vision_model = str(values.get("modelId") or "")
result.vision_llm_interface_type = resource.interface_type
result.vision_llm_values = values
result.vision_llm_secrets = secrets
result.vision_llm_support_image_input = resource.support_image_input
result.vision_llm_api_key = str(secrets.get("apiKey") or "")
result.vision_llm_base_url = str(values.get("apiUrl") or "")
return result
def config_with_main_llm_as_vision(cfg: AssistantConfig) -> AssistantConfig:
"""Legacy fallback when a Workflow LLM is not present in the resource map."""
if not cfg.llm_support_image_input:
raise ValueError("当前大语言模型不支持图片输入")
result = cfg.model_copy(deep=True)
result.vision_model_resource_id = None
result.vision_model = cfg.model
result.vision_llm_interface_type = cfg.llm_interface_type
result.vision_llm_values = dict(cfg.llm_values)
result.vision_llm_secrets = dict(cfg.llm_secrets)
result.vision_llm_support_image_input = cfg.llm_support_image_input
result.vision_llm_api_key = cfg.llm_api_key
result.vision_llm_base_url = cfg.llm_base_url
return result

12
backend/services/workflow/agent.py
View File

@@ -11,6 +11,7 @@ from pipecat.services.settings import LLMSettings
from services.brains.base import BrainRuntime
from services.runtime_variables import DynamicVariableStore
from services.vision import VISION_SYSTEM_HINT
from services.workflow_engine import WorkflowEngine
from services.workflow_router import WorkflowLLMRouter
@@ -39,6 +40,9 @@ class WorkflowAgentStage:
def role_message(self, node_id: str) -> str:
stage_prompt = self._engine.prompt_for(node_id, self._store)
stage = self._engine.agent_stage_config(node_id)
if stage.vision_enabled:
stage_prompt = f"{stage_prompt}\n\n[视觉能力]\n{VISION_SYSTEM_HINT}"
return (
f"{stage_prompt}\n\n[工作流执行规则]\n"
f"{AGENT_STAGE_INSTRUCTION}"
@@ -90,6 +94,14 @@ class WorkflowAgentStage:
"score_threshold": stage.knowledge_score_threshold,
}
)
if self._runtime.set_vision_scope:
self._runtime.set_vision_scope(
{
"enabled": stage.vision_enabled,
"vision_model_resource_id": stage.vision_model_resource_id,
"llm_resource_id": stage.llm_resource_id,
}
)
def node_config(
self,

14
backend/services/workflow_engine.py
View File

@@ -18,6 +18,8 @@ class AgentStageConfig:
llm_resource_id: str
asr_resource_id: str
tts_resource_id: str
vision_enabled: bool
vision_model_resource_id: str | None
tool_ids: tuple[str, ...]
knowledge_base_id: str | None
knowledge_mode: str
@@ -119,6 +121,10 @@ class WorkflowEngine:
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 ""),
vision_enabled=bool(source.get("visionEnabled", False)),
vision_model_resource_id=(
str(source.get("visionModelResourceId") or "") or None
),
tool_ids=tuple(str(tool_id) for tool_id in source.get("toolIds") or []),
knowledge_base_id=knowledge_base_id or None,
knowledge_mode=(
@@ -139,6 +145,14 @@ class WorkflowEngine:
turn_config=dict(turn_config),
)
def uses_vision(self) -> bool:
"""Return whether any effective Agent stage needs a camera stream."""
return any(
self.agent_stage_config(node_id).vision_enabled
for node_id, node in self.nodes.items()
if node.get("type") == "agent"
)
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())

75
backend/tests/test_brains.py
View File

@@ -470,6 +470,81 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
async def test_agent_vision_tool_is_scoped_to_effective_stage(self):
brain = WorkflowBrain(
{
"specVersion": 3,
"settings": {
"globalPrompt": "全局规则",
"defaultLlmResourceId": "llm_global",
"visionEnabled": True,
"visionModelResourceId": "vision_global",
},
"nodes": [
{"id": "start", "type": "start", "data": {}},
{
"id": "agent",
"type": "agent",
"data": {
"prompt": "观察用户需要展示的物品",
"inheritGlobalConfig": True,
},
},
],
"edges": [
{
"id": "begin",
"source": "start",
"target": "agent",
"data": {"mode": "always", "priority": 0},
}
],
}
)
scopes = []
vision_function = object()
async def queue_frame(_frame):
pass
brain._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(),
set_vision_scope=scopes.append,
vision_function=vision_function,
)
await brain._apply_agent_stage("agent")
inherited_config = brain._agent_config("agent")
self.assertIn(vision_function, inherited_config["functions"])
self.assertIn("fetch_user_image", inherited_config["role_message"])
self.assertEqual(
scopes[-1],
{
"enabled": True,
"vision_model_resource_id": "vision_global",
"llm_resource_id": "llm_global",
},
)
brain._engine.data("agent").update(
{
"inheritGlobalConfig": False,
"llmResourceId": "llm_agent",
"visionEnabled": False,
"visionModelResourceId": "",
}
)
await brain._apply_agent_stage("agent")
custom_config = brain._agent_config("agent")
self.assertNotIn(vision_function, custom_config["functions"])
self.assertNotIn("fetch_user_image", custom_config["role_message"])
self.assertFalse(scopes[-1]["enabled"])
async def test_initial_fixed_speech_waits_for_start_greeting_to_finish(self):
brain = WorkflowBrain(
{

108
backend/tests/test_workflow_v3.py
View File

@@ -1,11 +1,16 @@
from __future__ import annotations
import unittest
from types import SimpleNamespace
from models import AssistantConfig, RuntimeModelResource
from fastapi import HTTPException
from routes.assistants import _validate_workflow, _validate_workflow_references
from schemas import AssistantUpsert
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.vision import config_with_vision_resource
from services.workflow_engine import WorkflowEngine
@@ -49,6 +54,28 @@ def valid_graph():
class WorkflowGraphTests(unittest.TestCase):
def test_workflow_graph_owns_flat_vision_session_flag(self):
graph = valid_graph()
graph["settings"].update(
{
"defaultLlmResourceId": "llm_global",
"visionEnabled": True,
"visionModelResourceId": "vision_global",
}
)
body = AssistantUpsert(
name="视觉工作流",
type="workflow",
graph=graph,
visionEnabled=False,
visionModelResourceId="legacy_vision",
)
_validate_workflow(body)
self.assertTrue(body.vision_enabled)
self.assertIsNone(body.vision_model_resource_id)
def test_agent_entry_mode_defaults_and_validation(self):
graph = valid_graph()
normalized = normalize_graph(graph)
@@ -97,6 +124,8 @@ class WorkflowGraphTests(unittest.TestCase):
"defaultLlmResourceId": "llm_global",
"defaultAsrResourceId": "asr_global",
"defaultTtsResourceId": "tts_global",
"visionEnabled": True,
"visionModelResourceId": "vision_global",
"toolIds": ["tool_global"],
"knowledgeBaseId": "kb_global",
}
@@ -108,6 +137,8 @@ class WorkflowGraphTests(unittest.TestCase):
"llmResourceId": "llm_agent",
"asrResourceId": "asr_agent",
"ttsResourceId": "tts_agent",
"visionEnabled": True,
"visionModelResourceId": "vision_agent",
"toolIds": ["tool_agent"],
"knowledgeBaseId": "kb_agent",
}
@@ -120,9 +151,11 @@ class WorkflowGraphTests(unittest.TestCase):
"llm_global",
"asr_global",
"tts_global",
"vision_global",
"llm_agent",
"asr_agent",
"tts_agent",
"vision_agent",
},
)
self.assertEqual(refs["tools"], {"tool_global", "tool_agent"})
@@ -165,6 +198,8 @@ class WorkflowGraphTests(unittest.TestCase):
"defaultLlmResourceId": "llm_global",
"defaultAsrResourceId": "asr_global",
"defaultTtsResourceId": "tts_global",
"visionEnabled": True,
"visionModelResourceId": "vision_global",
"toolIds": ["tool_global"],
"knowledgeBaseId": "kb_global",
"knowledgeMode": "on_demand",
@@ -180,6 +215,12 @@ class WorkflowGraphTests(unittest.TestCase):
engine = WorkflowEngine(graph)
inherited = engine.agent_stage_config("agent")
self.assertEqual(inherited.llm_resource_id, "llm_global")
self.assertTrue(inherited.vision_enabled)
self.assertEqual(
inherited.vision_model_resource_id,
"vision_global",
)
self.assertTrue(engine.uses_vision())
self.assertEqual(inherited.tool_ids, ("tool_global",))
self.assertEqual(inherited.knowledge_mode, "on_demand")
self.assertFalse(inherited.enable_interrupt)
@@ -194,6 +235,8 @@ class WorkflowGraphTests(unittest.TestCase):
"llmResourceId": "llm_agent",
"toolIds": ["tool_agent"],
"knowledgeBaseId": "",
"visionEnabled": False,
"visionModelResourceId": "",
"enableInterrupt": True,
"turnConfig": {
"bargeIn": {"strategy": "vad"},
@@ -203,6 +246,9 @@ class WorkflowGraphTests(unittest.TestCase):
)
custom = engine.agent_stage_config("agent")
self.assertEqual(custom.llm_resource_id, "llm_agent")
self.assertFalse(custom.vision_enabled)
self.assertIsNone(custom.vision_model_resource_id)
self.assertFalse(engine.uses_vision())
self.assertEqual(custom.tool_ids, ("tool_agent",))
self.assertEqual(custom.knowledge_mode, "disabled")
self.assertTrue(custom.enable_interrupt)
@@ -211,6 +257,27 @@ class WorkflowGraphTests(unittest.TestCase):
"smart_turn",
)
def test_vision_resource_creates_isolated_runtime_config(self):
base = AssistantConfig(type="workflow", model="text-only")
resource = RuntimeModelResource(
id="vision_1",
capability="LLM",
interface_type="openai-llm",
values={
"modelId": "vision-model",
"apiUrl": "https://vision.test/v1",
},
secrets={"apiKey": "vision-secret"},
support_image_input=True,
)
resolved = config_with_vision_resource(base, resource)
self.assertEqual(resolved.vision_model, "vision-model")
self.assertEqual(resolved.vision_llm_api_key, "vision-secret")
self.assertEqual(resolved.vision_llm_base_url, "https://vision.test/v1")
self.assertEqual(base.model, "text-only")
def test_start_agent_action_and_handoff_may_have_no_outgoing_edge(self):
terminal_graphs = [
{
@@ -518,5 +585,46 @@ class WorkflowGraphTests(unittest.TestCase):
)
class WorkflowVisionReferenceTests(unittest.IsolatedAsyncioTestCase):
async def test_visual_model_must_support_image_input(self):
graph = valid_graph()
graph["settings"].update(
{
"defaultLlmResourceId": "llm_global",
"visionEnabled": True,
"visionModelResourceId": "vision_global",
}
)
body = AssistantUpsert(
name="视觉工作流",
type="workflow",
graph=graph,
)
_validate_workflow(body)
resources = {
"llm_global": SimpleNamespace(
enabled=True,
capability="LLM",
support_image_input=False,
),
"vision_global": SimpleNamespace(
enabled=True,
capability="LLM",
support_image_input=False,
),
}
class FakeSession:
async def get(self, _model, resource_id):
return resources.get(resource_id)
with self.assertRaisesRegex(HTTPException, "必须支持图片输入"):
await _validate_workflow_references(FakeSession(), body)
resources["vision_global"].support_image_input = True
await _validate_workflow_references(FakeSession(), body)
if __name__ == "__main__":
unittest.main()