"""A Pipecat processor that puts OpenCode (with ollama-cloud/gemma4:31b) in the LLM slot. Supports both: 1. OpenCode Server Daemon (`opencode serve --port 4096`) for zero-latency, persistent in-memory sessions. 2. OpenCode CLI (`opencode run --continue`) for direct process invocation. """ import asyncio import json import os import re import shutil from pathlib import Path import aiohttp 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]", } _server_proc: asyncio.subprocess.Process | None = None 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 ) async def ensure_opencode_server(port: int = 4096) -> tuple[bool, str]: """Ensure opencode serve daemon is running on port.""" global _server_proc url = f"http://localhost:{port}/session" try: async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=2.0)) as session: async with session.get(url) as resp: if resp.status == 200: return True, f"OpenCode server active on http://localhost:{port}" except Exception: pass cli = find_opencode_cli() if not cli: return False, "OpenCode CLI binary not found" logger.info(f"Starting OpenCode server daemon on port {port}...") try: _server_proc = await asyncio.create_subprocess_exec( cli, "serve", "--port", str(port), stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, ) await asyncio.sleep(1.2) return True, f"Started OpenCode server daemon on http://localhost:{port}" except Exception as e: return False, f"Failed to start OpenCode server daemon: {e}" 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 available ({cli}) with model {model}" class OpenCodeLLM(FrameProcessor): """Runs user turns through OpenCode Server Daemon or CLI.""" def __init__( self, *, model: str = "ollama-cloud/gemma4:31b", cwd: str | Path | None = None, port: int = 4096, system_prompt: str | None = None, observer=None, use_server: bool = False, **kwargs, ): super().__init__(**kwargs) self._model = model self._cwd = Path(cwd or Path.home() / "Workspace").expanduser().resolve() self._port = port 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" self._use_server = use_server self._server_session_id: str | None = None self._has_session = False 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) if self._use_server: ok, reason = await ensure_opencode_server(self._port) logger.info(f"OpenCode Server engine: {reason}") else: logger.info(f"OpenCode CLI 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}) 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] = [] if self._use_server: await self._run_turn_server(prompt_str, chunks) else: await self._run_turn_cli(prompt_str, chunks) 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) async def _run_turn_server(self, prompt_str: str, chunks: list[str]): """Run turn via OpenCode Server Daemon HTTP API.""" try: ok, _ = await ensure_opencode_server(self._port) if not ok: raise RuntimeError("OpenCode server daemon unavailable") async with aiohttp.ClientSession() as session: if not self._server_session_id: create_url = f"http://localhost:{self._port}/session" async with session.post(create_url, json={"directory": str(self._cwd)}) as res: if res.status == 200: data = await res.json() self._server_session_id = data.get("id") logger.info(f"OpenCode server session created: {self._server_session_id}") if not self._server_session_id: raise RuntimeError("Failed to create OpenCode server session") msg_url = f"http://localhost:{self._port}/session/{self._server_session_id}/message" model_id = self._model.split("/")[-1] if "/" in self._model else self._model provider_id = self._model.split("/")[0] if "/" in self._model else "ollama-cloud" payload = { "model": {"providerID": provider_id, "modelID": model_id}, "parts": [{"type": "text", "text": prompt_str}], } async with session.post(msg_url, json=payload) as resp: if resp.status == 200: data = await resp.json() parts = data.get("parts", []) if isinstance(data, dict) else [] for p in parts: if isinstance(p, dict): p_type = p.get("type") if p_type == "text" and "text" in p: text_chunk = _clean_spoken_text(p["text"]) if text_chunk: chunks.append(text_chunk) await self.push_frame(LLMTextFrame(text_chunk)) elif p_type not in ("step-start", "step-finish"): logger.info(f"OpenCode Tool: {p_type} -> {json.dumps(p)[:120]}") if not chunks and isinstance(data, dict): if "delta" in data: text_chunk = _clean_spoken_text(data["delta"]) if text_chunk: chunks.append(text_chunk) await self.push_frame(LLMTextFrame(text_chunk)) elif "text" in data: text_chunk = _clean_spoken_text(data["text"]) if text_chunk: chunks.append(text_chunk) await self.push_frame(LLMTextFrame(text_chunk)) else: err_text = await resp.text() logger.error(f"OpenCode server HTTP {resp.status}: {err_text}") except asyncio.CancelledError: logger.info("OpenCode server turn cancelled mid-response.") raise except Exception as e: logger.warning(f"OpenCode server error ({e}), falling back to CLI...") await self._run_turn_cli(prompt_str, chunks) finally: await self.push_frame(LLMFullResponseEndFrame()) async def _run_turn_cli(self, prompt_str: str, chunks: list[str]): """Fallback turn via OpenCode CLI.""" cmd = [ self._cli_path, "run", "-m", self._model, "--dir", str(self._cwd), "--auto", ] if self._has_session: cmd.append("--continue") cmd.append(prompt_str) proc = None try: proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, stdin=asyncio.subprocess.DEVNULL, cwd=str(self._cwd), ) self._has_session = True async def _read_stderr(stream): while True: line = await stream.readline() if not line: break decoded = line.decode("utf-8").strip() if decoded and not decoded.startswith(">"): logger.info(f"OpenCode Tool: {decoded}") stderr_task = asyncio.create_task(_read_stderr(proc.stderr)) 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() await stderr_task 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 CLI error: {e}") err_msg = "Sorry, I ran into an error generating a response." chunks.append(err_msg) await self.push_frame(LLMTextFrame(err_msg))