diff --git a/.gitignore b/.gitignore index 99c618d8..267b8d67 100644 --- a/.gitignore +++ b/.gitignore @@ -47,9 +47,6 @@ htmlcov/ bandit-report.json # Temp / local dev files -branch_structure.json -temp_auto_push.bat -temp_interactive_push.bat # Fonts (generated/local) public/fonts/ diff --git a/community/voice-persona-chat/README.md b/community/voice-persona-chat/README.md new file mode 100644 index 00000000..36cc91d4 --- /dev/null +++ b/community/voice-persona-chat/README.md @@ -0,0 +1,72 @@ +# Voice Persona Chat + +Have a real voice conversation with AI characters via Venice.ai — Einstein, Socrates, Alan Watts, and whoever Venice features next. The characters respond in their authentic style using Venice's native character engine, not a generic LLM with a system prompt. + +## What Makes This Different + +| | voice-persona-chat | dungeon-master-voice | +|---|---|---| +| Character engine | Venice `character_slug` (native persona) | System prompt injected into generic LLM | +| Character roster | Live — fetched from Venice at runtime | Static, hardcoded | +| Fast path | Name in trigger → instant connect | Always requires explicit selection step | +| Returning users | Remembers last character, offers one-tap resume | No memory between sessions | +| Character count | 10+ and growing (Venice's catalog) | 14 fixed DM archetypes | + +## Trigger Phrases + +- "talk to Einstein" +- "chat with Socrates" +- "let me talk to Alan Watts" +- "voice persona" +- "persona chat" +- "connect me to Shakespeare" +- "character chat" +- "conversation with Tesla" + +If you include a character name in the trigger, you skip the selection step entirely. + +## Setup + +1. Get a Venice API key from [venice.ai/settings/api/keys](https://venice.ai/settings/api/keys) +2. Add it to OpenHome settings under the key name: `venice_api_key` + +No other setup required. The character roster is fetched live from Venice each session. + +## How a Session Works + +``` +User: "talk to Einstein" +Assistant: Connecting you to Albert Einstein. Ready? +User: "yes" +Einstein: Ah, a curious mind reaches out! I am Albert Einstein — physicist, philosopher of science, and eternal questioner. What would you like to explore? +User: "what do you think about black holes?" +Einstein: [responds in Einstein's authentic voice and perspective] +... +User: "goodbye" +Einstein: [in-character farewell] +``` + +If you trigger without a name: +``` +User: "voice persona" +Assistant: I can connect you to Alan Watts, Albert Einstein, Socrates, and three others. Who would you like to talk to? +User: "Einstein" +Assistant: You want to talk to Albert Einstein? +User: "yes" +Einstein: [greeting] +``` + +Returning session: +``` +User: "persona chat" +Assistant: Want to talk to Albert Einstein again? +User: "yes" +Einstein: [greeting, no selection step] +``` + +## Notes + +- Conversation history is kept in memory for the session only — not persisted +- The character roster updates automatically as Venice adds new characters +- Say "stop", "quit", "goodbye", or "that's all" to end the conversation +- The character gives a proper in-character farewell on exit diff --git a/community/voice-persona-chat/__init__.py b/community/voice-persona-chat/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/community/voice-persona-chat/main.py b/community/voice-persona-chat/main.py new file mode 100644 index 00000000..38b4ae12 --- /dev/null +++ b/community/voice-persona-chat/main.py @@ -0,0 +1,302 @@ +import json +import requests + +from src.agent.capability import MatchingCapability +from src.agent.capability_worker import CapabilityWorker +from src.main import AgentWorker + +STORAGE_KEY = "voice_persona_chat_prefs" +VENICE_BASE = "https://api.venice.ai/api/v1" +CHAT_MODEL = "llama-3.3-70b" + +EXIT_WORDS = {"stop", "quit", "exit", "end", "bye", "goodbye", "done", "cancel"} +EXIT_PHRASES = {"that's all", "all done", "never mind", "i'm done", "no thanks"} + +HOTWORDS = { + "chat with", "talk to", "speak with", "persona chat", + "character chat", "voice persona", "connect me to", + "let me talk to", "conversation with", +} + + +class VoicePersonaChat(MatchingCapability): + worker: AgentWorker = None + capability_worker: CapabilityWorker = None + + # Do not change following tag of register capability + # {{register capability}} + + _venice_key: str = "" + + def does_match(self, text: str) -> bool: + t = text.lower() + return any(hw in t for hw in HOTWORDS) + + def call(self, worker: AgentWorker): + self.worker = worker + self.capability_worker = CapabilityWorker(self.worker) + self.worker.session_tasks.create(self._run()) + + async def _run(self): + try: + trigger = (await self.capability_worker.wait_for_complete_transcription() or "").strip() + + self._venice_key = self.capability_worker.get_api_keys("venice_api_key") or "" + if not self._venice_key: + await self.capability_worker.speak( + "I need a Venice A P I key to work. " + "Please add it in OpenHome settings under the key name venice underscore api underscore key." + ) + return + + personas = self._fetch_personas() + if not personas: + await self.capability_worker.speak( + "I couldn't reach the Venice character service right now. Please try again in a moment." + ) + return + + name, slug = "", "" + + # Path 1: name found directly in the trigger utterance — fastest path + inline_name, inline_slug = self._match_inline(trigger, personas) + if inline_name and inline_slug: + confirmed = await self.capability_worker.run_confirmation_loop( + f"Connecting you to {inline_name}. Ready?" + ) + if confirmed: + name, slug = inline_name, inline_slug + + # Path 2: returning user — offer last persona + if not slug: + prefs = self._load_prefs() + last_name = prefs.get("last_persona_name", "") + last_slug = prefs.get("last_persona_slug", "") + if last_name and last_slug and last_slug in personas.values(): + want_last = await self.capability_worker.run_confirmation_loop( + f"Want to talk to {last_name} again?" + ) + if want_last: + name, slug = last_name, last_slug + + # Path 3: full roster selection + if not slug: + roster_names = list(personas.keys())[:6] + roster_spoken = ", ".join(roster_names[:-1]) + f", and {roster_names[-1]}" + await self.capability_worker.speak( + f"I can connect you to {roster_spoken}. Who would you like to talk to?" + ) + result = await self._select_persona(personas) + if result is None: + await self.capability_worker.speak("No problem, see you next time!") + return + name, slug = result + + self._save_prefs(slug, name) + + # Open with the character greeting the user + opening = self._venice_chat( + [{"role": "user", "content": "Hello! Introduce yourself in one sentence."}], + slug, + ) + history: list = [] + if opening: + history = [ + {"role": "user", "content": "Hello! Introduce yourself in one sentence."}, + {"role": "assistant", "content": opening}, + ] + await self.capability_worker.speak(opening) + else: + await self.capability_worker.speak( + f"Connected to {name}. Say goodbye or stop whenever you're done." + ) + + await self._chat_loop(history, slug) + + except Exception as e: + self.worker.editor_logging_handler.error(f"[VoicePersonaChat] Error: {e}") + finally: + self.capability_worker.resume_normal_flow() + + # ------------------------------------------------------------------ + # Venice character roster + # ------------------------------------------------------------------ + + def _fetch_personas(self) -> dict: + """Returns {name: slug} for the top featured non-adult characters.""" + try: + resp = requests.get( + f"{VENICE_BASE}/characters", + headers={"Authorization": f"Bearer {self._venice_key}"}, + params={"sortBy": "featured", "limit": 10, "isAdult": "false"}, + timeout=10, + ) + if resp.status_code != 200: + self.worker.editor_logging_handler.error( + f"[VoicePersonaChat] Character fetch failed: {resp.status_code}" + ) + return {} + data = resp.json().get("data", []) + return { + entry["name"]: entry["slug"] + for entry in data + if entry.get("name") and entry.get("slug") + } + except Exception as e: + self.worker.editor_logging_handler.error(f"[VoicePersonaChat] Fetch personas error: {e}") + return {} + + # ------------------------------------------------------------------ + # Persona selection + # ------------------------------------------------------------------ + + def _match_inline(self, trigger: str, personas: dict) -> tuple: + """Scans trigger utterance for a known persona name. Returns (name, slug) or ('', '').""" + t = trigger.lower() + for name, slug in personas.items(): + if name.lower() in t: + return name, slug + return "", "" + + async def _select_persona(self, personas: dict) -> tuple | None: + """Retry loop: user speaks a name, LLM fuzzy-matches it. Returns (name, slug) or None on exit.""" + for attempt in range(4): + raw = (await self.capability_worker.user_response() or "").strip() + + if self._is_exit(raw): + return None + + slug = self._match_persona_name(raw, personas) + + if slug == "list_roster": + roster_names = list(personas.keys())[:6] + roster_spoken = ", ".join(roster_names[:-1]) + f", and {roster_names[-1]}" + await self.capability_worker.speak(f"Available: {roster_spoken}.") + continue + + if slug: + matched_name = next((n for n, s in personas.items() if s == slug), "") + confirmed = await self.capability_worker.run_confirmation_loop( + f"You want to talk to {matched_name}?" + ) + if confirmed: + return matched_name, slug + + if attempt < 3: + await self.capability_worker.speak( + "I didn't catch that. Say a character name, or ask me to list them." + ) + + return None + + def _match_persona_name(self, user_input: str, personas: dict) -> str: + """LLM fuzzy-match: returns a slug, 'list_roster', or empty string.""" + names_list = list(personas.keys()) + raw = (self.capability_worker.text_to_text_response( + f"User said: '{user_input}'. Available characters: {json.dumps(names_list)}. " + "If the user is asking to list or hear the characters, return: {\"match\": \"list_roster\"}. " + "Otherwise return the closest matching character name: {\"match\": \"\"}. " + "Return ONLY valid JSON, no other text." + ) or "").strip() + + if "```" in raw: + parts = raw.split("```") + raw = parts[1].lstrip("json").strip() if len(parts) > 1 else "" + + try: + match = (json.loads(raw).get("match") or "").strip() + if match == "list_roster": + return "list_roster" + return personas.get(match, "") + except Exception: + return "" + + # ------------------------------------------------------------------ + # Chat loop + # ------------------------------------------------------------------ + + async def _chat_loop(self, history: list, slug: str): + while True: + user_input = (await self.capability_worker.user_response() or "").strip() + + if not user_input: + await self.capability_worker.speak("I didn't catch that.") + continue + + if self._is_exit(user_input): + farewell = self._venice_chat( + history + [{"role": "user", "content": "Say goodbye in character in one sentence."}], + slug, + ) + if farewell: + await self.capability_worker.speak(farewell) + return + + history.append({"role": "user", "content": user_input}) + reply = self._venice_chat(history, slug) + + if not reply: + await self.capability_worker.speak("I didn't get a response. Please try again.") + history.pop() + continue + + history.append({"role": "assistant", "content": reply}) + await self.capability_worker.speak(reply) + + def _venice_chat(self, messages: list, slug: str) -> str: + try: + resp = requests.post( + f"{VENICE_BASE}/chat/completions", + headers={ + "Authorization": f"Bearer {self._venice_key}", + "Content-Type": "application/json", + }, + json={ + "model": CHAT_MODEL, + "messages": messages, + "venice_parameters": {"character_slug": slug}, + }, + timeout=30, + ) + if resp.status_code != 200: + self.worker.editor_logging_handler.error( + f"[VoicePersonaChat] Chat API error: {resp.status_code}" + ) + return "" + return (resp.json()["choices"][0]["message"]["content"] or "").strip() + except Exception as e: + self.worker.editor_logging_handler.error(f"[VoicePersonaChat] Venice chat error: {e}") + return "" + + # ------------------------------------------------------------------ + # Storage — Uzair create-first pattern + # ------------------------------------------------------------------ + + def _load_prefs(self) -> dict: + try: + raw = self.capability_worker.get_single_key(STORAGE_KEY) + return raw.get("value", raw) if raw else {} + except Exception: + return {} + + def _save_prefs(self, slug: str, name: str): + data = {"last_persona_slug": slug, "last_persona_name": name} + try: + result = self.capability_worker.create_key(STORAGE_KEY, data) + if not (result or {}).get("success"): + self.capability_worker.update_key(STORAGE_KEY, data) + except Exception as e: + self.worker.editor_logging_handler.error(f"[VoicePersonaChat] Save prefs error: {e!r}") + + # ------------------------------------------------------------------ + # Exit detection — two-tier (token match + phrase match) + # ------------------------------------------------------------------ + + def _is_exit(self, text: str) -> bool: + if not text: + return False + t = text.lower().strip() + tokens = set(t.split()) + if tokens & EXIT_WORDS: + return True + return any(phrase in t for phrase in EXIT_PHRASES)