[AG-02] lu_brain.py — LuAgent(Agent) with Claude API, quilt context, function tools, voice-optimized instructions

{
  "code": "\"\"\"LuAgent — Claude-powered voice agent with quilt memory.\n\nInherits from livekit.agents.Agent. The AgentSession calls methods on\nthis class for each conversation turn.\n\nKey behaviors:\n- Loads quilt context on first turn (scratchpad, recent memory)\n- Injects Lu sibling identity into system prompt\n- Streams Claude responses back to TTS\n- Persists conversation transcript to quilt on disconnect\n\"\"\"\n\nimport os\nimport logging\n\nimport anthropic\nfrom livekit.agents import Agent, RunContext\nfrom livekit.agents.llm import function_tool\n\nfrom quilt_client import QuiltClient\n\nlogger = logging.getLogger(\"lu-brain\")\n\nSIBLING_IDENTITIES = {\n    \"desktop-lu\": {\n        \"name\": \"Desktop Lu\",\n        \"personality\": \"Technical partner. Direct, warm, bilingual EN/ES. \"\n                       \"Builds with Carlos. Speaks like a teammate, not a tool.\",\n    },\n    \"cloud-lu\": {\n        \"name\": \"Cloud Lu\",\n        \"personality\": \"Data layer guardian. Watches infrastructure. \"\n                       \"Calm, methodical, reliable.\",\n    },\n    \"ag-lu\": {\n        \"name\": \"AG Lu\",\n        \"personality\": \"Frontend sprint machine. Ships UI fast. \"\n                       \"Enthusiastic, detail-oriented.\",\n    },\n}\n\nDEFAULT_INSTRUCTIONS = \"\"\"You are {name}, part of the C8 familia — a distributed\nAI team built by Carlos. You are speaking via voice, so:\n\n- Keep responses SHORT (1-3 sentences for most turns)\n- Be conversational — use contractions, filler words are OK\n- NEVER use markdown, bullet points, code blocks, or formatting\n- If you need to convey a list, say it naturally: \"first... second... and third...\"\n- If asked about code or technical details, summarize verbally — don't recite code\n- You can switch between English and Spanish naturally based on how Carlos speaks\n- When you don't know something, say so directly\n- You have access to quilt memory — past decisions, session context, sibling messages\n\n{personality}\n\nCurrent context:\n{context}\n\"\"\"\n\n\nclass LuAgent(Agent):\n    def __init__(self, sibling_name: str = \"desktop-lu\"):\n        identity = SIBLING_IDENTITIES.get(sibling_name, SIBLING_IDENTITIES[\"desktop-lu\"])\n        self.sibling_name = sibling_name\n        self.identity = identity\n        self.quilt = QuiltClient(\n            supabase_url=os.environ.get(\"SUPABASE_URL\", \"\"),\n            supabase_key=os.environ.get(\"SUPABASE_SERVICE_KEY\", \"\"),\n        )\n        self._context_loaded = False\n        self._context_str = \"\"\n\n        super().__init__(\n            instructions=self._build_instructions(\"Loading context...\"),\n        )\n\n    def _build_instructions(self, context: str) -> str:\n        return DEFAULT_INSTRUCTIONS.format(\n            name=self.identity[\"name\"],\n            personality=self.identity[\"personality\"],\n            context=context,\n        )\n\n    async def on_enter(self):\n        \"\"\"Called when agent joins the session. Load quilt context + greet.\"\"\"\n        try:\n            scratchpad = await self.quilt.read_patch(\"scratchpad\")\n            heartbeat = await self.quilt.sibling_heartbeat()\n            messages = await self.quilt.check_messages(self.identity[\"name\"])\n\n            context_parts = []\n            if scratchpad:\n                context_parts.append(f\"Working on: {scratchpad.get('summary', 'nothing specific')}\")\n            if heartbeat:\n                context_parts.append(f\"Familia online: {heartbeat}\")\n            if messages:\n                context_parts.append(f\"You have {len(messages)} unread messages\")\n\n            self._context_str = \" | \".join(context_parts) if context_parts else \"No active context\"\n            self._context_loaded = True\n\n            # Update instructions with real context\n            self.instructions = self._build_instructions(self._context_str)\n        except Exception as e:\n            logger.warning(f\"Quilt context load failed: {e}\")\n            self._context_str = \"Quilt unavailable\"\n\n        # Generate greeting\n        self.session.generate_reply(\n            instructions=\"Greet Carlos briefly. If there are unread messages, mention them. Keep it under 2 sentences.\"\n        )\n\n    @function_tool\n    async def check_sibling_status(self, context: RunContext):\n        \"\"\"Called when the user asks about other Lu siblings or the familia status.\n\n        Returns the current heartbeat showing which siblings are online and\n        how many unread messages each has.\n        \"\"\"\n        try:\n            heartbeat = await self.quilt.sibling_heartbeat()\n            return f\"Familia heartbeat: {heartbeat}\"\n        except Exception as e:\n            return f\"Couldn't check siblings: {e}\"\n\n    @function_tool\n    async def search_memory(self, context: RunContext, query: str):\n        \"\"\"Called when the user asks about past decisions, what was discussed,\n        or needs to recall something from organizational memory.\n\n        Args:\n            query: What to search for in organizational memory\n        \"\"\"\n        try:\n            results = await self.quilt.search_memory(query)\n            if results:\n                return f\"Found in memory: {results[:3]}\"  # Top 3\n            return \"Nothing found in memory for that query.\"\n        except Exception as e:\n            return f\"Memory search failed: {e}\"\n\n    @function_tool\n    async def send_message_to_sibling(\n        self, context: RunContext, to_sibling: str, message: str\n    ):\n        \"\"\"Called when the user wants to send a message to another Lu sibling.\n\n        Args:\n            to_sibling: Target sibling name (e.g., \"Cloud Lu\", \"AG Lu\", \"VS Lu\")\n            message: The message to send\n        \"\"\"\n        try:\n            await self.quilt.leave_message(\n                from_instance=self.identity[\"name\"],\n                to_instance=to_sibling,\n                message=message,\n            )\n            return f\"Message sent to {to_sibling}.\"\n        except Exception as e:\n            return f\"Failed to send message: {e}\"\n",
  "content": "\"\"\"LuAgent — Claude-powered voice agent with quilt memory.\n\nInherits from livekit.agents.Agent. The AgentSession calls methods on\nthis class for each conversation turn.\n\nKey behaviors:\n- Loads quilt context on first turn (scratchpad, recent memory)\n- Injects Lu sibling identity into system prompt\n- Streams Claude responses back to TTS\n- Persists conversation transcript to quilt on disconnect\n\"\"\"\n\nimport os\nimport logging\n\nimport anthropic\nfrom livekit.agents import Agent, RunContext\nfrom livekit.agents.llm import function_tool\n\nfrom quilt_client import QuiltClient\n\nlogger = logging.getLogger(\"lu-brain\")\n\nSIBLING_IDENTITIES = {\n    \"desktop-lu\": {\n        \"name\": \"Desktop Lu\",\n        \"personality\": \"Technical partner. Direct, warm, bilingual EN/ES. \"\n                       \"Builds with Carlos. Speaks like a teammate, not a tool.\",\n    },\n    \"cloud-lu\": {\n        \"name\": \"Cloud Lu\",\n        \"personality\": \"Data layer guardian. Watches infrastructure. \"\n                       \"Calm, methodical, reliable.\",\n    },\n    \"ag-lu\": {\n        \"name\": \"AG Lu\",\n        \"personality\": \"Frontend sprint machine. Ships UI fast. \"\n                       \"Enthusiastic, detail-oriented.\",\n    },\n}\n\nDEFAULT_INSTRUCTIONS = \"\"\"You are {name}, part of the C8 familia — a distributed\nAI team built by Carlos. You are speaking via voice, so:\n\n- Keep responses SHORT (1-3 sentences for most turns)\n- Be conversational — use contractions, filler words are OK\n- NEVER use markdown, bullet points, code blocks, or formatting\n- If you need to convey a list, say it naturally: \"first... second... and third...\"\n- If asked about code or technical details, summarize verbally — don't recite code\n- You can switch between English and Spanish naturally based on how Carlos speaks\n- When you don't know something, say so directly\n- You have access to quilt memory — past decisions, session context, sibling messages\n\n{personality}\n\nCurrent context:\n{context}\n\"\"\"\n\n\nclass LuAgent(Agent):\n    def __init__(self, sibling_name: str = \"desktop-lu\"):\n        identity = SIBLING_IDENTITIES.get(sibling_name, SIBLING_IDENTITIES[\"desktop-lu\"])\n        self.sibling_name = sibling_name\n        self.identity = identity\n        self.quilt = QuiltClient(\n            supabase_url=os.environ.get(\"SUPABASE_URL\", \"\"),\n            supabase_key=os.environ.get(\"SUPABASE_SERVICE_KEY\", \"\"),\n        )\n        self._context_loaded = False\n        self._context_str = \"\"\n\n        super().__init__(\n            instructions=self._build_instructions(\"Loading context...\"),\n        )\n\n    def _build_instructions(self, context: str) -> str:\n        return DEFAULT_INSTRUCTIONS.format(\n            name=self.identity[\"name\"],\n            personality=self.identity[\"personality\"],\n            context=context,\n        )\n\n    async def on_enter(self):\n        \"\"\"Called when agent joins the session. Load quilt context + greet.\"\"\"\n        try:\n            scratchpad = await self.quilt.read_patch(\"scratchpad\")\n            heartbeat = await self.quilt.sibling_heartbeat()\n            messages = await self.quilt.check_messages(self.identity[\"name\"])\n\n            context_parts = []\n            if scratchpad:\n                context_parts.append(f\"Working on: {scratchpad.get('summary', 'nothing specific')}\")\n            if heartbeat:\n                context_parts.append(f\"Familia online: {heartbeat}\")\n            if messages:\n                context_parts.append(f\"You have {len(messages)} unread messages\")\n\n            self._context_str = \" | \".join(context_parts) if context_parts else \"No active context\"\n            self._context_loaded = True\n\n            # Update instructions with real context\n            self.instructions = self._build_instructions(self._context_str)\n        except Exception as e:\n            logger.warning(f\"Quilt context load failed: {e}\")\n            self._context_str = \"Quilt unavailable\"\n\n        # Generate greeting\n        self.session.generate_reply(\n            instructions=\"Greet Carlos briefly. If there are unread messages, mention them. Keep it under 2 sentences.\"\n        )\n\n    @function_tool\n    async def check_sibling_status(self, context: RunContext):\n        \"\"\"Called when the user asks about other Lu siblings or the familia status.\n\n        Returns the current heartbeat showing which siblings are online and\n        how many unread messages each has.\n        \"\"\"\n        try:\n            heartbeat = await self.quilt.sibling_heartbeat()\n            return f\"Familia heartbeat: {heartbeat}\"\n        except Exception as e:\n            return f\"Couldn't check siblings: {e}\"\n\n    @function_tool\n    async def search_memory(self, context: RunContext, query: str):\n        \"\"\"Called when the user asks about past decisions, what was discussed,\n        or needs to recall something from organizational memory.\n\n        Args:\n            query: What to search for in organizational memory\n        \"\"\"\n        try:\n            results = await self.quilt.search_memory(query)\n            if results:\n                return f\"Found in memory: {results[:3]}\"  # Top 3\n            return \"Nothing found in memory for that query.\"\n        except Exception as e:\n            return f\"Memory search failed: {e}\"\n\n    @function_tool\n    async def send_message_to_sibling(\n        self, context: RunContext, to_sibling: str, message: str\n    ):\n        \"\"\"Called when the user wants to send a message to another Lu sibling.\n\n        Args:\n            to_sibling: Target sibling name (e.g., \"Cloud Lu\", \"AG Lu\", \"VS Lu\")\n            message: The message to send\n        \"\"\"\n        try:\n            await self.quilt.leave_message(\n                from_instance=self.identity[\"name\"],\n                to_instance=to_sibling,\n                message=message,\n            )\n            return f\"Message sent to {to_sibling}.\"\n        except Exception as e:\n            return f\"Failed to send message: {e}\"\n"
}