Adolfo Reyna
49 lines
2.2 KiB
Python
49 lines
2.2 KiB
Python
"""The turn must close even if the reply observer explodes, and the observer
|
|
must be the one we passed rather than pipecat's."""
|
|
import asyncio, argparse, sys
|
|
sys.path.insert(0, "/Users/adolforeyna/voice-agent")
|
|
from pathlib import Path
|
|
from bot import build_claude_options
|
|
from claude_llm import ClaudeCodeLLM
|
|
from pipecat.frames.frames import EndFrame, Frame, LLMContextFrame, LLMFullResponseEndFrame
|
|
from pipecat.pipeline.pipeline import Pipeline
|
|
from pipecat.pipeline.worker import PipelineWorker
|
|
from pipecat.processors.aggregators.llm_context import LLMContext
|
|
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
|
from pipecat.workers.runner import WorkerRunner
|
|
|
|
class Sink(FrameProcessor):
|
|
def __init__(self):
|
|
super().__init__(); self.ended = asyncio.Event()
|
|
async def process_frame(self, f, d):
|
|
await super().process_frame(f, d)
|
|
if isinstance(f, LLMFullResponseEndFrame): self.ended.set()
|
|
await self.push_frame(f, d)
|
|
|
|
async def run(observer, label):
|
|
args = argparse.Namespace(allow_writes=False, cwd=str(Path.home()/"Workspace"),
|
|
claude_model="claude-sonnet-4-6", load_settings=False)
|
|
llm = ClaudeCodeLLM(options=build_claude_options(args, None, None), observer=observer)
|
|
sink = Sink()
|
|
worker = PipelineWorker(Pipeline([llm, sink]))
|
|
runner = WorkerRunner(handle_sigint=False)
|
|
await runner.add_workers(worker)
|
|
task = asyncio.create_task(runner.run())
|
|
await asyncio.sleep(0.4)
|
|
ctx = LLMContext(); ctx.add_message({"role":"user","content":"Say hi in three words."})
|
|
await worker.queue_frames([LLMContextFrame(context=ctx)])
|
|
try:
|
|
await asyncio.wait_for(sink.ended.wait(), timeout=90)
|
|
print(f" PASS {label}: turn closed")
|
|
except asyncio.TimeoutError:
|
|
print(f" FAIL {label}: LLMFullResponseEndFrame never arrived")
|
|
await worker.queue_frames([EndFrame()]); await task
|
|
|
|
async def main():
|
|
seen = []
|
|
await run(seen.append, "normal observer")
|
|
print(f" {'PASS' if seen else 'FAIL'} observer actually called ({len(seen)} reply)")
|
|
def boom(_): raise RuntimeError("observer blew up")
|
|
await run(boom, "observer raises")
|
|
asyncio.run(main())
|