From b49f5d3ba5d7d9dfec510c15876908357e70111d Mon Sep 17 00:00:00 2001 From: Curry Date: Sat, 18 Jul 2026 20:04:38 +0800 Subject: [PATCH] feat(workflow): JoyAI-VL-Interaction tool capability node New tool node tool.realtime.vl.interaction bridging the workflow runtime to a self-hosted JoyAI-VL-Interaction deployment (jd-opensource, 8B realtime video-language interaction model, Apache-2.0, served OpenAI-compatible via vLLM-Omni). - backend/workflow/joyai_vl_executor.py: executor mode joyai_vl_interaction, one chat-completions interaction turn (prompt + videoUrl/imageUrls) against JOYAI_VL_URL, reply emitted as event.vl.interaction.v1 - tool_capabilities registry entry (event[] in / event[] out, runnable) - opencli_hda_tracer dispatch mirroring the okx executor precedent; failures (incl. unconfigured endpoint) degrade to event.vl.interaction.error.v1 instead of crashing the run - SUPPORTED_TOOL_EXECUTOR_MODES + WorkflowToolCapabilityExecutor.mode literal extended; .env.example documents JOYAI_VL_URL / JOYAI_VL_API_KEY - 4 unit tests: unconfigured guard, request shape + reply parse, empty choices, registry resolution Full-suite note: 6 order-pollution flakes pre-exist on clean main (baselined this session, logged in docs/BUG-TRIAGE-20260718.md P3-6); branch adds 4 passing tests, zero new failures. --- .env.example | 6 + backend/schemas/workflow.py | 2 +- backend/workflow/joyai_vl_executor.py | 148 +++++++++++++++++++++++++ backend/workflow/opencli_hda_tracer.py | 30 +++++ backend/workflow/runtime_registry.py | 2 +- backend/workflow/tool_capabilities.py | 28 +++++ docs/BUG-TRIAGE-20260718.md | 8 ++ tests/unit/test_joyai_vl_executor.py | 101 +++++++++++++++++ 8 files changed, 323 insertions(+), 2 deletions(-) create mode 100644 backend/workflow/joyai_vl_executor.py create mode 100644 tests/unit/test_joyai_vl_executor.py diff --git a/.env.example b/.env.example index aee1100..911fed1 100644 --- a/.env.example +++ b/.env.example @@ -58,6 +58,12 @@ DATABASE_URL=sqlite+aiosqlite:///./opencli_admin.db # docker-compose --profile celery up TASK_EXECUTOR=local +# JoyAI-VL-Interaction 工具节点（tool.realtime.vl.interaction） +# 自托管 vLLM-Omni 服务地址（jd-opensource/JoyAI-VL-Interaction, 8B, Apache-2.0）。 +# 留空 = 节点执行降级为 error 事件, 不崩 run。可选 JOYAI_VL_API_KEY 加 Bearer。 +# JOYAI_VL_URL=http://joyai-vl:8000 +# JOYAI_VL_API_KEY= + # 采集编排（NAS 推荐 iii） # admin — API 内置 scheduler.py 驱动定时采集（默认） # iii — III engine + schedule-bootstrap 驱动 cron；见 .env.nas.example diff --git a/backend/schemas/workflow.py b/backend/schemas/workflow.py index 5c7ce8d..3886e0a 100644 --- a/backend/schemas/workflow.py +++ b/backend/schemas/workflow.py @@ -390,7 +390,7 @@ class WorkflowToolCapabilityPort(BaseModel): class WorkflowToolCapabilityExecutor(BaseModel): - mode: Literal["fixture", "okx_market_ticker_snapshot"] + mode: Literal["fixture", "okx_market_ticker_snapshot", "joyai_vl_interaction"] description: Optional[str] = None params: dict[str, Any] = Field(default_factory=dict) diff --git a/backend/workflow/joyai_vl_executor.py b/backend/workflow/joyai_vl_executor.py new file mode 100644 index 0000000..5a2a9bc --- /dev/null +++ b/backend/workflow/joyai_vl_executor.py @@ -0,0 +1,148 @@ +"""JoyAI-VL-Interaction executor for OpenCLI Tool Capabilities. + +Bridges the workflow runtime to a self-hosted JoyAI-VL-Interaction deployment +(https://github.com/jd-opensource/JoyAI-VL-Interaction) — JD's 8B real-time +video-language interaction model, served OpenAI-compatible via vLLM-Omni. + +MVP scope: one synchronous interaction probe per tool call — send the node's +prompt plus media references (video URL / image frames) to the model's +chat-completions endpoint and emit the reply as an event.v1-style payload. +The always-on streaming mode (model watches a live feed and speaks up +unprompted) needs a persistent session and lands later on the same executor. +""" + +from __future__ import annotations + +import json +import os +import time +import urllib.request +from datetime import UTC, datetime +from typing import Any + +JOYAI_VL_INTERACTION_EXECUTOR = "joyai_vl_interaction" +JOYAI_VL_TOOL_CAPABILITY_ID = "tool.realtime.vl.interaction" +JOYAI_VL_DEFAULT_MODEL = "JoyAI-VL-Interaction-Preview" +JOYAI_VL_ENDPOINT_ENV = "JOYAI_VL_URL" + + +class JoyAIVLExecutionError(RuntimeError): + """Raised when the JoyAI-VL interaction executor cannot produce a reply.""" + + +def execute_joyai_vl_interaction(params: dict[str, Any]) -> dict[str, Any]: + """Run one vision-language interaction turn against a JoyAI-VL deployment.""" + + endpoint = ( + _read_string(params.get("endpoint")) + or _read_string(params.get("endpointUrl")) + or _read_string(os.environ.get(JOYAI_VL_ENDPOINT_ENV)) + ) + if not endpoint: + raise JoyAIVLExecutionError( + f"JoyAI-VL endpoint is not configured: set {JOYAI_VL_ENDPOINT_ENV} " + "(vLLM-Omni base URL, e.g. http://joyai-vl:8000) or pass params.endpoint" + ) + + model = _read_string(params.get("model")) or JOYAI_VL_DEFAULT_MODEL + prompt = _read_string(params.get("prompt")) or "描述当前画面正在发生什么, 如有需要人工注意的事件请指出." + video_url = _read_string(params.get("videoUrl")) or _read_string(params.get("video_url")) + image_urls = _read_string_list(params.get("imageUrls")) or _read_string_list( + params.get("image_urls") + ) + timeout_seconds = _read_timeout(params.get("timeoutSeconds")) + + content: list[dict[str, Any]] = [] + if video_url: + content.append({"type": "video_url", "video_url": {"url": video_url}}) + for url in image_urls: + content.append({"type": "image_url", "image_url": {"url": url}}) + content.append({"type": "text", "text": prompt}) + + request_body = { + "model": model, + "messages": [{"role": "user", "content": content}], + "stream": False, + } + max_tokens = _read_int(params.get("maxTokens")) + if max_tokens: + request_body["max_tokens"] = max_tokens + + url = endpoint.rstrip("/") + "/v1/chat/completions" + opened_at = time.time() + request = urllib.request.Request( + url, + data=json.dumps(request_body).encode("utf-8"), + headers={ + "Content-Type": "application/json", + "User-Agent": "Mozilla/5.0 OpenCLI-Admin-joyai-vl-executor/0.1", + **_auth_header(params), + }, + method="POST", + ) + + try: + with urllib.request.urlopen(request, timeout=timeout_seconds) as response: + payload = json.loads(response.read().decode("utf-8")) + except JoyAIVLExecutionError: + raise + except Exception as exc: # pragma: no cover - exercised by live smoke failures. + raise JoyAIVLExecutionError(f"JoyAI-VL request failed: {exc}") from exc + + choices = payload.get("choices") or [] + if not choices: + raise JoyAIVLExecutionError(f"JoyAI-VL returned no choices: {payload}") + message = choices[0].get("message") or {} + reply = message.get("content") + if not isinstance(reply, str) or not reply.strip(): + raise JoyAIVLExecutionError(f"JoyAI-VL returned an empty reply: {payload}") + + duration_ms = round((time.time() - opened_at) * 1000) + return { + "schema": "event.vl.interaction.v1", + "source": "joyai-vl", + "eventType": "vl.interaction", + "observedAt": datetime.now(tz=UTC).isoformat(), + "model": payload.get("model") or model, + "reply": reply, + "media": { + "videoUrl": video_url, + "imageCount": len(image_urls), + }, + "request": { + "url": url, + "prompt": prompt, + "durationMs": duration_ms, + }, + "usage": payload.get("usage") or {}, + } + + +def _auth_header(params: dict[str, Any]) -> dict[str, str]: + api_key = _read_string(params.get("apiKey")) or _read_string( + os.environ.get("JOYAI_VL_API_KEY") + ) + return {"Authorization": f"Bearer {api_key}"} if api_key else {} + + +def _read_string(value: object) -> str | None: + return value.strip() if isinstance(value, str) and value.strip() else None + + +def _read_string_list(value: object) -> list[str]: + if not isinstance(value, list): + return [] + return [item.strip() for item in value if isinstance(item, str) and item.strip()] + + +def _read_int(value: object) -> int | None: + try: + return int(value) if value is not None else None + except (TypeError, ValueError): + return None + + +def _read_timeout(value: object) -> float: + if isinstance(value, int | float) and value > 0: + return float(value) + return 30.0 diff --git a/backend/workflow/opencli_hda_tracer.py b/backend/workflow/opencli_hda_tracer.py index 889fda8..998b6af 100644 --- a/backend/workflow/opencli_hda_tracer.py +++ b/backend/workflow/opencli_hda_tracer.py @@ -46,6 +46,12 @@ from backend.workflow.compiler import INTERNAL_ID_SEPARATOR, compile_workflow_project from backend.workflow.event_mirror import publish_workflow_run_event_mirror from backend.workflow.fleet_inventory import match_workflow_fleet_capability +from backend.workflow.joyai_vl_executor import ( + JOYAI_VL_INTERACTION_EXECUTOR, + JOYAI_VL_TOOL_CAPABILITY_ID, + JoyAIVLExecutionError, + execute_joyai_vl_interaction, +) from backend.workflow.realtime_market_executor import ( OKX_MARKET_TICKER_SNAPSHOT_EXECUTOR, RealtimeMarketExecutionError, @@ -1677,6 +1683,13 @@ def _execute_external_tool_capability( output = _execute_okx_market_tool(binding_input) return [_external_tool_output(node, output, input_items, run_id, 0, binding_input)] + if ( + binding_input.get("executorMode") == JOYAI_VL_INTERACTION_EXECUTOR + and binding_input.get("toolCapabilityId") == JOYAI_VL_TOOL_CAPABILITY_ID + ): + output = _execute_joyai_vl_tool(binding_input) + return [_external_tool_output(node, output, input_items, run_id, 0, binding_input)] + fixture_outputs = _read_dict_list(binding_input.get("fixtureOutputs")) fixture_output = _read_dict(binding_input.get("fixtureOutput")) if not fixture_outputs and fixture_output: @@ -1707,6 +1720,23 @@ def _execute_okx_market_tool(binding_input: dict[str, Any]) -> dict[str, Any]: } +def _execute_joyai_vl_tool(binding_input: dict[str, Any]) -> dict[str, Any]: + params = { + **_read_dict(binding_input.get("executorParams")), + **_read_dict(binding_input.get("toolParams")), + } + try: + return execute_joyai_vl_interaction(params) + except JoyAIVLExecutionError as exc: + return { + "schema": "event.vl.interaction.error.v1", + "source": "joyai-vl", + "eventType": "vl.interaction.error", + "status": "error", + "message": str(exc), + } + + def _external_tool_output( node: CompiledWorkflowNode, output: dict[str, Any], diff --git a/backend/workflow/runtime_registry.py b/backend/workflow/runtime_registry.py index 4dd735b..b9921d4 100644 --- a/backend/workflow/runtime_registry.py +++ b/backend/workflow/runtime_registry.py @@ -51,7 +51,7 @@ WEBHOOK_NOTIFY_BINDING_ID = "workflow.notifier.webhook.send" NOTIFY_SEND_BINDING_ID = "workflow.notify.send" EXTERNAL_TOOL_BINDING_ID = "workflow.external-tool.capability" -SUPPORTED_TOOL_EXECUTOR_MODES = {"fixture", "okx_market_ticker_snapshot"} +SUPPORTED_TOOL_EXECUTOR_MODES = {"fixture", "okx_market_ticker_snapshot", "joyai_vl_interaction"} class WorkflowRuntimeBinding(BaseModel): diff --git a/backend/workflow/tool_capabilities.py b/backend/workflow/tool_capabilities.py index 3a95f26..333b9ca 100644 --- a/backend/workflow/tool_capabilities.py +++ b/backend/workflow/tool_capabilities.py @@ -8,6 +8,10 @@ WorkflowToolCapabilityExecutor, WorkflowToolCapabilityPort, ) +from backend.workflow.joyai_vl_executor import ( + JOYAI_VL_INTERACTION_EXECUTOR, + JOYAI_VL_TOOL_CAPABILITY_ID, +) from backend.workflow.realtime_market_executor import OKX_MARKET_TICKER_SNAPSHOT_EXECUTOR @@ -141,6 +145,30 @@ def _tool_capabilities() -> list[WorkflowToolCapability]: schema="tool-capability.realtime-signal-emit.v1", resources=["signal_policy", "run_trace"], ), + _realtime_tool( + id=JOYAI_VL_TOOL_CAPABILITY_ID, + label="JoyAI VL Interaction", + description=( + "Vision-language interaction over video/image media via a " + "self-hosted JoyAI-VL-Interaction deployment (JD open source, " + "8B, vLLM-Omni serving). Sends the node prompt plus media " + "references, emits the model's interaction reply as event[]." + ), + input_type="event[]", + output_type="event[]", + tags=["tool", "realtime", "vision", "vl", "interaction", "joyai"], + schema="tool-capability.vl-interaction.v1", + resources=["media_reference", "run_trace"], + executor=WorkflowToolCapabilityExecutor( + mode=JOYAI_VL_INTERACTION_EXECUTOR, + description=( + "POSTs one chat-completions interaction turn to the " + "JOYAI_VL_URL vLLM-Omni endpoint; unconfigured endpoint " + "degrades to an error event, never a run crash." + ), + params={"model": "JoyAI-VL-Interaction-Preview"}, + ), + ), ] diff --git a/docs/BUG-TRIAGE-20260718.md b/docs/BUG-TRIAGE-20260718.md index 0372c87..7901ab7 100644 --- a/docs/BUG-TRIAGE-20260718.md +++ b/docs/BUG-TRIAGE-20260718.md @@ -51,3 +51,11 @@ cd D:\projects\opencli-admin $env:API_AUTH_TOKEN=''; $env:AGENT_API_TOKEN=''; uv run --extra dev pytest tests/unit tests/integration -q --no-cov ``` + +## P3 补录 (2026-07-18 晚) + +### 6. 全套跑测试顺序污染 flake (main 存量) +- `pytest tests/integration tests/unit tests/skills` 全套跑稳定挂 6 个, 单跑全过; clean main 与 feature 分支同样 6 个 → 存量顺序依赖, 非新代码 +- 名单: test_workflow_opencli_hda_trace_api (projection/stream ×2), test_workflow_turbopush_publish_api (capabilities ×1), test_nodes_install_script (netbird/ssh ×2), +1 +- 疑似共享进程级状态 (capability registry / settings snapshot / os.environ 写入未回滚) +- 修向: 定位先污染后断言的 test 对 (pytest -p no:randomly 二分), 或对相关 fixture 加隔离 diff --git a/tests/unit/test_joyai_vl_executor.py b/tests/unit/test_joyai_vl_executor.py new file mode 100644 index 0000000..484b254 --- /dev/null +++ b/tests/unit/test_joyai_vl_executor.py @@ -0,0 +1,101 @@ +"""JoyAI-VL interaction tool executor: config guard, request shape, reply parsing.""" + +import io +import json + +import pytest + +from backend.workflow.joyai_vl_executor import ( + JOYAI_VL_ENDPOINT_ENV, + JOYAI_VL_TOOL_CAPABILITY_ID, + JoyAIVLExecutionError, + execute_joyai_vl_interaction, +) +from backend.workflow.tool_capabilities import ( + list_workflow_tool_capabilities, + resolve_workflow_tool_capability, +) + + +def test_unconfigured_endpoint_raises_actionable_error(monkeypatch): + monkeypatch.delenv(JOYAI_VL_ENDPOINT_ENV, raising=False) + with pytest.raises(JoyAIVLExecutionError, match=JOYAI_VL_ENDPOINT_ENV): + execute_joyai_vl_interaction({}) + + +def test_success_turn_builds_openai_media_request_and_parses_reply(monkeypatch): + captured: dict = {} + + class _FakeResponse(io.BytesIO): + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def fake_urlopen(request, timeout): + captured["url"] = request.full_url + captured["timeout"] = timeout + captured["body"] = json.loads(request.data.decode("utf-8")) + return _FakeResponse( + json.dumps( + { + "model": "JoyAI-VL-Interaction-Preview", + "choices": [{"message": {"role": "assistant", "content": "锅要溢出来了"}}], + "usage": {"total_tokens": 42}, + } + ).encode("utf-8") + ) + + monkeypatch.setattr("urllib.request.urlopen", fake_urlopen) + + output = execute_joyai_vl_interaction( + { + "endpoint": "http://joyai-vl:8000/", + "prompt": "看着炉子", + "videoUrl": "http://media/stove.mp4", + "imageUrls": ["http://media/frame1.jpg"], + "timeoutSeconds": 5, + } + ) + + assert captured["url"] == "http://joyai-vl:8000/v1/chat/completions" + assert captured["timeout"] == 5.0 + content = captured["body"]["messages"][0]["content"] + assert {"type": "video_url", "video_url": {"url": "http://media/stove.mp4"}} in content + assert {"type": "image_url", "image_url": {"url": "http://media/frame1.jpg"}} in content + assert content[-1] == {"type": "text", "text": "看着炉子"} + + assert output["schema"] == "event.vl.interaction.v1" + assert output["source"] == "joyai-vl" + assert output["reply"] == "锅要溢出来了" + assert output["media"] == {"videoUrl": "http://media/stove.mp4", "imageCount": 1} + assert output["usage"] == {"total_tokens": 42} + + +def test_empty_choices_raises(monkeypatch): + class _FakeResponse(io.BytesIO): + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + monkeypatch.setattr( + "urllib.request.urlopen", + lambda request, timeout: _FakeResponse(json.dumps({"choices": []}).encode("utf-8")), + ) + with pytest.raises(JoyAIVLExecutionError, match="no choices"): + execute_joyai_vl_interaction({"endpoint": "http://joyai-vl:8000"}) + + +def test_tool_capability_is_registered_and_resolvable(): + tool = resolve_workflow_tool_capability(JOYAI_VL_TOOL_CAPABILITY_ID) + assert tool is not None + assert tool.executor.mode == "joyai_vl_interaction" + assert tool.status == "runnable" + assert [port.type for port in tool.inputPorts] == ["event[]"] + assert [port.type for port in tool.outputPorts] == ["event[]"] + assert JOYAI_VL_TOOL_CAPABILITY_ID in { + t.id for t in list_workflow_tool_capabilities().tools + }