Refactor pipeline and assistant page components for improved structure and performance
- Remove unused imports and classes from pipeline.py to streamline the codebase. - Consolidate dynamic variable handling and workflow management in AssistantPage, enhancing clarity and maintainability. - Update WorkflowEditor to utilize a more modular approach, improving the overall architecture and reducing complexity. - Enhance the import structure across components for better organization and readability.
This commit is contained in:
Xin Wang
452
backend/services/pipecat/processors.py
Normal file
452
backend/services/pipecat/processors.py
Normal file
@@ -0,0 +1,452 @@
|
||||
"""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
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user