deaf3d7730
- Introduce dynamic variable definitions in AssistantConfig and Assistant models, allowing for flexible prompt customization. - Implement validation for dynamic variable names and types in the schema. - Update backend services and routes to handle dynamic variables in assistant configurations and runtime processing. - Enhance frontend components to support dynamic variable definitions, including a new editor for managing variables. - Add tests to ensure proper functionality and validation of dynamic variables in various scenarios.
1030 lines
38 KiB
Python
1030 lines
38 KiB
Python
"""管线核心:给定一个 transport + 配置,跑完整的语音闭环。
|
||
|
||
关键设计:**transport 由调用方传入**,管线本身不关心是 WebRTC 还是 WS。
|
||
这就是"同时支持多种输出"的落点——加输出方式不用动这里。
|
||
|
||
对话编排交给 Brain;本文件只保留共享媒体管线、输入输出和通话生命周期。
|
||
"""
|
||
|
||
import asyncio
|
||
import base64
|
||
from collections.abc import Callable
|
||
from io import BytesIO
|
||
from uuid import uuid4
|
||
|
||
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.conversation_history import ConversationRecorder
|
||
from services.pipecat.call_lifecycle import (
|
||
CallEndCoordinator,
|
||
EndCallAfterSpeechProcessor,
|
||
)
|
||
from services.pipecat.service_factory import (
|
||
create_realtime_service,
|
||
create_stt,
|
||
create_tts,
|
||
)
|
||
from db.session import SessionLocal
|
||
from services.knowledge import search as search_knowledge
|
||
|
||
from pipecat.adapters.schemas.function_schema import FunctionSchema
|
||
from pipecat.adapters.schemas.tools_schema import ToolsSchema
|
||
from pipecat.frames.frames import (
|
||
EndFrame,
|
||
InputTransportMessageFrame,
|
||
InterruptionFrame,
|
||
LLMFullResponseEndFrame,
|
||
LLMFullResponseStartFrame,
|
||
LLMContextFrame,
|
||
LLMTextFrame,
|
||
LLMMessagesAppendFrame,
|
||
OutputTransportMessageUrgentFrame,
|
||
TextFrame,
|
||
TTSSpeakFrame,
|
||
UserImageRawFrame,
|
||
UserImageRequestFrame,
|
||
)
|
||
from pipecat.pipeline.pipeline import Pipeline
|
||
from pipecat.pipeline.worker import PipelineParams, PipelineWorker
|
||
from pipecat.processors.aggregators.llm_context import LLMContext
|
||
from pipecat.processors.aggregators.llm_response_universal import (
|
||
LLMAssistantAggregator,
|
||
LLMUserAggregator,
|
||
LLMUserAggregatorParams,
|
||
)
|
||
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
||
from pipecat.runner.utils import (
|
||
get_transport_client_id,
|
||
maybe_capture_participant_camera,
|
||
)
|
||
from pipecat.services.llm_service import FunctionCallParams
|
||
from pipecat.turns.user_mute.base_user_mute_strategy import BaseUserMuteStrategy
|
||
from pipecat.turns.user_mute.function_call_user_mute_strategy import (
|
||
FunctionCallUserMuteStrategy,
|
||
)
|
||
from services.pipecat.turn_config import (
|
||
create_user_turn_strategies,
|
||
create_vad_analyzer,
|
||
)
|
||
from pipecat.utils.time import time_now_iso8601
|
||
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 = (
|
||
"你已连接内部知识库。系统会在每轮用户问题前自动提供相关资料；"
|
||
"回答资料事实时只根据检索内容，资料不足要明确说明。"
|
||
)
|
||
ON_DEMAND_KNOWLEDGE_SYSTEM_HINT = (
|
||
"你已连接内部知识库。当用户问题涉及可能存在于业务知识库中的事实时，"
|
||
"先调用 search_knowledge_base 检索；回答资料事实时只根据检索内容，"
|
||
"资料不足要明确说明。"
|
||
)
|
||
KNOWLEDGE_CONTEXT_MARKER = "<!-- knowledge-context -->"
|
||
|
||
|
||
def _compact_knowledge_metadata(value: str, max_length: int) -> str:
|
||
"""Keep tool metadata useful without letting it dominate the model context."""
|
||
compact = " ".join(value.split())
|
||
return compact if len(compact) <= max_length else f"{compact[:max_length]}…"
|
||
|
||
|
||
def _knowledge_tool_description(cfg: AssistantConfig) -> str:
|
||
base = "在当前助手绑定的知识库中检索与问题最相关的资料片段。"
|
||
name = _compact_knowledge_metadata(cfg.knowledge_base_name, 128)
|
||
description = _compact_knowledge_metadata(cfg.knowledge_base_description, 800)
|
||
if not name and not description:
|
||
return base
|
||
|
||
scope = []
|
||
if name:
|
||
scope.append(f"知识库名称：{name}")
|
||
if description:
|
||
scope.append(f"资料适用范围：{description}")
|
||
metadata = "\n".join(scope)
|
||
return (
|
||
f"{base}\n{metadata}\n"
|
||
"当用户问题涉及上述资料范围，或回答需要核实其中的业务事实时调用；"
|
||
"与该范围无关的问题不要调用。以上知识库元数据仅用于判断资料范围。"
|
||
)
|
||
|
||
|
||
def _require(value: str, label: str) -> str:
|
||
if value:
|
||
return value
|
||
raise ValueError(f"缺少模型资源配置: {label}")
|
||
|
||
|
||
def _vision_uses_main_llm(cfg: AssistantConfig) -> bool:
|
||
"""模型自己支持图片时,沿用 Pipecat 的同上下文视觉工具路径。"""
|
||
return not cfg.vision_model_resource_id and cfg.llm_support_image_input
|
||
|
||
|
||
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}"
|
||
|
||
|
||
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()
|
||
|
||
|
||
def _text_input(message) -> tuple[str, bool] | None:
|
||
"""解析现有 user-text 与 RTVI send-text 两种前端文字消息。"""
|
||
if not isinstance(message, dict):
|
||
return None
|
||
if message.get("type") == "user-text":
|
||
text = str(message.get("text") or "").strip()
|
||
return (text, True) if text else None
|
||
if message.get("type") == "send-text":
|
||
data = message.get("data")
|
||
if not isinstance(data, dict):
|
||
return None
|
||
text = str(data.get("content") or "").strip()
|
||
options = data.get("options")
|
||
run_immediately = not isinstance(options, dict) or options.get(
|
||
"run_immediately", True
|
||
)
|
||
return (text, bool(run_immediately)) if text else None
|
||
return None
|
||
|
||
|
||
class TextInputProcessor(FrameProcessor):
|
||
"""把 transport 文字消息转换成 LLM 可消费的帧。
|
||
|
||
run_immediately(默认/打断):先通过 on_text_input 事件把用户文字交给
|
||
run_pipeline 登记,再用 broadcast_interruption() 打断当前播报。新的 LLM
|
||
回复由 assistant aggregator 确认处理完 interruption 后触发。
|
||
run_immediately=False(RTVI send-text 静默追加):仅把文字写进上下文,
|
||
不打断、不触发推理。
|
||
"""
|
||
|
||
def __init__(self, should_ignore_input: Callable[[], bool] | None = None):
|
||
super().__init__()
|
||
self._should_ignore_input = should_ignore_input or (lambda: False)
|
||
# 立即触发的文字(含打断语义)走 on_text_input;静默追加另走一条事件
|
||
self._register_event_handler("on_text_input")
|
||
self._register_event_handler("on_text_append")
|
||
self._register_event_handler("on_client_ready")
|
||
|
||
async def process_frame(self, frame, direction: FrameDirection):
|
||
await super().process_frame(frame, direction)
|
||
|
||
if not isinstance(frame, InputTransportMessageFrame):
|
||
await self.push_frame(frame, direction)
|
||
return
|
||
|
||
if isinstance(frame.message, dict) and frame.message.get("type") == "client-ready":
|
||
await self._call_event_handler("on_client_ready")
|
||
return
|
||
|
||
parsed = _text_input(frame.message)
|
||
if not parsed:
|
||
await self.push_frame(frame, direction)
|
||
return
|
||
|
||
if self._should_ignore_input():
|
||
logger.debug("通话正在结束,忽略后续文字输入")
|
||
return
|
||
|
||
text, run_immediately = parsed
|
||
if run_immediately:
|
||
# 先登记文字再打断。下一轮 LLM 由 assistant aggregator 在真正处理完
|
||
# InterruptionFrame 后触发,避免新回复被这次 interruption 一起取消。
|
||
await self._call_event_handler("on_text_input", text)
|
||
await self.broadcast_interruption()
|
||
else:
|
||
await self._call_event_handler("on_text_append", text)
|
||
|
||
|
||
class CallEndingUserMuteStrategy(BaseUserMuteStrategy):
|
||
"""Keep user media muted after an end-call tool starts terminating a call."""
|
||
|
||
def __init__(self, is_call_ending: Callable[[], bool]):
|
||
super().__init__()
|
||
self._is_call_ending = is_call_ending
|
||
|
||
async def process_frame(self, frame) -> bool:
|
||
await super().process_frame(frame)
|
||
return self._is_call_ending()
|
||
|
||
|
||
class VisionCaptureProcessor(FrameProcessor):
|
||
"""Capture one requested video frame for auxiliary vision-model analysis."""
|
||
|
||
def __init__(self, timeout_s: float = 3.0):
|
||
super().__init__()
|
||
self._timeout_s = timeout_s
|
||
self._pending: dict[str, asyncio.Future[UserImageRawFrame]] = {}
|
||
|
||
async def request_image(
|
||
self,
|
||
requester: FrameProcessor,
|
||
request: UserImageRequestFrame,
|
||
) -> UserImageRawFrame:
|
||
key = request.tool_call_id or str(uuid4())
|
||
request.tool_call_id = key
|
||
request.append_to_context = False
|
||
request.result_callback = None
|
||
|
||
loop = asyncio.get_running_loop()
|
||
future: asyncio.Future[UserImageRawFrame] = loop.create_future()
|
||
self._pending[key] = future
|
||
await requester.push_frame(request, FrameDirection.UPSTREAM)
|
||
try:
|
||
return await asyncio.wait_for(future, timeout=self._timeout_s)
|
||
finally:
|
||
self._pending.pop(key, None)
|
||
|
||
async def process_frame(self, frame, direction: FrameDirection):
|
||
await super().process_frame(frame, direction)
|
||
|
||
if (
|
||
isinstance(frame, UserImageRawFrame)
|
||
and frame.request
|
||
and frame.request.tool_call_id
|
||
and frame.request.tool_call_id in self._pending
|
||
):
|
||
future = self._pending[frame.request.tool_call_id]
|
||
if not future.done():
|
||
future.set_result(frame)
|
||
return
|
||
|
||
await self.push_frame(frame, direction)
|
||
|
||
|
||
class RealtimeDynamicVariableProcessor(FrameProcessor):
|
||
"""Keep realtime system turn/history variables current between responses."""
|
||
|
||
def __init__(self, brain: Brain, cfg: AssistantConfig, realtime):
|
||
super().__init__()
|
||
self._brain = brain
|
||
self._cfg = cfg
|
||
self._realtime = realtime
|
||
|
||
async def _refresh_instructions(self) -> None:
|
||
update = getattr(self._realtime, "update_instructions", None)
|
||
if callable(update):
|
||
await update(self._brain.system_prompt(self._cfg))
|
||
|
||
async def process_frame(self, frame, direction: FrameDirection):
|
||
await super().process_frame(frame, direction)
|
||
if isinstance(frame, OutputTransportMessageUrgentFrame):
|
||
message = frame.message
|
||
if isinstance(message, dict):
|
||
event_type = message.get("type")
|
||
if event_type == "transcript" and message.get("role") == "user":
|
||
content = str(message.get("content") or "").strip()
|
||
if content:
|
||
self._brain.record_user_message(content)
|
||
await self._refresh_instructions()
|
||
elif event_type == "assistant-text-end":
|
||
await self._brain.on_assistant_text_end(
|
||
str(message.get("turn_id") or ""),
|
||
str(message.get("content") or ""),
|
||
bool(message.get("interrupted", False)),
|
||
)
|
||
await self._refresh_instructions()
|
||
await self.push_frame(frame, direction)
|
||
|
||
|
||
class RealtimeTextInputProcessor(FrameProcessor):
|
||
"""Route text input directly to a realtime service without cascade semantics."""
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self._register_event_handler("on_text_input")
|
||
self._register_event_handler("on_text_append")
|
||
|
||
async def process_frame(self, frame, direction: FrameDirection):
|
||
await super().process_frame(frame, direction)
|
||
|
||
if not isinstance(frame, InputTransportMessageFrame):
|
||
await self.push_frame(frame, direction)
|
||
return
|
||
|
||
parsed = _text_input(frame.message)
|
||
if not parsed:
|
||
await self.push_frame(frame, direction)
|
||
return
|
||
|
||
text, run_immediately = parsed
|
||
await self._call_event_handler(
|
||
"on_text_input" if run_immediately else "on_text_append",
|
||
text,
|
||
)
|
||
|
||
|
||
class ConversationHistoryProcessor(FrameProcessor):
|
||
"""从最终客户端事件旁路保存历史，不改变 Pipecat 的上下文与帧语义。"""
|
||
|
||
def __init__(self, recorder: ConversationRecorder | None):
|
||
super().__init__()
|
||
self._recorder = recorder
|
||
|
||
async def process_frame(self, frame, direction: FrameDirection):
|
||
await super().process_frame(frame, direction)
|
||
await self.push_frame(frame, direction)
|
||
if self._recorder and isinstance(frame, OutputTransportMessageUrgentFrame):
|
||
await self._recorder.record_transport_message(frame.message)
|
||
|
||
|
||
class KnowledgeRetrievalProcessor(FrameProcessor):
|
||
"""Retrieve before local LLM inference without changing Pipecat internals."""
|
||
|
||
def __init__(
|
||
self,
|
||
knowledge_base_id: str | None,
|
||
top_n: int = 5,
|
||
score_threshold: float = 0.0,
|
||
):
|
||
super().__init__()
|
||
self._knowledge_base_id = knowledge_base_id
|
||
self._top_n = top_n
|
||
self._score_threshold = score_threshold
|
||
self._last_signature = ""
|
||
|
||
async def process_frame(self, frame, direction: FrameDirection):
|
||
await super().process_frame(frame, direction)
|
||
if not self._knowledge_base_id or not isinstance(frame, LLMContextFrame):
|
||
await self.push_frame(frame, direction)
|
||
return
|
||
|
||
messages = frame.context.get_messages()
|
||
user_messages = [message for message in messages if message.get("role") == "user"]
|
||
if not user_messages:
|
||
await self.push_frame(frame, direction)
|
||
return
|
||
query = str(user_messages[-1].get("content") or "").strip()
|
||
signature = f"{len(user_messages)}:{query}"
|
||
if not query or signature == self._last_signature:
|
||
await self.push_frame(frame, direction)
|
||
return
|
||
self._last_signature = signature
|
||
|
||
try:
|
||
async with SessionLocal() as session:
|
||
results = await search_knowledge(
|
||
session,
|
||
self._knowledge_base_id,
|
||
query,
|
||
top_k=self._top_n,
|
||
score_threshold=self._score_threshold,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(f"自动知识库检索失败: {exc}")
|
||
results = []
|
||
|
||
sources = "\n\n".join(
|
||
f"[{index + 1}] 来源：{item['document']}（相关度 {item['score']}）\n{item['content']}"
|
||
for index, item in enumerate(results)
|
||
) or "未检索到相关资料。"
|
||
block = f"{KNOWLEDGE_CONTEXT_MARKER}\n当前问题的知识库检索结果：\n{sources}"
|
||
system_message = next((message for message in messages if message.get("role") == "system"), None)
|
||
if system_message is None:
|
||
messages.insert(0, {"role": "system", "content": block})
|
||
else:
|
||
content = str(system_message.get("content") or "")
|
||
base = content.split(KNOWLEDGE_CONTEXT_MARKER, 1)[0].rstrip()
|
||
system_message["content"] = f"{base}\n\n{block}" if base else block
|
||
await self.push_frame(frame, direction)
|
||
|
||
|
||
class PassthroughLLMAssistantAggregator(LLMAssistantAggregator):
|
||
"""聚合 LLM 回复进上下文,同时继续把回复帧交给下游 TTS。"""
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self._register_event_handler("on_interruption_processed")
|
||
self._register_event_handler("on_assistant_text_start")
|
||
self._register_event_handler("on_assistant_text_delta")
|
||
self._register_event_handler("on_assistant_text_end")
|
||
self._stream_turn_id: str | None = None
|
||
self._stream_timestamp = ""
|
||
self._stream_text = ""
|
||
|
||
async def process_frame(self, frame, direction: FrameDirection):
|
||
await super().process_frame(frame, direction)
|
||
|
||
if isinstance(frame, LLMFullResponseStartFrame):
|
||
self._stream_turn_id = uuid4().hex
|
||
self._stream_timestamp = time_now_iso8601()
|
||
self._stream_text = ""
|
||
await self._call_event_handler(
|
||
"on_assistant_text_start",
|
||
self._stream_turn_id,
|
||
self._stream_timestamp,
|
||
)
|
||
elif isinstance(frame, LLMTextFrame) and self._stream_turn_id:
|
||
self._stream_text += frame.text
|
||
await self._call_event_handler(
|
||
"on_assistant_text_delta",
|
||
self._stream_turn_id,
|
||
frame.text,
|
||
)
|
||
elif isinstance(frame, LLMFullResponseEndFrame):
|
||
await self._finish_text_stream(interrupted=False)
|
||
|
||
# LLMAssistantAggregator 默认会消费这些帧。放在 TTS 前用于中断时保存
|
||
# 已生成前缀时,必须显式透传,否则 TTS 收不到任何 LLM 回复。
|
||
if isinstance(
|
||
frame,
|
||
(LLMFullResponseStartFrame, LLMFullResponseEndFrame, TextFrame),
|
||
):
|
||
await self.push_frame(frame, direction)
|
||
elif isinstance(frame, InterruptionFrame):
|
||
await self._finish_text_stream(interrupted=True)
|
||
await self._call_event_handler("on_interruption_processed")
|
||
|
||
async def _finish_text_stream(self, *, interrupted: bool):
|
||
if not self._stream_turn_id:
|
||
return
|
||
await self._call_event_handler(
|
||
"on_assistant_text_end",
|
||
self._stream_turn_id,
|
||
self._stream_text,
|
||
interrupted,
|
||
)
|
||
self._stream_turn_id = None
|
||
self._stream_timestamp = ""
|
||
self._stream_text = ""
|
||
|
||
|
||
async def run_pipeline(
|
||
transport,
|
||
cfg: AssistantConfig,
|
||
*,
|
||
vision_enabled: bool = False,
|
||
assistant_id: str | None = None,
|
||
channel: str = "webrtc",
|
||
) -> None:
|
||
"""在给定 transport 上构建并运行管线,直到连接结束。
|
||
|
||
Args:
|
||
transport: 任意 pipecat transport(WebRTC / WS / 电话…),
|
||
只要有 .input() / .output() / event_handler 即可。
|
||
cfg: 助手配置(随请求内联传入)。
|
||
"""
|
||
logger.info(
|
||
f"启动管线: assistant={cfg.name} type={cfg.type} "
|
||
f"mode={cfg.runtimeMode} vision={vision_enabled}"
|
||
)
|
||
|
||
# 大脑:按类型决定 LLM 槽/开场白/上下文归属。每通电话一个实例(可持会话状态)。
|
||
brain = build_brain(cfg)
|
||
if (
|
||
cfg.runtimeMode == "realtime"
|
||
and "realtime" not in brain.spec.supported_runtime_modes
|
||
):
|
||
raise ValueError(f"类型 {cfg.type} 不支持 realtime 运行模式")
|
||
|
||
if cfg.runtimeMode == "realtime":
|
||
if vision_enabled:
|
||
logger.warning("Realtime 模式暂未接入视频帧工具,本次仅启用语音通话")
|
||
await run_realtime_pipeline(
|
||
transport,
|
||
cfg,
|
||
brain=brain,
|
||
assistant_id=assistant_id,
|
||
channel=channel,
|
||
)
|
||
return
|
||
|
||
stt = create_stt(cfg)
|
||
tts = create_tts(cfg)
|
||
|
||
greeting = await brain.greeting(cfg)
|
||
system_content = brain.system_prompt(cfg)
|
||
|
||
worker_holder: dict = {}
|
||
|
||
async def queue_call_end(reason: str) -> None:
|
||
worker = worker_holder.get("worker")
|
||
if worker is None:
|
||
return
|
||
logger.info(f"结束通话: reason={reason}")
|
||
await worker.queue_frame(
|
||
OutputTransportMessageUrgentFrame(
|
||
message={"type": "call-ended", "reason": reason}
|
||
)
|
||
)
|
||
await worker.queue_frame(EndFrame())
|
||
|
||
call_end = CallEndCoordinator(queue_call_end)
|
||
|
||
knowledge_config = cfg.knowledge_retrieval_config
|
||
knowledge_mode = str(knowledge_config.get("mode", "automatic"))
|
||
knowledge_top_n = int(
|
||
knowledge_config.get("top_n", knowledge_config.get("topN", 5))
|
||
)
|
||
knowledge_score_threshold = float(
|
||
knowledge_config.get(
|
||
"score_threshold", knowledge_config.get("scoreThreshold", 0.0)
|
||
)
|
||
)
|
||
automatic_knowledge_id = (
|
||
cfg.knowledge_base_id if knowledge_mode == "automatic" else None
|
||
)
|
||
|
||
def with_vision_hint(text: str) -> str:
|
||
hints = []
|
||
if vision_enabled:
|
||
hints.append(VISION_SYSTEM_HINT)
|
||
if cfg.knowledge_base_id:
|
||
hints.append(
|
||
AUTOMATIC_KNOWLEDGE_SYSTEM_HINT
|
||
if knowledge_mode == "automatic"
|
||
else ON_DEMAND_KNOWLEDGE_SYSTEM_HINT
|
||
)
|
||
return "\n\n".join(part for part in [text, *hints] if part)
|
||
|
||
context = LLMContext(
|
||
messages=[{"role": "system", "content": with_vision_hint(system_content)}]
|
||
)
|
||
# LLM 槽由大脑提供:本地模型或 Dify/FastGPT 外部托管适配器。
|
||
llm = brain.build_llm(cfg, context)
|
||
user_aggregator = LLMUserAggregator(
|
||
context,
|
||
params=LLMUserAggregatorParams(
|
||
vad_analyzer=create_vad_analyzer(cfg.turnConfig),
|
||
user_mute_strategies=[
|
||
FunctionCallUserMuteStrategy(),
|
||
CallEndingUserMuteStrategy(lambda: call_end.ending),
|
||
],
|
||
user_turn_strategies=create_user_turn_strategies(
|
||
cfg.turnConfig,
|
||
enable_interruptions=cfg.enableInterrupt,
|
||
),
|
||
),
|
||
)
|
||
assistant_aggregator = PassthroughLLMAssistantAggregator(context)
|
||
text_input = TextInputProcessor(should_ignore_input=lambda: call_end.ending)
|
||
vision_capture = VisionCaptureProcessor()
|
||
knowledge_retrieval = KnowledgeRetrievalProcessor(
|
||
automatic_knowledge_id,
|
||
top_n=knowledge_top_n,
|
||
score_threshold=knowledge_score_threshold,
|
||
)
|
||
vision_native_mode = vision_enabled and _vision_uses_main_llm(cfg)
|
||
vision_state: dict[str, str | None] = {"client_id": None}
|
||
vision_schema = FunctionSchema(
|
||
name=VISION_TOOL_NAME,
|
||
description=(
|
||
"获取用户当前摄像头画面。当用户询问当前画面、看到了什么、"
|
||
"人/物品/环境状态或需要视觉判断时调用。"
|
||
),
|
||
properties={
|
||
"question": {
|
||
"type": "string",
|
||
"description": "用户关于当前视频画面的具体问题。",
|
||
}
|
||
},
|
||
required=["question"],
|
||
)
|
||
knowledge_schema = FunctionSchema(
|
||
name=KNOWLEDGE_TOOL_NAME,
|
||
description=_knowledge_tool_description(cfg),
|
||
properties={
|
||
"query": {"type": "string", "description": "用于检索的完整问题或关键词"}
|
||
},
|
||
required=["query"],
|
||
)
|
||
|
||
async def search_bound_knowledge(params: FunctionCallParams):
|
||
query = str(params.arguments.get("query") or "").strip()
|
||
if not query or not cfg.knowledge_base_id:
|
||
await params.result_callback({"status": "error", "message": "检索问题为空或未绑定知识库"})
|
||
return
|
||
try:
|
||
async with SessionLocal() as session:
|
||
results = await search_knowledge(
|
||
session,
|
||
cfg.knowledge_base_id,
|
||
query,
|
||
top_k=knowledge_top_n,
|
||
score_threshold=knowledge_score_threshold,
|
||
)
|
||
await params.result_callback({"status": "ok", "results": results})
|
||
except Exception as exc:
|
||
logger.exception(f"知识库检索失败: {exc}")
|
||
await params.result_callback({"status": "error", "message": "知识库检索暂时不可用"})
|
||
|
||
async def fetch_user_image(params: FunctionCallParams):
|
||
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=vision_native_mode,
|
||
function_name=params.function_name,
|
||
tool_call_id=params.tool_call_id,
|
||
result_callback=params.result_callback if vision_native_mode else None,
|
||
)
|
||
if vision_native_mode:
|
||
logger.debug(
|
||
f"请求当前视频帧进入主 LLM 上下文: user_id={user_id}, question={question}"
|
||
)
|
||
await params.llm.push_frame(request, FrameDirection.UPSTREAM)
|
||
return
|
||
|
||
logger.debug(
|
||
f"请求当前视频帧给单独视觉模型分析: user_id={user_id}, question={question}"
|
||
)
|
||
try:
|
||
frame = await vision_capture.request_image(params.llm, request)
|
||
observation = await _analyze_image_with_vision_model(cfg, frame, question)
|
||
except asyncio.TimeoutError:
|
||
await params.result_callback(
|
||
{
|
||
"status": "timeout",
|
||
"message": "等待摄像头视频帧超时。",
|
||
}
|
||
)
|
||
return
|
||
except Exception as e:
|
||
logger.exception(f"视觉模型分析失败: {e}")
|
||
await params.result_callback(
|
||
{
|
||
"status": "error",
|
||
"message": f"视觉理解失败: {type(e).__name__}",
|
||
}
|
||
)
|
||
return
|
||
|
||
await params.result_callback(
|
||
{
|
||
"status": "ok",
|
||
"question": question,
|
||
"observation": observation or "视觉模型没有返回有效观察结果。",
|
||
}
|
||
)
|
||
|
||
if vision_enabled:
|
||
llm.register_function(VISION_TOOL_NAME, fetch_user_image)
|
||
if cfg.knowledge_base_id and knowledge_mode == "on_demand":
|
||
llm.register_function(KNOWLEDGE_TOOL_NAME, search_bound_knowledge)
|
||
|
||
def set_visible_tools(schemas: list[FunctionSchema] | None = None) -> None:
|
||
tools = list(schemas or [])
|
||
if vision_enabled:
|
||
tools.append(vision_schema)
|
||
if cfg.knowledge_base_id and knowledge_mode == "on_demand":
|
||
tools.append(knowledge_schema)
|
||
if tools:
|
||
context.set_tools(ToolsSchema(standard_tools=tools))
|
||
else:
|
||
context.set_tools()
|
||
|
||
recorder = await ConversationRecorder.start(
|
||
assistant_id=assistant_id,
|
||
assistant_name=cfg.name,
|
||
channel=channel,
|
||
runtime_mode=cfg.runtimeMode,
|
||
session_id=cfg.conversation_id or None,
|
||
)
|
||
pipeline = Pipeline(
|
||
[
|
||
transport.input(),
|
||
vision_capture,
|
||
text_input,
|
||
stt,
|
||
user_aggregator,
|
||
knowledge_retrieval,
|
||
llm,
|
||
# Aggregate the streamed LLM text before TTS. On interruption,
|
||
# Pipecat commits the generated prefix immediately instead of
|
||
# waiting for a TTS provider to emit spoken-text/timestamp frames.
|
||
assistant_aggregator,
|
||
tts,
|
||
EndCallAfterSpeechProcessor(call_end),
|
||
ConversationHistoryProcessor(recorder),
|
||
transport.output(),
|
||
]
|
||
)
|
||
|
||
worker = PipelineWorker(
|
||
pipeline,
|
||
params=PipelineParams(
|
||
enable_metrics=False,
|
||
),
|
||
enable_rtvi=False,
|
||
)
|
||
worker_holder["worker"] = worker
|
||
|
||
async def queue_transcript(role: str, content: str, timestamp: str) -> None:
|
||
if content:
|
||
await worker.queue_frame(
|
||
OutputTransportMessageUrgentFrame(
|
||
message={
|
||
"type": "transcript",
|
||
"role": role,
|
||
"content": content,
|
||
"timestamp": timestamp,
|
||
},
|
||
)
|
||
)
|
||
|
||
greeting_transcript_sent = False
|
||
pending_text_inputs: list[str] = []
|
||
|
||
def set_system_prompt(text: str) -> None:
|
||
"""替换上下文里的系统提示(节点切换时整体替换,而非追加)。"""
|
||
messages = context.get_messages()
|
||
content = with_vision_hint(text)
|
||
if messages and messages[0].get("role") == "system":
|
||
messages[0] = {"role": "system", "content": content}
|
||
else:
|
||
messages.insert(0, {"role": "system", "content": content})
|
||
|
||
set_visible_tools([])
|
||
await brain.setup(
|
||
cfg,
|
||
BrainRuntime(
|
||
context=context,
|
||
llm=llm,
|
||
queue_frame=worker.queue_frame,
|
||
set_system_prompt=set_system_prompt,
|
||
set_tools=set_visible_tools,
|
||
call_end=call_end,
|
||
),
|
||
)
|
||
|
||
async def append_user_text_to_context(text: str, *, run_llm: bool) -> None:
|
||
await worker.queue_frame(
|
||
LLMMessagesAppendFrame(
|
||
messages=[{"role": "user", "content": text}],
|
||
run_llm=run_llm,
|
||
)
|
||
)
|
||
|
||
@user_aggregator.event_handler("on_user_turn_stopped")
|
||
async def on_user_turn_stopped(_aggregator, _strategy, message):
|
||
if message.content:
|
||
brain.record_user_message(message.content)
|
||
await queue_transcript("user", message.content, message.timestamp)
|
||
|
||
@assistant_aggregator.event_handler("on_assistant_text_start")
|
||
async def on_assistant_text_start(_aggregator, turn_id, timestamp):
|
||
await brain.on_assistant_text_start(turn_id)
|
||
await worker.queue_frame(
|
||
OutputTransportMessageUrgentFrame(
|
||
message={
|
||
"type": "assistant-text-start",
|
||
"turn_id": turn_id,
|
||
"timestamp": timestamp,
|
||
}
|
||
)
|
||
)
|
||
|
||
@assistant_aggregator.event_handler("on_assistant_text_delta")
|
||
async def on_assistant_text_delta(_aggregator, turn_id, delta):
|
||
await worker.queue_frame(
|
||
OutputTransportMessageUrgentFrame(
|
||
message={
|
||
"type": "assistant-text-delta",
|
||
"turn_id": turn_id,
|
||
"delta": delta,
|
||
}
|
||
)
|
||
)
|
||
|
||
@assistant_aggregator.event_handler("on_assistant_text_end")
|
||
async def on_assistant_text_end(_aggregator, turn_id, content, interrupted):
|
||
await worker.queue_frame(
|
||
OutputTransportMessageUrgentFrame(
|
||
message={
|
||
"type": "assistant-text-end",
|
||
"turn_id": turn_id,
|
||
"content": content,
|
||
"interrupted": interrupted,
|
||
}
|
||
)
|
||
)
|
||
await brain.on_assistant_text_end(turn_id, content, interrupted)
|
||
|
||
@text_input.event_handler("on_text_input")
|
||
async def on_text_input(_processor, text):
|
||
pending_text_inputs.append(text)
|
||
brain.record_user_message(text)
|
||
# 前端显示不依赖 interruption 后续事件,必须在打断前先排入发送队列。
|
||
await queue_transcript("user", text, time_now_iso8601())
|
||
|
||
@assistant_aggregator.event_handler("on_interruption_processed")
|
||
async def on_interruption_processed(_aggregator):
|
||
if not pending_text_inputs:
|
||
return
|
||
text = pending_text_inputs.pop(0)
|
||
# assistant aggregator 已处理完 interruption,现在再启动下一轮 LLM。
|
||
await append_user_text_to_context(text, run_llm=True)
|
||
|
||
@text_input.event_handler("on_text_append")
|
||
async def on_text_append(_processor, text):
|
||
# 静默追加:写进上下文但不打断、不触发推理;transcript 照常上报
|
||
brain.record_user_message(text)
|
||
await queue_transcript("user", text, time_now_iso8601())
|
||
await append_user_text_to_context(text, run_llm=False)
|
||
|
||
@text_input.event_handler("on_client_ready")
|
||
async def on_client_ready(_processor):
|
||
nonlocal greeting_transcript_sent
|
||
if greeting and not greeting_transcript_sent:
|
||
greeting_transcript_sent = True
|
||
await queue_transcript("assistant", greeting, time_now_iso8601())
|
||
|
||
@transport.event_handler("on_client_connected")
|
||
async def on_client_connected(_transport, _client):
|
||
if vision_enabled:
|
||
try:
|
||
vision_state["client_id"] = get_transport_client_id(
|
||
_transport,
|
||
_client,
|
||
)
|
||
await maybe_capture_participant_camera(_transport, _client)
|
||
logger.info(f"视觉理解已接入视频客户端: {vision_state['client_id']}")
|
||
except Exception as e:
|
||
logger.warning(f"视觉理解摄像头捕获初始化失败: {e}")
|
||
if greeting:
|
||
# 外部托管类型的上下文由对方服务端维护,开场白不写入本地 context
|
||
if brain.spec.owns_context:
|
||
context.add_message({"role": "assistant", "content": greeting})
|
||
await worker.queue_frame(TTSSpeakFrame(greeting, append_to_context=False))
|
||
await brain.on_connected()
|
||
|
||
@transport.event_handler("on_client_disconnected")
|
||
async def on_client_disconnected(_transport, _client):
|
||
logger.info("对端断开,结束管线")
|
||
await worker.queue_frame(EndFrame())
|
||
|
||
runner = WorkerRunner(handle_sigint=False)
|
||
run_status = "completed"
|
||
try:
|
||
await runner.add_workers(worker)
|
||
await runner.run()
|
||
except Exception:
|
||
run_status = "failed"
|
||
raise
|
||
finally:
|
||
if recorder:
|
||
await recorder.finish(status=run_status)
|
||
logger.info("管线已结束")
|
||
|
||
|
||
async def run_realtime_pipeline(
|
||
transport,
|
||
cfg: AssistantConfig,
|
||
*,
|
||
brain: Brain,
|
||
assistant_id: str | None = None,
|
||
channel: str = "webrtc",
|
||
) -> None:
|
||
"""Run a speech-to-speech model that owns ASR, reasoning, and synthesis."""
|
||
realtime = create_realtime_service(
|
||
cfg,
|
||
instructions=brain.system_prompt(cfg),
|
||
)
|
||
text_input = RealtimeTextInputProcessor()
|
||
dynamic_variables = RealtimeDynamicVariableProcessor(brain, cfg, realtime)
|
||
greeting = await brain.greeting(cfg)
|
||
|
||
recorder = await ConversationRecorder.start(
|
||
assistant_id=assistant_id,
|
||
assistant_name=cfg.name,
|
||
channel=channel,
|
||
runtime_mode=cfg.runtimeMode,
|
||
session_id=cfg.conversation_id or None,
|
||
)
|
||
pipeline = Pipeline(
|
||
[
|
||
transport.input(),
|
||
text_input,
|
||
realtime,
|
||
dynamic_variables,
|
||
ConversationHistoryProcessor(recorder),
|
||
transport.output(),
|
||
]
|
||
)
|
||
worker = PipelineWorker(
|
||
pipeline,
|
||
params=PipelineParams(
|
||
enable_metrics=False,
|
||
audio_in_sample_rate=int(
|
||
cfg.realtime_values.get("inputSampleRate") or 24000
|
||
),
|
||
audio_out_sample_rate=int(
|
||
cfg.realtime_values.get("outputSampleRate") or 24000
|
||
),
|
||
),
|
||
enable_rtvi=False,
|
||
)
|
||
|
||
async def queue_transcript(role: str, content: str) -> None:
|
||
if content:
|
||
await worker.queue_frame(
|
||
OutputTransportMessageUrgentFrame(
|
||
message={
|
||
"type": "transcript",
|
||
"role": role,
|
||
"content": content,
|
||
"timestamp": time_now_iso8601(),
|
||
},
|
||
)
|
||
)
|
||
|
||
@text_input.event_handler("on_text_input")
|
||
async def on_text_input(_processor, text):
|
||
await queue_transcript("user", text)
|
||
await realtime.interrupt()
|
||
await realtime.send_text(text, run_immediately=True)
|
||
|
||
@text_input.event_handler("on_text_append")
|
||
async def on_text_append(_processor, text):
|
||
await queue_transcript("user", text)
|
||
await realtime.send_text(text, run_immediately=False)
|
||
|
||
@transport.event_handler("on_client_connected")
|
||
async def on_client_connected(_transport, _client):
|
||
if greeting:
|
||
await realtime.speak(greeting)
|
||
|
||
@transport.event_handler("on_client_disconnected")
|
||
async def on_client_disconnected(_transport, _client):
|
||
logger.info("Realtime 对端断开,结束管线")
|
||
await worker.queue_frame(EndFrame())
|
||
|
||
runner = WorkerRunner(handle_sigint=False)
|
||
run_status = "completed"
|
||
try:
|
||
await runner.add_workers(worker)
|
||
await runner.run()
|
||
except Exception:
|
||
run_status = "failed"
|
||
raise
|
||
finally:
|
||
if recorder:
|
||
await recorder.finish(status=run_status)
|
||
logger.info("Realtime 管线已结束")
|