# Project export: The Conductor

This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.

## Project metadata

- Hackathon: OpenAI Build Week
- Tagline: An open, multi-agent framework separating orchestration from intelligence. It routes tactile, physical triggers to specialized AI agents to seamlessly orchestrate your day.
- Devpost: https://devpost.com/software/the-conductor-z2pqhs
- GitHub: https://github.com/falisha-shoun/Conductor-Hackathon
- Video: https://www.youtube.com/embed/Hciln74TWnk?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 1 GitHub contributor(s) — falisha-shoun (4 commits)

## Devpost submission (written by the team)

### Inspiration

Most AI experiences today are trapped behind a monolithic chat box. But our daily lives aren't lived in text threads—they are tactile, physical, and multi-sensory. I wanted to step out of the browser and build a framework for a context-aware partner in physical space. This inspired the "Orchestra Metaphor." Instead of one giant AI trying to do everything, I envisioned a "Conductor" (a central event router) that delegates tasks to specialized, single-purpose AI agents (the "Line Cooks").

### What it does

The Conductor is an open, event-driven framework for orchestrating cooperative AI agents. In this Version 0.1 MVP, the system uses a Streamlit dashboard to simulate a physical hardware trigger. When triggered, the Conductor executes a JSON "Score." Rather than executing logic itself, it simultaneously routes commands across an Event Bus to specialized agents. A single tap can trigger a Memory Agent to log the stateful interaction, a Voice Agent to speak a greeting, and a Music Agent to queue a playlist—all running concurrently.

### How we built it

We adhered to a 7-Layer Multi-Agent Architecture. We built a zero-latency web dashboard using Streamlit, connected to a Python backend powered by asyncio to handle non-blocking event routing. To ensure the Conductor had "Working Memory" without the latency of a vector database, we utilized a session-scoped Python dictionary. Codex and GPT-5.6 acted as my Senior Implementation Engineers, instantly writing the asyncio.gather() logic required for concurrent multi-agent actions.

### Challenges we ran into

Building an asynchronous event loop from scratch in Python is inherently complex. Managing state so that the Conductor didn't suffer from "amnesia" required careful planning. Furthermore, translating the vision of a physical cardboard prototype into a purely digital MVP required pivoting our hardware integration strategy to a web-based UI.

### Accomplishments we're proud of

We successfully separated orchestration from intelligence. Building a fully functional, zero-latency Event Bus and stateful Memory cache using lightweight Python dictionaries instead of heavy databases proved that multi-agent systems can be incredibly fast.

### What we learned

I learned how to deeply integrate asynchronous Python programming (asyncio) and how to use Codex as a strategic architectural partner rather than just a simple code generator.

### What's next

The next step is Layer 6: Hardware Integration. We plan to move the simulation out of the browser and into our tactile cardboard enclosure, mapping physical sensors (like piezoelectric vibration sensors) to the Event Bus to bring The Conductor into the physical world.

## README (from the GitHub repository)

# 🎼 The Conductor

An open, event-driven framework for orchestrating cooperative AI agents across physical and digital environments.

## 🌟 The Vision (Concept UI)

<img width="1536" height="1024" alt="Prototype_1" src="https://github.com/user-attachments/assets/a83136ac-bb83-47a2-8c33-cd980f4d250e" />

<img width="1024" height="1536" alt="Prototype_2" src="https://github.com/user-attachments/assets/e6a0058a-073d-4ad5-83dd-f0bb23e04d93" />

## 🧠 The Philosophy (The Orchestra Metaphor)
Most AI experiences today are trapped behind a monolithic chat box. "The Conductor" separates orchestration from intelligence. 
* **The Conductor (The Head Chef):** Standing at the kitchen pass, the Conductor reads incoming order tickets and coordinates who executes them. The Conductor never executes logic itself.
* **Specialized Agents (The Line Cooks):** Independent modules designed to do one single task perfectly (e.g., streaming music, speech).
* **The Event Bus (The Ticket Rail):** The central channel where events are published. No agent talks directly to another.

## 📦 What We Built (Hackathon MVP v0.1)
Codex and I built a fully functional **Layer 1 & 2 Prototype** using Streamlit and Python `asyncio`. 
* **Concurrent Execution:** Using the Score Engine, a single trigger can tell our Voice Agent to speak and our Music Agent to play a playlist simultaneously.
* **Stateful Memory:** A lightweight, session-scoped working memory cache that logs all events so the Conductor doesn't suffer from amnesia.

## 🤖 Codex Collaboration
Codex and GPT-5.6 acted as my Senior Implementation Engineers. I fed Codex our "7-Layer Architecture" as raw text, and it successfully built our asynchronous event loop. It accelerated the workflow by instantly writing the `asyncio.gather()` logic required for concurrent multi-agent actions, allowing me to focus on the overarching system design.

## 🚀 How to Run Locally
1. `pip install -r requirements.txt`
2. `streamlit run dashboard.py`


## Detected evidence (automated analysis)

Indexed codebase: 6 recognized source files, 16 KB.
- Python (language) — detected in the code
- Streamlit (technology) — detected in the code
- OpenAI (technology) — claimed on Devpost, not found in the code

## Codebase structure (from repository index)

### Files (9 of 9)

```
ARCHITECTURE.md
dashboard.py
DESIGN.md
LICENSE
main.py
PRODUCT_SPEC_BOOK.txt
PROJECT_CHARTER.md
README.md
requirements.txt
```

### Dependencies

- requirements.txt: streamlit@>=1.36,<2

### Recent commits (newest first)

- Add files via upload
- Update README.md
- Add files via upload
- Initial commit

## Key source files (fetched from GitHub, selected and truncated for size)

### DESIGN.md

```markdown
# Design Book: The Conductor

## 1. The Physical Experience (Hardware)
* **Material:** Tactile Cardboard Enclosure.
* **Vibe:** "Scrappy Maker" meets Embodied AI.
* **Interaction:** The primary human interaction is physical (a double-tap on the box). The AI responds via internal audio (Bluetooth speaker).

## 2. The Digital Experience (Companion App)
The Companion App is NOT the product; it is the "remote control" and visual monitor for the physical speaker.
* **Style:** Clean, modern, dark-mode dashboard.
* **Live Activity:** A visual cascading timeline showing the Event Bus in real-time (e.g., Sensor -> Conductor -> Voice).
* **Agent Manager:** Simple toggle cards showing which agents are currently active.
* **The Metaphor:** The user should feel like they are conducting an orchestra, not managing a server.

```

### ARCHITECTURE.md

```markdown
# Architecture: The Conductor

## Core Philosophy
The Conductor coordinates; specialized agents execute. Everything in this system communicates via an asynchronous Event Bus. There is no direct agent-to-agent communication.

## The Event Flow
1. **Input:** A physical sensor (or simulated spacebar tap) triggers a `sensor.touch` event.
2. **The Bus:** The Event Bus receives the event and broadcasts it.
3. **The Conductor:** The central Orchestrator reads the event, determines the intent, and publishes new task events.
4. **The Agents:** Specialized agents (Voice, Music, Memory) listen for tasks assigned to them and execute (e.g., triggering Text-to-Speech).

## The Components
* **Event Bus:** The central nervous system.
* **The Conductor:** The lightweight router. Contains no AI reasoning logic.
* **Specialized Agents:** Independent modules that perform exactly one job.
* **Hardware Abstraction:** The code does not care if the trigger is a keyboard spacebar or an Adafruit sensor.

```

### requirements.txt

```
streamlit>=1.36,<2
```

### main.py

```python
"""Minimal event-routing backend for The Conductor MVP."""
from __future__ import annotations

import asyncio
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Awaitable, Callable


@dataclass(frozen=True)
class Activity:
    source: str
    target: str
    event_type: str
    detail: str
    timestamp: str


ActivityListener = Callable[[Activity], Awaitable[None] | None]


class BaseAgent:
    def __init__(self, name: str) -> None:
        self.name = name

    async def on_event(self, event_type: str, data: object) -> str:
        raise NotImplementedError


class VoiceAgent(BaseAgent):
    async def on_event(self, event_type: str, data: object) -> str:
        if event_type != "SPEAK" or not isinstance(data, str):
            return f"Ignored {event_type}"
        return data


class MusicAgent(BaseAgent):
    async def on_event(self, event_type: str, data: object) -> str:
        if event_type != "PLAY" or not isinstance(data, str):
            return f"Ignored {event_type}"
        return f"Playing {data}"


class MemoryAgent(BaseAgent):
    """Fast, session-scoped working memory for recent orchestration events."""

    def __init__(self, name: str) -> None:
        super().__init__(name)
        self.working_memory: dict[str, object] = {
            "event_count": 0,
            "event_counts": {},
            "last_event": None,
        }

    async def on_event(self, event_type: str, data: object) -> str:
        if event_type != "STORE" or not isinstance(data, dict):
            return f"Ignored {event_type}"

        command_type = str(data.get("event_type", "UNKNOWN"))
        event_counts = self.working_memory["event_counts"]
        assert isinstance(event_counts, dict)

        event_counts[command_type] = event_counts.get(command_type, 0) + 1
        self.working_memory["event_count"] = int(self.working_memory["event_count"]) + 1
        self.working_memory["last_event"] = data.copy()

        return f"Stored event #{self.working_memory['event_count']}"


class Conductor:
    """Routes commands only; agents own the work."""

    def __init__(self, activity_listener: ActivityListener | None = None) -> None:
        self.agents: dict[str, BaseAgent] = {}
        self.activity_listener = activity_listener

    def register_agent(self, role: str, agent: BaseAgent) -> None:
        self.agents[role] = agent

    async def _emit(
        self,
        source: str,
        target: str,
        event_type: str,
        detail: str,
    ) -> None:
        if not self.activity_listener:
            return

        activity = Activity(
            source,
            target,
            event_type,
            detail,
            datetime.now(timezone.utc).strftime("%H:%M:%S"),
        )
        result = self.activity_listener(activity)
        if asyncio.iscoroutine(result):
            await result

    async def route_command(
        self,
        role: str,
        event_type: str,
        data: str,
        source: str = "Sensor",
    ) -> str:
        await self._emit(source, "Conductor", event_type, "Event received")

        agent = self.agents.get(role)
        if not agent:
            raise ValueError(f"No agent registered for role '{role}'")

        memory_agent = self.agents.get("memory")
        if isinstance(memory_agent, MemoryAgent):
            await self._emit("Conductor", memory_agent.name, "STORE", "Logging event")
            await memory_agent.on_event(
                "STORE",
                {
                    "source": source,
                    "role": role,
                    "event_type": event_type,
                    "data": data,
                },
            )

        await self._emit("Conductor", agent.name, event_type, "Routing command")
        result = await agent.on_event(event_type, data)
        await self._emit(agent.name, "Experience", "COMPLETE", result)
        return result

    async def run_score(
        self,
        score: dict[str, object],
        source: str = "Sensor",
    ) -> list[str]:
        """Run all valid score actions concurrently through normal routing."""
        actions = score.get("actions")
        if not isinstance(actions, list) or not actions:
            raise ValueError("A score must include a non-empty actions list")

        commands = []
        for action in actions:
            if not isinstance(action, dict):
                raise ValueError("Each score action must be a dictionary")

            role = action.get("role")
            event_type = action.get("event_type")
            data = action.get("data")

            if not all(isinstance(value, str) for value in (role, event_type, data)):
                raise ValueError(
                    "Each score action requires string role, event_type, and data values"
                )

            commands.append(self.route_command(role, event_type, data, source))

        return await asyncio.gather(*commands)


def build_demo_conductor(
    activity_listener: ActivityListener | None = None,
) -> Conductor:
    conductor = Conductor(activity_listener)
    conductor.register_agent("memory", MemoryAgent("Memory Agent"))
    conductor.register_agent("voice", VoiceAgent("Voice Agent"))
    conductor.register_agent("music", MusicAgent("Music Agent"))
    return conductor


async def main() -> None:
    conductor = build_demo_conductor()

    morning_focus_score = {
        "name": "Morning Focus",
        "trigger": "sensor.double_tap",
        "actions": [
            {
                "role": "voice",
                "event_type": "SPEAK",
                "data": "Good morning. Your focus session is ready.",
            },
            {
                "role": "music",
                "event_type": "PLAY",
                "data
[truncated — 205 more characters]
```

### dashboard.py

```python
"""Streamlit companion dashboard for The Conductor."""
from __future__ import annotations

import asyncio
import streamlit as st

from main import Activity, build_demo_conductor

st.set_page_config(page_title="The Conductor", page_icon="🎛️", layout="wide")

def setup_state() -> None:
    if "activities" not in st.session_state:
        st.session_state.activities = []
    if "last_message" not in st.session_state:
        st.session_state.last_message = "Ready for the next cue."

def record_activity(activity: Activity) -> None:
    st.session_state.activities.insert(0, activity)
    st.session_state.activities = st.session_state.activities[:12]

def run_double_tap() -> None:
    conductor = build_demo_conductor(record_activity)
    
    # The new Score we want the button to run!
    morning_focus_score = {
        "name": "Morning Focus",
        "trigger": "sensor.double_tap",
        "actions": [
            {
                "role": "voice",
                "event_type": "SPEAK",
                "data": "Good morning. Your focus session is ready.",
            },
            {
                "role": "music",
                "event_type": "PLAY",
                "data": "an atmospheric Ludovico Einaudi instrumental playlist",
            },
        ],
    }
    
    results = asyncio.run(conductor.run_score(morning_focus_score))
    st.session_state.last_message = f"Score complete: {results}"

def activity_html(activity: Activity) -> str:
    return f"""<div class="event">
      <span class="time">{activity.timestamp}</span>
      <div><b>{activity.source}</b><span class="arrow"> → </span><b>{activity.target}</b></div>
      <small>{activity.event_type} · {activity.detail}</small>
    </div>"""

setup_state()

st.markdown(
    """
<style>
  .stApp { background: #0b0d12; color: #edf0f7; }
  [data-testid="stHeader"] { background: transparent; }
  .title { font-size: 2.5rem; font-weight: 750; letter-spacing: -0.06em; margin-bottom: 0; }
  .subtitle, small { color: #969eae; }
  .eyebrow { color: #b8ff5a; font-weight: 700; font-size: .75rem; letter-spacing: .12em; }
  .agent-card { background: #161a23; border: 1px solid #272e3c; border-radius: 14px; padding: 18px; min-height: 140px; }
  .agent-icon { font-size: 1.6rem; }
  .online { color: #b8ff5a; font-size: .8rem; font-weight: 700; }
  .timeline { border-left: 1px solid #394252; margin: 14px 0 0 8px; padding-left: 18px; }
  .event { position: relative; background: #161a23; border: 1px solid #272e3c; border-radius: 10px; margin: 0 0 12px; padding: 11px 12px; animation: arrive .25s ease-out; }
  .event:before { content: ''; position: absolute; left: -24px; top: 18px; width: 10px; height: 10px; background: #b8ff5a; border-radius: 50%; box-shadow: 0 0 10px #b8ff5a; }
  .time { float: right; color: #788195; font-size: .72rem; }
  .arrow { color: #b8ff5a; }
  @keyframes arrive { from { opacity: 0; transform: translateY(-8px); } to { opacity: 1; transform: none; } }
  div.stButton > button { background: #b8ff5a; color: #101408; border: 0; border-radius: 10px; font-weight: 800; padding: .65rem 1rem; }
</style>
""",
    unsafe_allow_html=True,
)

left, right = st.columns([1.65, 1], gap="large")

with left:
    st.markdown('<div class="eyebrow">LIVE EXPERIENCE ENGINE · V0.1</div>', unsafe_allow_html=True)
    st.markdown(
        '<div class="title">The Conductor</div><div class="subtitle">A tactile orchestra for your day.</div>',
        unsafe_allow_html=True,
    )

    st.write("")
    if st.button("Simulate double tap"):
        run_double_tap()

    st.caption(f"Last output: {st.session_state.last_message}")
    st.write("")
    st.subheader("Agent status")

    cards = [
        ("◉", "Sensor Agent", "Listening for double taps", "ONLINE"),
        ("✦", "Conductor", "Routing the score", "ONLINE"),
        ("♫", "Voice Agent", "Ready to speak", "ONLINE"),
        ("🎵", "Music Agent", "Ready to play", "ONLINE"),
    ]
    cols = st.columns(4)
    for col, (icon, title, detail, status) in zip(cols, cards):
        with col:
            st.markdown(
                f'<div class="agent-card"><div class="agent-icon">{icon}</div><br>'
                f"<b>{title}</b><br><small>{detail}</small><br><br>"
                f'<span class="online">● {status}</span></div>',
                unsafe_allow_html=True,
            )

with right:
    st.markdown('<div class="eyebrow">EVENT BUS</div>', unsafe_allow_html=True)
    st.subheader("Live Activity")

    if st.session_state.activities:
        events = "".join(activity_html(activity) for activity in st.session_state.activities)
        st.markdown(f'<div class="timeline">{events}</div>', unsafe_allow_html=True)
    else:
        st.info("Waiting for a sensor event.")

    if st.button("Clear timeline", key="clear"):
        st.session_state.activities = []
        st.rerun()
```

[1 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]