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

feat: add realtime workflow vision tools

Browse Source
This commit is contained in:
Xin Wang
2026-08-05 13:35:00 +08:00
parent 11374238f3
commit 10bea48d72
16 changed files with 362 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -7,14 +7,10 @@
"""
import asyncio
import base64
from io import BytesIO
from typing import Any
from loguru import logger
from models import AssistantConfig
from openai import AsyncOpenAI
from PIL import Image
from services.brains import Brain, BrainRuntime, build_brain
from services.brains.base import RealtimeBrainRuntime
from services.conversation_history import ConversationRecorder
@@ -34,11 +30,12 @@ from services.knowledge import search as search_knowledge
from services.client_tools import ClientToolBroker
from services.tool_policy import policy_for_tool
from services.vision import (
VISION_ANALYSIS_SYSTEM_PROMPT,
VISION_SYSTEM_HINT,
VISION_TOOL_NAME,
analyze_image_with_vision_model,
config_with_main_llm_as_vision,
config_with_vision_resource,
image_data_uri,
)
from services.workflow_engine import WorkflowEngine
@@ -164,19 +161,6 @@ def _workflow_vision_uses_main_llm(
return True
def _image_data_uri(frame: UserImageRawFrame) -> str:
if not frame.format:
raise ValueError("摄像头图片帧缺少 format,无法编码给视觉模型")
buffer = BytesIO()
Image.frombytes(frame.format, frame.size, frame.image).save(
buffer,
format="JPEG",
quality=85,
)
encoded = base64.b64encode(buffer.getvalue()).decode("utf-8")
return f"data:image/jpeg;base64,{encoded}"
def _multimodal_user_input_frame(
image_frame: UserImageRawFrame,
prompt_text: str,
@@ -197,7 +181,7 @@ def _multimodal_user_input_frame(
{"type": "text", "text": prompt_text},
{
"type": "image_url",
"image_url": {"url": _image_data_uri(image_frame)},
"image_url": {"url": image_data_uri(image_frame)},
},
],
}
@@ -206,45 +190,6 @@ def _multimodal_user_input_frame(
)
async def _analyze_image_with_vision_model(
cfg: AssistantConfig,
frame: UserImageRawFrame,
question: str,
) -> str:
if cfg.vision_llm_interface_type not in {"openai-llm", "dashscope-llm"}:
raise ValueError(f"不支持的视觉 LLM 接口类型: {cfg.vision_llm_interface_type}")
data_uri = await asyncio.to_thread(_image_data_uri, frame)
extra_body = cfg.vision_llm_values.get("extraBody")
extra = {"extra_body": extra_body} if isinstance(extra_body, dict) else {}
client = AsyncOpenAI(
api_key=_require(cfg.vision_llm_api_key, "Vision LLM apiKey"),
base_url=_require(cfg.vision_llm_base_url, "Vision LLM apiUrl"),
)
try:
response = await client.chat.completions.create(
model=_require(cfg.vision_model, "Vision LLM modelId"),
messages=[
{"role": "system", "content": VISION_ANALYSIS_SYSTEM_PROMPT},
{
"role": "user",
"content": [
{"type": "text", "text": question},
{"type": "image_url", "image_url": {"url": data_uri}},
],
},
],
**extra,
)
finally:
await client.close()
content = response.choices[0].message.content if response.choices else ""
if isinstance(content, str):
return content.strip()
return str(content or "").strip()
async def run_pipeline(
transport,
cfg: AssistantConfig,
@@ -278,12 +223,11 @@ async def run_pipeline(
raise ValueError(f"类型 {cfg.type} 不支持 realtime 运行模式")
if cfg.runtimeMode == "realtime":
if vision_enabled:
logger.warning("Realtime 模式暂未接入视频帧工具,本次仅启用语音通话")
await run_realtime_pipeline(
transport,
cfg,
brain=brain,
vision_enabled=vision_enabled,
assistant_id=assistant_id,
channel=channel,
)
@@ -524,7 +468,7 @@ async def run_pipeline(
)
try:
frame = await vision_capture.request_image(params.llm, request)
observation = await _analyze_image_with_vision_model(cfg, frame, question)
observation = await analyze_image_with_vision_model(cfg, frame, question)
except asyncio.TimeoutError:
await params.result_callback(
{
@@ -614,7 +558,7 @@ async def run_pipeline(
)
try:
frame = await vision_capture.request_image(llm, request)
observation = await _analyze_image_with_vision_model(
observation = await analyze_image_with_vision_model(
active_vision_config(),
frame,
question,
@@ -830,7 +774,7 @@ async def run_pipeline(
try:
assert analysis_cfg is not None
observation = await _analyze_image_with_vision_model(
observation = await analyze_image_with_vision_model(
analysis_cfg,
image_frame,
value.prompt_text,
@@ -883,6 +827,7 @@ async def run_realtime_pipeline(
cfg: AssistantConfig,
*,
brain: Brain,
vision_enabled: bool = False,
assistant_id: str | None = None,
channel: str = "webrtc",
) -> None:
@@ -894,6 +839,8 @@ async def run_realtime_pipeline(
input_sample_rate, output_sample_rate = realtime_audio_sample_rates(cfg)
worker_holder: dict[str, PipelineWorker] = {}
input_state = {"enabled": True}
vision_state: dict[str, str | None] = {"client_id": None}
vision_capture = VisionCaptureProcessor()
async def queue_call_end(reason: str) -> None:
worker = worker_holder.get("worker")
@@ -946,6 +893,18 @@ async def run_realtime_pipeline(
)
greeting = await brain.greeting(cfg)
async def capture_image(question: str):
user_id = vision_state["client_id"]
if not user_id:
return None
request = UserImageRequestFrame(
user_id=user_id,
text=question,
append_to_context=False,
function_name=VISION_TOOL_NAME,
)
return await vision_capture.request_image(realtime, request)
recorder = await ConversationRecorder.start(
assistant_id=assistant_id,
assistant_name=cfg.name,
@@ -961,6 +920,7 @@ async def run_realtime_pipeline(
pipeline = Pipeline(
[
transport.input(),
vision_capture,
client_tools,
session_update,
user_input,
@@ -995,6 +955,7 @@ async def run_realtime_pipeline(
session_id=cfg.conversation_id or "",
client_tools=client_tools,
set_input_enabled=set_input_enabled,
capture_image=capture_image if vision_enabled else None,
),
)
@@ -1005,6 +966,8 @@ async def run_realtime_pipeline(
brain=brain,
text_input=user_input,
greeting=greeting,
vision_enabled=vision_enabled,
vision_state=vision_state,
)
runner = WorkerRunner(handle_sigint=False)
run_status = "completed"