#!/usr/bin/env python3
"""
JARVIS v3 PoC — Pipecat + Hermes API
=====================================
Bot WebSocket texte : transport → LLM → transport
Sans agrégateur — notre processeur custom gère tout.
"""
import json, os, subprocess, sys
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
import uvicorn
import pipecat.frames.frames as pframes
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.frame_processor import FrameProcessor, FrameDirection
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.websocket.fastapi import (
FastAPIWebsocketParams,
FastAPIWebsocketTransport,
)
PORT = int(os.environ.get("JARVIS_PIPECAT_PORT", "7081"))
# ─── Clé API ─────────────────────────────────────────────────
def _get_hermes_key():
key = os.environ.get("HERMES_API_KEY", "") or os.environ.get("HERMES_ALEXA_KEY", "")
if key: return key
try:
creds = json.load(open("/opt/infisical/admin-credentials.json"))
r = subprocess.run(
["/home/openclaw/.local/bin/infisical", "secrets", "get",
"API_SERVER_KEY", "--domain", "https://secret.dracodev.net",
"--projectId", "192c1eed-6666-41ee-80df-a4636a2da0ad",
"--env", "prod", "--path", "/hermes-claw", "--plain",
"--token", creds["token"]],
capture_output=True, text=True, timeout=15)
return r.stdout.strip()
except Exception as e:
print(f"⚠️ Infisical: {e}"); return ""
_hermes_key = _get_hermes_key()
HERMES_URL = "http://localhost:8642/v1"
SYSTEM_PROMPT = (
"Tu es JARVIS, l'assistant vocal de Bruno. "
"Parle en français, sois concis (3 phrases max), chaleureux et utile. "
"Pas de markdown. Réponds comme si tu parlais à l'oral."
)
# ─── Processeur custom ultra-simple ──────────────────────────
class SimpleLLMRelay(FrameProcessor):
"""
Capture les messages texte du transport et les envoie au LLM via
LLMContextFrame. Les réponses du LLM passent directement au transport.
"""
def __init__(self, system_prompt: str):
super().__init__()
self._messages = [{"role": "system", "content": system_prompt}]
async def process_frame(self, frame, direction):
await self.push_frame(frame, direction)
if isinstance(frame, pframes.InputTransportMessageFrame):
text = frame.message
if isinstance(text, str) and text.strip():
self._messages.append({"role": "user", "content": text})
if len(self._messages) > 21:
self._messages = [self._messages[0]] + self._messages[-20:]
print(f"👤 {text[:80]}")
ctx = pframes.LLMContextFrame(context=list(self._messages))
await self.push_frame(ctx)
# ─── Page HTML ───────────────────────────────────────────────
HTML_PAGE = open(os.path.join(os.path.dirname(__file__), "test.html"), "r").read() if os.path.exists("test.html") else """
JARVIS v3
⚡ JARVIS v3 — Pipecat PoC
🔴 Déconnecté
"""
# ─── FastAPI ─────────────────────────────────────────────────
app = FastAPI(title="JARVIS v3 PoC", version="0.1.0")
@app.get("/")
async def root():
return HTMLResponse(content=HTML_PAGE)
@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
transport = FastAPIWebsocketTransport(
websocket=websocket,
params=FastAPIWebsocketParams(
add_wav_header=False,
session_timeout=None,
# allowed_origins vide = autorise toutes les origines
),
)
relay = SimpleLLMRelay(SYSTEM_PROMPT)
llm = OpenAILLMService(
api_key=_hermes_key,
base_url=HERMES_URL,
settings=OpenAILLMService.Settings(
model="Hermes Agent", temperature=0.8, max_tokens=300,
),
)
pipeline = Pipeline(processors=[transport.input(), relay, llm, transport.output()])
task = PipelineTask(pipeline)
@transport.event_handler("on_client_connected")
async def on_connect(transport, client):
print(f"✅ Connecté")
await transport.output().send_message(
pframes.OutputTransportMessageFrame(
message="👋 Bonjour Bruno ! Je suis JARVIS v3. Pose-moi une question !"
)
)
@transport.event_handler("on_client_disconnected")
async def on_disconnect(transport, client):
print(f"👋 Déconnecté")
await task.cancel()
runner = PipelineRunner()
await runner.add_workers(task)
try:
await runner.run()
except Exception as e:
print(f"Pipeline error: {e}")
# ─── Main ────────────────────────────────────────────────────
if __name__ == "__main__":
if not _hermes_key:
print("❌ Pas de clé API Hermes"); sys.exit(1)
print(f"🔑 {_hermes_key[:15]}...")
print(f"🌐 {HERMES_URL}")
print(f"🚀 http://0.0.0.0:{PORT}")
print(f" ws://0.0.0.0:{PORT}/ws")
uvicorn.run(app, host="0.0.0.0", port=PORT, log_level="info")