From a8ad741a4ddc2bb803a0737c834a8df716e31095 Mon Sep 17 00:00:00 2001 From: Adolfo Reyna Date: Fri, 7 Aug 2026 19:41:07 -0400 Subject: [PATCH] Implement OpenCode server daemon integration with automatic fallback to CLI --- opencode_llm.py | 140 ++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 123 insertions(+), 17 deletions(-) diff --git a/opencode_llm.py b/opencode_llm.py index 9c03a99..7fe1546 100644 --- a/opencode_llm.py +++ b/opencode_llm.py @@ -1,7 +1,8 @@ """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. +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 @@ -10,6 +11,7 @@ import os import re import shutil from pathlib import Path +import aiohttp from loguru import logger from pipecat.frames.frames import ( @@ -37,6 +39,8 @@ _NOISE_TRANSCRIPTS = { "[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.""" @@ -65,33 +69,70 @@ def find_opencode_cli() -> str | 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 CLI available ({cli}) with model {model}" + return True, f"OpenCode available ({cli}) with model {model}" class OpenCodeLLM(FrameProcessor): - """Runs user turns through the OpenCode CLI driving OpenCode Cloud models.""" + """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 = True, **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): @@ -99,7 +140,11 @@ class OpenCodeLLM(FrameProcessor): if isinstance(frame, StartFrame): await self.push_frame(frame, direction) - logger.info(f"OpenCode LLM engine ready: CLI={self._cli_path}, model={self._model}") + 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) @@ -144,7 +189,6 @@ class OpenCodeLLM(FrameProcessor): 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']}" @@ -155,7 +199,77 @@ class OpenCodeLLM(FrameProcessor): await self.push_frame(LLMFullResponseStartFrame()) chunks: list[str] = [] - # Use --continue for subsequent turns to persist tool session state + 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: + async for line in resp.content: + line_str = line.decode("utf-8").strip() + if not line_str: + continue + try: + data = json.loads(line_str) + 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)) + except json.JSONDecodeError: + pass + + 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", @@ -167,6 +281,7 @@ class OpenCodeLLM(FrameProcessor): cmd.append("--continue") cmd.append(prompt_str) + proc = None try: proc = await asyncio.create_subprocess_exec( *cmd, @@ -210,16 +325,7 @@ class OpenCodeLLM(FrameProcessor): pass raise except Exception as e: - logger.error(f"OpenCode LLM error: {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)) - 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)