"""A Pipecat processor that puts OpenCode (with ollama-cloud/gemma4:31b) in the LLM slot. Drives the installed OpenCode CLI (`opencode run -m ollama-cloud/gemma4:31b`) to stream cloud responses to text-to-speech downstream with zero local GPU overhead. """ import asyncio import json import os import re import shutil from pathlib import Path from loguru import logger from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, InterruptionFrame, LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMTextFrame, StartFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor _NOISE_TRANSCRIPTS = { "", ".", "thank you.", "thanks for watching!", "you", "bye.", "okay.", "[blank_audio]", "[silence]", } def _clean_spoken_text(text: str) -> str: """Clean text for speech output and truncate fake turn generations.""" if not text: return "" # Truncate if model hallucinates fake turn markers for marker in ("User:", "Human:", "Assistant:", "\nUser", "\nHuman", "\nAssistant"): if marker in text: text = text.split(marker)[0] # Remove markdown code blocks text = re.sub(r"```[\s\S]*?```", "", text) # Remove inline code ticks text = re.sub(r"`[^`]*`", "", text) # Remove markdown syntax characters text = re.sub(r"[\#\*\_\~]", "", text) # Flatten newlines into clear speech lines = [line.strip() for line in text.splitlines() if line.strip()] return " ".join(lines).strip() def find_opencode_cli() -> str | None: return shutil.which("opencode") or ( "/Users/adolforeyna/.opencode/bin/opencode" if os.path.exists("/Users/adolforeyna/.opencode/bin/opencode") else None ) def probe_opencode(model: str = "ollama-cloud/gemma4:31b") -> tuple[bool, str]: cli = find_opencode_cli() if not cli: return False, "OpenCode CLI binary not found" return True, f"OpenCode CLI available ({cli}) with model {model}" class OpenCodeLLM(FrameProcessor): """Runs user turns through the OpenCode CLI driving OpenCode Cloud models.""" def __init__( self, *, model: str = "ollama-cloud/gemma4:31b", cwd: str | Path | None = None, system_prompt: str | None = None, observer=None, **kwargs, ): super().__init__(**kwargs) self._model = model self._cwd = Path(cwd or Path.home() / "Workspace").expanduser().resolve() self._system_prompt = system_prompt or "You are a helpful spoken voice assistant. Keep answers brief and conversational." self._on_reply = observer self._turn_task: asyncio.Task | None = None self._history: list[dict[str, str]] = [] self._cli_path = find_opencode_cli() or "opencode" async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, StartFrame): await self.push_frame(frame, direction) logger.info(f"OpenCode LLM engine ready: CLI={self._cli_path}, model={self._model}") elif isinstance(frame, (EndFrame, CancelFrame)): await self._cancel_turn() await self.push_frame(frame, direction) elif isinstance(frame, InterruptionFrame): await self._cancel_turn() await self.push_frame(frame, direction) elif isinstance(frame, LLMContextFrame): text = self._latest_user_text(frame.context) await self._maybe_start_turn(text) else: await self.push_frame(frame, direction) def _latest_user_text(self, context) -> str: if not context or not hasattr(context, "messages"): return "" for msg in reversed(context.messages): if isinstance(msg, dict) and msg.get("role") == "user": content = msg.get("content", "") if isinstance(content, str): return content elif isinstance(content, list): text_parts = [c.get("text", "") for c in content if isinstance(c, dict)] return " ".join(text_parts) return "" async def _maybe_start_turn(self, text: str): utterance = text.strip() if utterance.lower() in _NOISE_TRANSCRIPTS or len(utterance) < 2: logger.debug(f"Ignoring noise transcript: {utterance!r}") return await self._cancel_turn() logger.info(f"You: {utterance}") self._turn_task = self.create_task(self._run_turn(utterance)) async def _cancel_turn(self): if not self._turn_task: return task, self._turn_task = self._turn_task, None await self.cancel_task(task) async def _run_turn(self, utterance: str): self._history.append({"role": "user", "content": utterance}) # Format prompt with system instructions and recent conversation history recent_history = self._history[-6:] conv_text = "\n".join( f"{'User' if m['role']=='user' else 'Assistant'}: {m['content']}" for m in recent_history ) prompt_str = f"{self._system_prompt}\n\n{conv_text}\nAssistant:" await self.push_frame(LLMFullResponseStartFrame()) chunks: list[str] = [] try: proc = await asyncio.create_subprocess_exec( self._cli_path, "run", "-m", self._model, "--dir", str(self._cwd), "--pure", prompt_str, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, stdin=asyncio.subprocess.DEVNULL, cwd=str(self._cwd), ) while True: line = await proc.stdout.readline() if not line: break text_line = line.decode("utf-8") cleaned = _clean_spoken_text(text_line) if cleaned: chunks.append(cleaned) await self.push_frame(LLMTextFrame(cleaned)) await proc.wait() except asyncio.CancelledError: logger.info("OpenCode turn cancelled mid-response.") if proc and proc.returncode is None: try: proc.kill() except Exception: pass raise except Exception as e: logger.error(f"OpenCode LLM error: {e}") err_msg = "Sorry, I ran into an error generating a response." chunks.append(err_msg) await self.push_frame(LLMTextFrame(err_msg)) finally: await self.push_frame(LLMFullResponseEndFrame()) full_reply = _clean_spoken_text(" ".join(chunks)) if full_reply: self._history.append({"role": "assistant", "content": full_reply}) logger.info(f"OpenCode LLM ({self._model}): {full_reply}") if self._on_reply: self._on_reply(full_reply)