453 lines
17 KiB
Python
453 lines
17 KiB
Python
|
"""Reusable frame processors shared by cascade and realtime pipelines."""
|
|||
|
|
|
|||
|
|
import asyncio
|
|||
|
|
from collections.abc import Callable
|
|||
|
|
from uuid import uuid4
|
|||
|
|
|
|||
|
|
from loguru import logger
|
|||
|
|
from models import AssistantConfig
|
|||
|
|
from services.brains import Brain
|
|||
|
|
from services.conversation_history import ConversationRecorder
|
|||
|
|
from services.knowledge import search as search_knowledge
|
|||
|
|
from db.session import SessionLocal
|
|||
|
|
|
|||
|
|
from pipecat.frames.frames import (
|
|||
|
|
InputTransportMessageFrame,
|
|||
|
|
InterruptionFrame,
|
|||
|
|
LLMContextFrame,
|
|||
|
|
LLMFullResponseEndFrame,
|
|||
|
|
LLMFullResponseStartFrame,
|
|||
|
|
LLMTextFrame,
|
|||
|
|
OutputTransportMessageUrgentFrame,
|
|||
|
|
TextFrame,
|
|||
|
|
UserImageRawFrame,
|
|||
|
|
UserImageRequestFrame,
|
|||
|
|
)
|
|||
|
|
from pipecat.processors.aggregators.llm_response_universal import (
|
|||
|
|
LLMAssistantAggregator,
|
|||
|
|
)
|
|||
|
|
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
|||
|
|
from pipecat.turns.user_mute.base_user_mute_strategy import BaseUserMuteStrategy
|
|||
|
|
from pipecat.utils.time import time_now_iso8601
|
|||
|
|
|
|||
|
|
|
|||
|
|
KNOWLEDGE_CONTEXT_MARKER = "<!-- knowledge-context -->"
|
|||
|
|
|
|||
|
|
|
|||
|
|
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._mode = "automatic" if knowledge_base_id else "disabled"
|
|||
|
|
self._last_signature = ""
|
|||
|
|
|
|||
|
|
def set_scope(self, scope: dict) -> None:
|
|||
|
|
self._knowledge_base_id = scope.get("knowledge_base_id") or None
|
|||
|
|
self._mode = str(scope.get("mode") or "disabled")
|
|||
|
|
self._top_n = int(scope.get("top_n") or 5)
|
|||
|
|
self._score_threshold = float(scope.get("score_threshold") or 0.0)
|
|||
|
|
self._last_signature = ""
|
|||
|
|
|
|||
|
|
def _clear_context(self, messages: list[dict]) -> None:
|
|||
|
|
# Remove the legacy Workflow knowledge message so an in-flight context
|
|||
|
|
# created before this compatibility fix cannot keep sending that role.
|
|||
|
|
messages[:] = [
|
|||
|
|
message
|
|||
|
|
for message in messages
|
|||
|
|
if not (
|
|||
|
|
message.get("role") == "developer"
|
|||
|
|
and KNOWLEDGE_CONTEXT_MARKER in str(message.get("content") or "")
|
|||
|
|
)
|
|||
|
|
]
|
|||
|
|
system_message = next(
|
|||
|
|
(message for message in messages if message.get("role") == "system"),
|
|||
|
|
None,
|
|||
|
|
)
|
|||
|
|
if system_message is not None:
|
|||
|
|
content = str(system_message.get("content") or "")
|
|||
|
|
system_message["content"] = content.split(KNOWLEDGE_CONTEXT_MARKER, 1)[0].rstrip()
|
|||
|
|
|
|||
|
|
def _set_context(self, messages: list[dict], block: str) -> None:
|
|||
|
|
"""Store retrieved knowledge in a provider-compatible system message."""
|
|||
|
|
self._clear_context(messages)
|
|||
|
|
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})
|
|||
|
|
return
|
|||
|
|
content = str(system_message.get("content") or "").rstrip()
|
|||
|
|
system_message["content"] = f"{content}\n\n{block}" if content else block
|
|||
|
|
|
|||
|
|
async def process_frame(self, frame, direction: FrameDirection):
|
|||
|
|
await super().process_frame(frame, direction)
|
|||
|
|
if not isinstance(frame, LLMContextFrame):
|
|||
|
|
await self.push_frame(frame, direction)
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
messages = frame.context.get_messages()
|
|||
|
|
if self._mode != "automatic" or not self._knowledge_base_id:
|
|||
|
|
self._clear_context(messages)
|
|||
|
|
await self.push_frame(frame, direction)
|
|||
|
|
return
|
|||
|
|
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}"
|
|||
|
|
self._set_context(messages, block)
|
|||
|
|
await self.push_frame(frame, direction)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class UserTurnRoutingProcessor(FrameProcessor):
|
|||
|
|
"""Give a brain first right of refusal before a new user turn reaches the LLM."""
|
|||
|
|
|
|||
|
|
def __init__(self, brain: Brain):
|
|||
|
|
super().__init__()
|
|||
|
|
self._brain = brain
|
|||
|
|
self._last_user_message: dict | None = None
|
|||
|
|
|
|||
|
|
async def process_frame(self, frame, direction: FrameDirection):
|
|||
|
|
await super().process_frame(frame, direction)
|
|||
|
|
if direction != FrameDirection.DOWNSTREAM or not isinstance(
|
|||
|
|
frame, LLMContextFrame
|
|||
|
|
):
|
|||
|
|
await self.push_frame(frame, direction)
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
user_message = next(
|
|||
|
|
(
|
|||
|
|
message
|
|||
|
|
for message in reversed(frame.context.get_messages())
|
|||
|
|
if message.get("role") == "user"
|
|||
|
|
and isinstance(message.get("content"), str)
|
|||
|
|
and str(message.get("content") or "").strip()
|
|||
|
|
),
|
|||
|
|
None,
|
|||
|
|
)
|
|||
|
|
if user_message is None:
|
|||
|
|
await self.push_frame(frame, direction)
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
if user_message is self._last_user_message:
|
|||
|
|
# Programmatic LLMRunFrame after a node transition reuses the same
|
|||
|
|
# user message. It is a response run, not another routing event.
|
|||
|
|
await self.push_frame(frame, direction)
|
|||
|
|
return
|
|||
|
|
self._last_user_message = user_message
|
|||
|
|
|
|||
|
|
content = str(user_message.get("content") or "").strip()
|
|||
|
|
handled = await self._brain.on_user_turn_end(content)
|
|||
|
|
if not handled:
|
|||
|
|
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 = ""
|
|||
|
|
|
|||
|
|
|
|||
|
|
class WorkflowAggregatorPair:
|
|||
|
|
"""Small public-shape adapter required by Pipecat FlowManager."""
|
|||
|
|
|
|||
|
|
def __init__(self, user_aggregator, assistant_aggregator):
|
|||
|
|
self._user = user_aggregator
|
|||
|
|
self._assistant = assistant_aggregator
|
|||
|
|
|
|||
|
|
def user(self):
|
|||
|
|
return self._user
|
|||
|
|
|
|||
|
|
def assistant(self):
|
|||
|
|
return self._assistant
|
|||
|
|
|
|||
|
|
|
|||
|
|
|