diff --git a/00-chat-loop/README.md b/00-chat-loop/README.md index 34137e8..7932e3c 100644 --- a/00-chat-loop/README.md +++ b/00-chat-loop/README.md @@ -49,21 +49,21 @@ class ChatLoop: async def run(self) -> None: """Run the interactive chat loop.""" - rprint( + self.console.print( Panel( Text("Welcome to my-bot!", style="bold cyan"), title="Chat", border_style="cyan", ) ) - rprint("Type 'quit' or 'exit' to end the session.\n") + self.console.print("Type 'quit' or 'exit' to end the session.\n") try: while True: user_input = await asyncio.to_thread(self.get_user_input) if user_input.lower() in ("quit", "exit", "q"): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") break if not user_input: @@ -73,10 +73,10 @@ class ChatLoop: response = await self.session.chat(user_input) self.display_agent_response(response) except Exception as e: - rprint(f"\n[bold red]Error:[/bold red] {e}\n") + self.console.print(f"\n[bold red]Error:[/bold red] {e}\n") except (KeyboardInterrupt, EOFError): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") ``` [src/core/agent.py](src/core/agent.py) diff --git a/00-chat-loop/pyproject.toml b/00-chat-loop/pyproject.toml index f4842c7..875f458 100644 --- a/00-chat-loop/pyproject.toml +++ b/00-chat-loop/pyproject.toml @@ -17,7 +17,7 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["mybot"] +packages = ["src/mybot"] [project.scripts] my-bot = "mybot.cli.main:app" \ No newline at end of file diff --git a/00-chat-loop/src/mybot/cli/chat.py b/00-chat-loop/src/mybot/cli/chat.py index 5983859..d252576 100644 --- a/00-chat-loop/src/mybot/cli/chat.py +++ b/00-chat-loop/src/mybot/cli/chat.py @@ -5,7 +5,6 @@ import asyncio import typer from rich.console import Console from rich.panel import Panel -from rich import print as rprint from rich.prompt import Prompt from rich.text import Text @@ -38,28 +37,28 @@ class ChatLoop: def display_agent_response(self, content: str) -> None: """Display agent response with styled prefix.""" - prefix = Text(f"{self.agent.agent_def.id}: ", style="green") + prefix = Text(f"{self.agent_def.id}: ", style="green") self.console.print(prefix, end="") self.console.print(content) async def run(self) -> None: """Run the interactive chat loop.""" - rprint( + self.console.print( Panel( Text("Welcome to my-bot!", style="bold cyan"), title="Chat", border_style="cyan", ) ) - rprint("Type 'quit' or 'exit' to end the session.\n") + self.console.print("Type 'quit' or 'exit' to end the session.\n") try: while True: user_input = await asyncio.to_thread(self.get_user_input) if user_input.lower() in ("quit", "exit", "q"): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") break if not user_input: @@ -69,10 +68,10 @@ class ChatLoop: response = await self.session.chat(user_input) self.display_agent_response(response) except Exception as e: - rprint(f"\n[bold red]Error:[/bold red] {e}\n") + self.console.print(f"\n[bold red]Error:[/bold red] {e}\n") except (KeyboardInterrupt, EOFError): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None: diff --git a/01-tools/pyproject.toml b/01-tools/pyproject.toml index f4842c7..875f458 100644 --- a/01-tools/pyproject.toml +++ b/01-tools/pyproject.toml @@ -17,7 +17,7 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["mybot"] +packages = ["src/mybot"] [project.scripts] my-bot = "mybot.cli.main:app" \ No newline at end of file diff --git a/01-tools/src/mybot/cli/chat.py b/01-tools/src/mybot/cli/chat.py index 5983859..d252576 100644 --- a/01-tools/src/mybot/cli/chat.py +++ b/01-tools/src/mybot/cli/chat.py @@ -5,7 +5,6 @@ import asyncio import typer from rich.console import Console from rich.panel import Panel -from rich import print as rprint from rich.prompt import Prompt from rich.text import Text @@ -38,28 +37,28 @@ class ChatLoop: def display_agent_response(self, content: str) -> None: """Display agent response with styled prefix.""" - prefix = Text(f"{self.agent.agent_def.id}: ", style="green") + prefix = Text(f"{self.agent_def.id}: ", style="green") self.console.print(prefix, end="") self.console.print(content) async def run(self) -> None: """Run the interactive chat loop.""" - rprint( + self.console.print( Panel( Text("Welcome to my-bot!", style="bold cyan"), title="Chat", border_style="cyan", ) ) - rprint("Type 'quit' or 'exit' to end the session.\n") + self.console.print("Type 'quit' or 'exit' to end the session.\n") try: while True: user_input = await asyncio.to_thread(self.get_user_input) if user_input.lower() in ("quit", "exit", "q"): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") break if not user_input: @@ -69,10 +68,10 @@ class ChatLoop: response = await self.session.chat(user_input) self.display_agent_response(response) except Exception as e: - rprint(f"\n[bold red]Error:[/bold red] {e}\n") + self.console.print(f"\n[bold red]Error:[/bold red] {e}\n") except (KeyboardInterrupt, EOFError): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None: diff --git a/02-skills/pyproject.toml b/02-skills/pyproject.toml index f4842c7..875f458 100644 --- a/02-skills/pyproject.toml +++ b/02-skills/pyproject.toml @@ -17,7 +17,7 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["mybot"] +packages = ["src/mybot"] [project.scripts] my-bot = "mybot.cli.main:app" \ No newline at end of file diff --git a/02-skills/src/mybot/cli/chat.py b/02-skills/src/mybot/cli/chat.py index 5983859..d252576 100644 --- a/02-skills/src/mybot/cli/chat.py +++ b/02-skills/src/mybot/cli/chat.py @@ -5,7 +5,6 @@ import asyncio import typer from rich.console import Console from rich.panel import Panel -from rich import print as rprint from rich.prompt import Prompt from rich.text import Text @@ -38,28 +37,28 @@ class ChatLoop: def display_agent_response(self, content: str) -> None: """Display agent response with styled prefix.""" - prefix = Text(f"{self.agent.agent_def.id}: ", style="green") + prefix = Text(f"{self.agent_def.id}: ", style="green") self.console.print(prefix, end="") self.console.print(content) async def run(self) -> None: """Run the interactive chat loop.""" - rprint( + self.console.print( Panel( Text("Welcome to my-bot!", style="bold cyan"), title="Chat", border_style="cyan", ) ) - rprint("Type 'quit' or 'exit' to end the session.\n") + self.console.print("Type 'quit' or 'exit' to end the session.\n") try: while True: user_input = await asyncio.to_thread(self.get_user_input) if user_input.lower() in ("quit", "exit", "q"): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") break if not user_input: @@ -69,10 +68,10 @@ class ChatLoop: response = await self.session.chat(user_input) self.display_agent_response(response) except Exception as e: - rprint(f"\n[bold red]Error:[/bold red] {e}\n") + self.console.print(f"\n[bold red]Error:[/bold red] {e}\n") except (KeyboardInterrupt, EOFError): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None: diff --git a/03-persistence/pyproject.toml b/03-persistence/pyproject.toml index f4842c7..875f458 100644 --- a/03-persistence/pyproject.toml +++ b/03-persistence/pyproject.toml @@ -17,7 +17,7 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["mybot"] +packages = ["src/mybot"] [project.scripts] my-bot = "mybot.cli.main:app" \ No newline at end of file diff --git a/03-persistence/src/mybot/cli/chat.py b/03-persistence/src/mybot/cli/chat.py index 8d505c6..60488f5 100644 --- a/03-persistence/src/mybot/cli/chat.py +++ b/03-persistence/src/mybot/cli/chat.py @@ -5,7 +5,6 @@ import asyncio import typer from rich.console import Console from rich.panel import Panel -from rich import print as rprint from rich.prompt import Prompt from rich.text import Text @@ -38,28 +37,28 @@ class ChatLoop: def display_agent_response(self, content: str) -> None: """Display agent response with styled prefix.""" - prefix = Text(f"{self.agent.agent_def.id}: ", style="green") + prefix = Text(f"{self.agent_def.id}: ", style="green") self.console.print(prefix, end="") self.console.print(content) async def run(self) -> None: """Run the interactive chat loop.""" - rprint( + self.console.print( Panel( Text("Welcome to my-bot!", style="bold cyan"), title="Chat", border_style="cyan", ) ) - rprint("Type 'quit' or 'exit' to end the session.\n") + self.console.print("Type 'quit' or 'exit' to end the session.\n") try: while True: user_input = await asyncio.to_thread(self.get_user_input) if user_input.lower() in ("quit", "exit", "q"): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") break if not user_input: @@ -69,10 +68,10 @@ class ChatLoop: response = await self.session.chat(user_input) self.display_agent_response(response) except Exception as e: - rprint(f"\n[bold red]Error:[/bold red] {e}\n") + self.console.print(f"\n[bold red]Error:[/bold red] {e}\n") except (KeyboardInterrupt, EOFError): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") diff --git a/04-slash-commands/README.md b/04-slash-commands/README.md index e5361b5..961eeae 100644 --- a/04-slash-commands/README.md +++ b/04-slash-commands/README.md @@ -64,7 +64,7 @@ cmd_response = await self.session.command_registry.dispatch( user_input, self.session ) if cmd_response is not None: - rprint(cmd_response) + self.console.print(cmd_response) continue # Normal chat diff --git a/04-slash-commands/pyproject.toml b/04-slash-commands/pyproject.toml index f4842c7..875f458 100644 --- a/04-slash-commands/pyproject.toml +++ b/04-slash-commands/pyproject.toml @@ -17,7 +17,7 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["mybot"] +packages = ["src/mybot"] [project.scripts] my-bot = "mybot.cli.main:app" \ No newline at end of file diff --git a/04-slash-commands/src/mybot/cli/chat.py b/04-slash-commands/src/mybot/cli/chat.py index 4496431..5e46255 100644 --- a/04-slash-commands/src/mybot/cli/chat.py +++ b/04-slash-commands/src/mybot/cli/chat.py @@ -5,7 +5,6 @@ import asyncio import typer from rich.console import Console from rich.panel import Panel -from rich import print as rprint from rich.prompt import Prompt from rich.text import Text @@ -38,28 +37,28 @@ class ChatLoop: def display_agent_response(self, content: str) -> None: """Display agent response with styled prefix.""" - prefix = Text(f"{self.agent.agent_def.id}: ", style="green") + prefix = Text(f"{self.agent_def.id}: ", style="green") self.console.print(prefix, end="") self.console.print(content) async def run(self) -> None: """Run the interactive chat loop.""" - rprint( + self.console.print( Panel( Text("Welcome to my-bot!", style="bold cyan"), title="Chat", border_style="cyan", ) ) - rprint("Type '/help' for commands, 'quit' or 'exit' to end.\n") + self.console.print("Type '/help' for commands, 'quit' or 'exit' to end.\n") try: while True: user_input = await asyncio.to_thread(self.get_user_input) if user_input.lower() in ("quit", "exit", "q"): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") break if not user_input: @@ -71,17 +70,17 @@ class ChatLoop: user_input, self.session ) if cmd_response is not None: - rprint(cmd_response) + self.console.print(cmd_response) continue # Normal chat response = await self.session.chat(user_input) self.display_agent_response(response) except Exception as e: - rprint(f"\n[bold red]Error:[/bold red] {e}\n") + self.console.print(f"\n[bold red]Error:[/bold red] {e}\n") except (KeyboardInterrupt, EOFError): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") diff --git a/04-slash-commands/src/mybot/core/commands/handlers.py b/04-slash-commands/src/mybot/core/commands/handlers.py index 26ea3e9..7d4988c 100644 --- a/04-slash-commands/src/mybot/core/commands/handlers.py +++ b/04-slash-commands/src/mybot/core/commands/handlers.py @@ -25,7 +25,6 @@ class SessionCommand(Command): f"**Agent:** {session.agent.agent_def.name} (`{session.agent.agent_def.id}`)", f"**Created:** {created_str}", f"**Messages:** {len(session.state.messages)}", - f"**Source:** `{session.source}`", ] return "\n".join(lines) diff --git a/05-compaction/pyproject.toml b/05-compaction/pyproject.toml index 522ed40..2343d18 100644 --- a/05-compaction/pyproject.toml +++ b/05-compaction/pyproject.toml @@ -17,7 +17,7 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["mybot"] +packages = ["src/mybot"] [project.scripts] my-bot = "mybot.cli.main:app" diff --git a/05-compaction/src/mybot/cli/chat.py b/05-compaction/src/mybot/cli/chat.py index 1592d01..de66d1a 100644 --- a/05-compaction/src/mybot/cli/chat.py +++ b/05-compaction/src/mybot/cli/chat.py @@ -5,7 +5,6 @@ import asyncio import typer from rich.console import Console from rich.panel import Panel -from rich import print as rprint from rich.prompt import Prompt from rich.text import Text @@ -38,28 +37,28 @@ class ChatLoop: def display_agent_response(self, content: str) -> None: """Display agent response with styled prefix.""" - prefix = Text(f"{self.agent.agent_def.id}: ", style="green") + prefix = Text(f"{self.agent_def.id}: ", style="green") self.console.print(prefix, end="") self.console.print(content) async def run(self) -> None: """Run the interactive chat loop.""" - rprint( + self.console.print( Panel( Text("Welcome to my-bot!", style="bold cyan"), title="Chat", border_style="cyan", ) ) - rprint("Type '/help' for commands, 'quit' or 'exit' to end.\n") + self.console.print("Type '/help' for commands, 'quit' or 'exit' to end.\n") try: while True: user_input = await asyncio.to_thread(self.get_user_input) if user_input.lower() in ("quit", "exit", "q"): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") break if not user_input: @@ -71,17 +70,17 @@ class ChatLoop: user_input, self.session ) if cmd_response is not None: - rprint(cmd_response) + self.console.print(cmd_response) continue # Normal chat response = await self.session.chat(user_input) self.display_agent_response(response) except Exception as e: - rprint(f"\n[bold red]Error:[/bold red] {e}\n") + self.console.print(f"\n[bold red]Error:[/bold red] {e}\n") except (KeyboardInterrupt, EOFError): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None: diff --git a/05-compaction/src/mybot/core/commands/handlers.py b/05-compaction/src/mybot/core/commands/handlers.py index 406a559..93e562a 100644 --- a/05-compaction/src/mybot/core/commands/handlers.py +++ b/05-compaction/src/mybot/core/commands/handlers.py @@ -24,7 +24,6 @@ class SessionCommand(Command): f"**Agent:** {session.agent.agent_def.name} (`{session.agent.agent_def.id}`)", f"**Created:** {created_str}", f"**Messages:** {len(session.state.messages)}", - f"**Source:** `{session.source}`", ] return "\n".join(lines) diff --git a/06-web-tools/pyproject.toml b/06-web-tools/pyproject.toml index da52de4..1f1fc5e 100644 --- a/06-web-tools/pyproject.toml +++ b/06-web-tools/pyproject.toml @@ -19,7 +19,7 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["mybot"] +packages = ["src/mybot"] [project.scripts] my-bot = "mybot.cli.main:app" diff --git a/06-web-tools/src/mybot/cli/chat.py b/06-web-tools/src/mybot/cli/chat.py index 1592d01..de66d1a 100644 --- a/06-web-tools/src/mybot/cli/chat.py +++ b/06-web-tools/src/mybot/cli/chat.py @@ -5,7 +5,6 @@ import asyncio import typer from rich.console import Console from rich.panel import Panel -from rich import print as rprint from rich.prompt import Prompt from rich.text import Text @@ -38,28 +37,28 @@ class ChatLoop: def display_agent_response(self, content: str) -> None: """Display agent response with styled prefix.""" - prefix = Text(f"{self.agent.agent_def.id}: ", style="green") + prefix = Text(f"{self.agent_def.id}: ", style="green") self.console.print(prefix, end="") self.console.print(content) async def run(self) -> None: """Run the interactive chat loop.""" - rprint( + self.console.print( Panel( Text("Welcome to my-bot!", style="bold cyan"), title="Chat", border_style="cyan", ) ) - rprint("Type '/help' for commands, 'quit' or 'exit' to end.\n") + self.console.print("Type '/help' for commands, 'quit' or 'exit' to end.\n") try: while True: user_input = await asyncio.to_thread(self.get_user_input) if user_input.lower() in ("quit", "exit", "q"): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") break if not user_input: @@ -71,17 +70,17 @@ class ChatLoop: user_input, self.session ) if cmd_response is not None: - rprint(cmd_response) + self.console.print(cmd_response) continue # Normal chat response = await self.session.chat(user_input) self.display_agent_response(response) except Exception as e: - rprint(f"\n[bold red]Error:[/bold red] {e}\n") + self.console.print(f"\n[bold red]Error:[/bold red] {e}\n") except (KeyboardInterrupt, EOFError): - rprint("\n[bold yellow]Goodbye![/bold yellow]") + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None: diff --git a/06-web-tools/src/mybot/core/commands/handlers.py b/06-web-tools/src/mybot/core/commands/handlers.py index 7093a3a..ed4e4e4 100644 --- a/06-web-tools/src/mybot/core/commands/handlers.py +++ b/06-web-tools/src/mybot/core/commands/handlers.py @@ -25,7 +25,6 @@ class SessionCommand(Command): f"**Agent:** {session.agent.agent_def.name} (`{session.agent.agent_def.id}`)", f"**Created:** {created_str}", f"**Messages:** {len(session.state.messages)}", - f"**Source:** `{session.source}`", ] return "\n".join(lines) diff --git a/07-event-driven/README.md b/07-event-driven/README.md new file mode 100644 index 0000000..aa3dc84 --- /dev/null +++ b/07-event-driven/README.md @@ -0,0 +1,140 @@ +# Step 07: Event-Driven Architecture + +Replace direct `chat()` calls with event-driven architecture using EventBus and Workers for scalability. + +## What We Will Build + +``` +┌─────────────┐ ┌──────────────┐ +│ CLI │──InboundEvent────▶│ EventBus │ +│ │ │ │ +│ │◀──OutboundEvent───│ │ +└─────────────┘ └──────┬───────┘ + │ + ▼ + ┌──────────────┐ + │ AgentWorker │ + │ │ + │ - Executes │ + │ sessions │ + └──────────────┘ +``` + +**Key Components:** +- **EventBus** - Central pub/sub for event distribution +- **Event Types** - InboundEvent, OutboundEvent +- **Workers** - Background tasks that process events +- **AgentWorker** - Handles InboundEvent → executes agent session → emits OutboundEvent + +## Key Changes + +### 1. Event Types ([src/mybot/core/events.py](src/mybot/core/events.py)) + +```python +@dataclass +class InboundEvent(Event): + """Event for external work entering the system.""" + session_id: str + content: str + retry_count: int = 0 + +@dataclass +class OutboundEvent(Event): + """Event for agent responses.""" + session_id: str + content: str + error: str | None = None +``` + +### 2. EventBus ([src/mybot/core/eventbus.py](src/mybot/core/eventbus.py)) + +```python +class EventBus(Worker): + def __init__(self, context): + self._subscribers: dict[type[Event], list[Handler]] = defaultdict(list) + self._queue: asyncio.Queue[Event] = asyncio.Queue() + + async def publish(self, event: Event) -> None: + await self._queue.put(event) + + async def run(self) -> None: + while True: + event = await self._queue.get() + await self._dispatch(event) +``` + +### 3. AgentWorker ([src/mybot/server/agent_worker.py](src/mybot/server/agent_worker.py)) + +```python +class AgentWorker(SubscriberWorker): + def __init__(self, context): + self.context.eventbus.subscribe(InboundEvent, self.dispatch_event) + + async def dispatch_event(self, event: InboundEvent): + agent = Agent(agent_def, self.context) + session = agent.resume_session(event.session_id) + response = await session.chat(event.content) + + result = OutboundEvent( + session_id=event.session_id, + content=response, + ) + await self.context.eventbus.publish(result) +``` + +### 4. CLI Uses Events ([src/mybot/cli/chat.py](src/mybot/cli/chat.py)) + +```python +class ChatLoop: + def __init__(self, config: Config): + self.context = SharedContext(config) + self.workers = [self.context.eventbus, AgentWorker(self.context)] + self.context.eventbus.subscribe(OutboundEvent, self.handle_outbound_event) + + async def run(self): + for worker in self.workers: + worker.start() + + event = InboundEvent( + session_id=session_id, + content=user_input, + ) + await self.context.eventbus.publish(event) + + response = await self.response_queue.get() + self.display_agent_response(response.content) +``` + +### 5. SharedContext ([src/mybot/core/context.py](src/mybot/core/context.py)) + +```python +class SharedContext: + """Global shared state for the application.""" + + def __init__(self, config: Config): + self.config = config + self.history_store = HistoryStore.from_config(config) + self.agent_loader = AgentLoader.from_config(config) + self.skill_loader = SkillLoader.from_config(config) + self.command_registry = CommandRegistry.with_builtins() + self.eventbus = EventBus(self) +``` + +## How to Run + +```bash +cd 07-event-driven +uv run my-bot chat +``` + +Example interaction: +``` +You: What tools do you have? +default: I have access to read, write, and bash tools for file operations + and command execution, plus web_search and web_read tools for + internet access. +``` + +## What's Next + +Step 08 will add **Config Hot Reload** - automatically reload configuration when files change without restarting. diff --git a/07-event-driven/pyproject.toml b/07-event-driven/pyproject.toml new file mode 100644 index 0000000..1f1fc5e --- /dev/null +++ b/07-event-driven/pyproject.toml @@ -0,0 +1,25 @@ +[project] +name = "my-bot" +version = "0.1.0" +description = "Step 06: Web Tools - Access the Internet" +requires-python = ">=3.11" + +dependencies = [ + "litellm>=1.0.0", + "typer>=0.9.0", + "rich>=13.0.0", + "pydantic>=2.0.0", + "pyyaml>=6.0", + "httpx>=0.27.0", + "crawl4ai>=0.3.0", +] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/mybot"] + +[project.scripts] +my-bot = "mybot.cli.main:app" diff --git a/07-event-driven/src/mybot/cli/__init__.py b/07-event-driven/src/mybot/cli/__init__.py new file mode 100644 index 0000000..dc4a0ca --- /dev/null +++ b/07-event-driven/src/mybot/cli/__init__.py @@ -0,0 +1,5 @@ +"""CLI interface for my-bot.""" + +from mybot.cli.main import app + +__all__ = ["app"] diff --git a/07-event-driven/src/mybot/cli/chat.py b/07-event-driven/src/mybot/cli/chat.py new file mode 100644 index 0000000..7e8d6fa --- /dev/null +++ b/07-event-driven/src/mybot/cli/chat.py @@ -0,0 +1,116 @@ +"""Chat CLI command for interactive sessions with event-driven architecture.""" + +import asyncio + +import typer +from rich.console import Console +from rich.panel import Panel +from rich.prompt import Prompt +from rich.text import Text + +from mybot.core.agent import Agent +from mybot.core.context import SharedContext +from mybot.core.events import ( + OutboundEvent, + InboundEvent, +) +from mybot.server import ( + AgentWorker, + Worker, +) +from mybot.utils.config import Config + + +class ChatLoop: + """Interactive chat session using event-driven architecture.""" + + def __init__(self, config: Config, agent_id: str | None = None): + self.config = config + self.console = Console() + self.context = SharedContext(config=config) + + self.workers: list[Worker] = [ + self.context.eventbus, + AgentWorker(self.context), + ] + + self.response_queue: asyncio.Queue[OutboundEvent] = asyncio.Queue() + self.context.eventbus.subscribe(OutboundEvent, self.handle_outbound_event) + + agent_id = agent_id or config.default_agent + self.agent_def = self.context.agent_loader.load(agent_id) + + async def handle_outbound_event(self, event: OutboundEvent) -> None: + """Handle outbound events by adding to response queue.""" + await self.response_queue.put(event) + + def get_user_input(self) -> str: + """Get user input with styled prompt.""" + prompt_text = Text("You", style="cyan") + user_input = Prompt.ask(prompt_text, console=self.console) + return user_input.strip() + + def display_agent_response(self, content: str) -> None: + """Display agent response with styled prefix.""" + prefix = Text(f"{self.agent_def.id}: ", style="green") + + self.console.print(prefix, end="") + self.console.print(content) + + async def run(self) -> None: + """Run the interactive chat loop.""" + self.console.print( + Panel( + Text("Welcome to my-bot!", style="bold cyan"), + title="Chat", + border_style="cyan", + ) + ) + self.console.print("Type '/help' for commands, 'quit' or 'exit' to end.\n") + + for worker in self.workers: + worker.start() + + session_id = ( + Agent(self.agent_def, self.context).new_session().session_id + ) + + try: + while True: + user_input = await asyncio.to_thread(self.get_user_input) + if user_input.lower() in ("quit", "exit", "q"): + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") + break + + if not user_input: + continue + + event = InboundEvent( + session_id=session_id, + content=user_input, + ) + await self.context.eventbus.publish(event) + + try: + response = await asyncio.wait_for( + self.response_queue.get(), timeout=60.0 + ) + + self.display_agent_response(response.content) + except asyncio.TimeoutError: + self.console.print("[red]Agent response timed out[/red]") + self.console.print() + + except (KeyboardInterrupt, EOFError): + self.console.print("\n[bold yellow]Goodbye![/bold yellow]") + finally: + for worker in self.workers: + await worker.stop() + + +def chat_command(ctx: typer.Context, agent_id: str | None = None) -> None: + """Start interactive chat session.""" + config = ctx.obj.get("config") + + chat_loop = ChatLoop(config, agent_id=agent_id) + asyncio.run(chat_loop.run()) diff --git a/07-event-driven/src/mybot/cli/main.py b/07-event-driven/src/mybot/cli/main.py new file mode 100644 index 0000000..35d9a61 --- /dev/null +++ b/07-event-driven/src/mybot/cli/main.py @@ -0,0 +1,73 @@ +"""CLI interface for my-bot using Typer.""" + +from pathlib import Path +from typing import Annotated + +import typer +from rich.console import Console + +from mybot.cli.chat import chat_command +from mybot.utils.config import Config + +app = typer.Typer( + name="my-bot", + help="my-bot: Personal AI Assistant", + no_args_is_help=True, + add_completion=True, +) + +console = Console() + + +def workspace_callback(ctx: typer.Context, workspace: str) -> Path: + """Store workspace path in context for later use.""" + ctx.ensure_object(dict) + ctx.obj["workspace"] = Path(workspace) + return Path(workspace) + + +@app.callback() +def main( + ctx: typer.Context, + workspace: str = typer.Option( + "../default_workspace", + "--workspace", + "-w", + help="Path to workspace directory", + callback=workspace_callback, + ), +) -> None: + """Configuration is loaded from workspace/config.user.yaml by default.""" + workspace_path = ctx.obj["workspace"] + config_file = workspace_path / "config.user.yaml" + + if not config_file.exists(): + console.print(f"[yellow]No configuration found at {config_file}[/yellow]") + raise typer.Exit(1) + + try: + cfg = Config.load(workspace_path) + ctx.obj["config"] = cfg + except Exception as e: + console.print(f"[red]Error loading config: {e}[/red]") + raise typer.Exit(1) + + +@app.command("chat") +def chat( + ctx: typer.Context, + agent: Annotated[ + str | None, + typer.Option( + "--agent", + "-a", + help="Agent ID to use (overrides default_agent from config)", + ), + ] = None, +) -> None: + """Start interactive chat session.""" + chat_command(ctx, agent_id=agent) + + +if __name__ == "__main__": + app() diff --git a/07-event-driven/src/mybot/core/__init__.py b/07-event-driven/src/mybot/core/__init__.py new file mode 100644 index 0000000..f263fbb --- /dev/null +++ b/07-event-driven/src/mybot/core/__init__.py @@ -0,0 +1,6 @@ +"""Core agent components.""" + +from .agent import Agent, AgentSession +from .agent_loader import AgentDef, AgentLoader + +__all__ = ["Agent", "AgentSession", "AgentDef", "AgentLoader"] diff --git a/07-event-driven/src/mybot/core/agent.py b/07-event-driven/src/mybot/core/agent.py new file mode 100644 index 0000000..610be08 --- /dev/null +++ b/07-event-driven/src/mybot/core/agent.py @@ -0,0 +1,224 @@ +import uuid +import json +import asyncio +from dataclasses import dataclass, field +from datetime import datetime +from typing import TYPE_CHECKING + +from mybot.core.context_guard import ContextGuard +from mybot.core.session_state import SessionState +from mybot.provider.llm import LLMProvider +from mybot.tools.registry import ToolRegistry +from mybot.tools.skill_tool import create_skill_tool +from mybot.tools.websearch_tool import create_websearch_tool +from mybot.tools.webread_tool import create_webread_tool + +from litellm.types.completion import ( + ChatCompletionMessageParam as Message, + ChatCompletionMessageToolCallParam, +) + +if TYPE_CHECKING: + from mybot.core.context import SharedContext + from mybot.core.agent_loader import AgentDef + from mybot.provider.llm import LLMToolCall + + +class Agent: + """A configured agent that creates and manages conversation sessions.""" + + def __init__(self, agent_def: "AgentDef", context: "SharedContext") -> None: + self.agent_def = agent_def + self.context = context + self.llm = LLMProvider.from_config(agent_def.llm) + + def _build_tools(self) -> ToolRegistry: + """Build a ToolRegistry with tools appropriate for the session.""" + registry = ToolRegistry.with_builtins() + + # Register skill tool if allowed + if self.agent_def.allow_skills: + skill_tool = create_skill_tool(self.context.skill_loader) + if skill_tool: + registry.register(skill_tool) + + websearch_tool = create_websearch_tool(self.context) + if websearch_tool: + registry.register(websearch_tool) + + webread_tool = create_webread_tool(self.context) + if webread_tool: + registry.register(webread_tool) + + return registry + + def _get_token_threshold(self) -> int: + """Get token threshold based on model's context window.""" + # Default to 80% of 200k context + return 160000 + + def new_session( + self, + session_id: str | None = None, + ) -> "AgentSession": + """Create a new conversation session.""" + session_id = session_id or str(uuid.uuid4()) + tools = self._build_tools() + + # Create context guard for this session + context_guard = ContextGuard( + shared_context=self.context, + token_threshold=self._get_token_threshold(), + ) + + state = SessionState( + session_id=session_id, + agent=self, + messages=[], + shared_context=self.context, + ) + + session = AgentSession( + agent=self, + state=state, + context_guard=context_guard, + tools=tools, + ) + + self.context.history_store.create_session(self.agent_def.id, session_id) + return session + + def resume_session(self, session_id: str) -> "AgentSession": + """Load an existing conversation session.""" + session_query = [ + session + for session in self.context.history_store.list_sessions() + if session.id == session_id + ] + if not session_query: + raise ValueError(f"Session not found: {session_id}") + + session_info = session_query[0] + + # Get all messages (no max_history limit) + history_messages = self.context.history_store.get_messages(session_id) + + # Convert HistoryMessage to litellm Message format + messages: list[Message] = [msg.to_message() for msg in history_messages] + + # Build tools for resumed session + tools = self._build_tools() + + # Create context guard + context_guard = ContextGuard( + shared_context=self.context, + token_threshold=self._get_token_threshold(), + ) + + # Create SessionState with loaded messages + state = SessionState( + session_id=session_info.id, + agent=self, + messages=messages, + shared_context=self.context, + ) + + return AgentSession( + agent=self, + state=state, + context_guard=context_guard, + tools=tools, + ) + + +@dataclass +class AgentSession: + """Chat orchestrator - operates on swappable SessionState.""" + + agent: Agent + state: SessionState + context_guard: ContextGuard + tools: ToolRegistry + started_at: datetime = field(default_factory=datetime.now) + + @property + def session_id(self) -> str: + """Delegate to state.""" + return self.state.session_id + + @property + def shared_context(self) -> "SharedContext": + """Delegate to state.""" + return self.state.shared_context + + async def chat(self, message: str) -> str: + """Send a message to the LLM and get a response.""" + user_msg: Message = {"role": "user", "content": message} + self.state.add_message(user_msg) + + tool_schemas = self.tools.get_tool_schemas() + + while True: + messages = self.state.build_messages() + self.state = await self.context_guard.check_and_compact(self.state) + content, tool_calls = await self.agent.llm.chat(messages, tool_schemas) + + tool_call_dicts: list[ChatCompletionMessageToolCallParam] = [ + { + "id": tc.id, + "type": "function", + "function": {"name": tc.name, "arguments": tc.arguments}, + } + for tc in tool_calls + ] + assistant_msg: Message = { + "role": "assistant", + "content": content, + "tool_calls": tool_call_dicts, + } + + self.state.add_message(assistant_msg) + + if not tool_calls: + break + + await self._handle_tool_calls(tool_calls) + + continue + + return content + + async def _handle_tool_calls( + self, + tool_calls: list["LLMToolCall"], + ) -> None: + """Handle tool calls from the LLM response.""" + tool_call_results = await asyncio.gather( + *[self._execute_tool_call(tool_call) for tool_call in tool_calls] + ) + + for tool_call, result in zip(tool_calls, tool_call_results): + tool_msg: Message = { + "role": "tool", + "content": result, + "tool_call_id": tool_call.id, + } + self.state.add_message(tool_msg) + + async def _execute_tool_call( + self, + tool_call: "LLMToolCall", + ) -> str: + """Execute a single tool call.""" + # Extract key arguments + try: + args = json.loads(tool_call.arguments) + except json.JSONDecodeError: + args = {} + + try: + result = await self.tools.execute_tool(tool_call.name, session=self, **args) + except Exception as e: + result = f"Error executing tool: {e}" + + return result diff --git a/07-event-driven/src/mybot/core/agent_loader.py b/07-event-driven/src/mybot/core/agent_loader.py new file mode 100644 index 0000000..082f62a --- /dev/null +++ b/07-event-driven/src/mybot/core/agent_loader.py @@ -0,0 +1,77 @@ +"""Agent definition loader.""" + +from typing import Any + +from pydantic import BaseModel, ValidationError + +from mybot.utils.config import Config, LLMConfig +from mybot.utils.def_loader import ( + DefNotFoundError, + InvalidDefError, + parse_definition, +) + + +class AgentDef(BaseModel): + """Loaded agent definition with merged settings.""" + + id: str + name: str + description: str = "" + agent_md: str + llm: LLMConfig + allow_skills: bool = False + + +class AgentLoader: + """Loads agent definitions from AGENT.md files.""" + + @staticmethod + def from_config(config: Config) -> "AgentLoader": + return AgentLoader(config) + + def __init__(self, config: Config): + """Initialize AgentLoader.""" + self.config = config + + def load(self, agent_id: str) -> AgentDef: + """Load agent by ID.""" + agent_file = self.config.agents_path / agent_id / "AGENT.md" + if not agent_file.exists(): + raise DefNotFoundError("agent", agent_id) + + try: + content = agent_file.read_text() + agent_def = parse_definition(content, agent_id, self._parse_agent_def) + except InvalidDefError: + raise + except Exception as e: + raise InvalidDefError("agent", agent_id, str(e)) + + return agent_def + + def _parse_agent_def( + self, def_id: str, frontmatter: dict[str, Any], body: str + ) -> AgentDef: + """Parse agent definition from frontmatter (callback for parse_definition).""" + llm_overrides = frontmatter.get("llm") + merged_llm = self._merge_llm_config(llm_overrides) + + try: + return AgentDef( + id=def_id, + name=frontmatter["name"], # type: ignore[misc] + description=frontmatter.get("description", ""), + agent_md=body.strip(), + llm=merged_llm, + allow_skills=frontmatter.get("allow_skills", False), + ) + except ValidationError as e: + raise InvalidDefError("agent", def_id, str(e)) + + def _merge_llm_config(self, agent_llm: dict[str, Any] | None) -> LLMConfig: + """Deep merge agent's llm config with global defaults.""" + base = self.config.llm.model_dump() + if agent_llm: + base = {**base, **agent_llm} + return LLMConfig(**base) diff --git a/07-event-driven/src/mybot/core/commands/__init__.py b/07-event-driven/src/mybot/core/commands/__init__.py new file mode 100644 index 0000000..71b0ea7 --- /dev/null +++ b/07-event-driven/src/mybot/core/commands/__init__.py @@ -0,0 +1,6 @@ +"""Slash commands module.""" + +from mybot.core.commands.base import Command +from mybot.core.commands.registry import CommandRegistry + +__all__ = ["Command", "CommandRegistry"] diff --git a/07-event-driven/src/mybot/core/commands/base.py b/07-event-driven/src/mybot/core/commands/base.py new file mode 100644 index 0000000..588f3d2 --- /dev/null +++ b/07-event-driven/src/mybot/core/commands/base.py @@ -0,0 +1,20 @@ +"""Base classes for slash commands.""" + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + + +class Command(ABC): + """Base class for slash commands.""" + + name: str + aliases: list[str] = [] + description: str = "" + + @abstractmethod + async def execute(self, args: str, session: "AgentSession") -> str: + """Execute the command and return response string.""" + pass diff --git a/07-event-driven/src/mybot/core/commands/handlers.py b/07-event-driven/src/mybot/core/commands/handlers.py new file mode 100644 index 0000000..ed4e4e4 --- /dev/null +++ b/07-event-driven/src/mybot/core/commands/handlers.py @@ -0,0 +1,108 @@ +"""Built-in slash command handlers.""" + +from typing import TYPE_CHECKING + +from mybot.core.commands.base import Command + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + + +class SessionCommand(Command): + """Show current session details.""" + + name = "session" + description = "Show current session details" + + async def execute(self, args: str, session: "AgentSession") -> str: + info = session.agent.history_store.history_store.get_session_info(session.session_id) + + # Handle case where session not found in index + created_str = info.created_at if info else "Unknown" + + lines = [ + f"**Session ID:** `{session.session_id}`", + f"**Agent:** {session.agent.agent_def.name} (`{session.agent.agent_def.id}`)", + f"**Created:** {created_str}", + f"**Messages:** {len(session.state.messages)}", + ] + return "\n".join(lines) + + +class HelpCommand(Command): + """Show available commands.""" + + name = "help" + aliases = ["?"] + description = "Show available commands" + + async def execute(self, args: str, session: "AgentSession") -> str: + lines = ["**Available Commands:**"] + for cmd in session.command_registry.list_commands(): + names = [f"/{cmd.name}"] + [f"/{a}" for a in cmd.aliases] + lines.append(f"{', '.join(names)} - {cmd.description}") + return "\n".join(lines) + + +class CompactCommand(Command): + """Trigger manual context compaction.""" + + name = "compact" + description = "Compact conversation context manually" + + async def execute(self, args: str, session: "AgentSession") -> str: + # Force compaction regardless of threshold + await session.context_guard._compact_messages(session.state) + msg_count = len(session.state.messages) + return f"✓ Context compacted. {msg_count} messages retained." + + +class ContextCommand(Command): + """Show session context information.""" + + name = "context" + description = "Show session context information" + + async def execute(self, args: str, session: "AgentSession") -> str: + token_count = session.context_guard.estimate_tokens(session.state) + threshold = session.context_guard.token_threshold + usage_pct = (token_count / threshold) * 100 if threshold > 0 else 0 + + lines = [ + f"**Messages:** {len(session.state.messages)}", + f"**Tokens:** {token_count:,} ({usage_pct:.1f}% of {threshold:,} threshold)", + ] + return "\n".join(lines) + + +class SkillsCommand(Command): + """List all skills or show skill details.""" + + name = "skills" + description = "List all skills or show skill details" + + async def execute(self, args: str, session: "AgentSession") -> str: + if not args: + skills = session.agent.skill_loader.discover_skills() + if not skills: + return "No skills configured." + + lines = ["**Skills:**"] + for skill in skills: + lines.append(f"- `{skill.id}`: {skill.description}") + return "\n".join(lines) + + # Show specific skill details + skill_id = args.strip() + try: + skill = session.agent.skill_loader.load_skill(skill_id) + except FileNotFoundError: + return f"✗ Skill `{skill_id}` not found." + + lines = [ + f"**Skill:** `{skill.id}`", + f"**Name:** {skill.name}", + f"**Description:** {skill.description}", + f"\n---\n\n**SKILL.md:**\n```\n{skill.content}\n```", + ] + return "\n".join(lines) diff --git a/07-event-driven/src/mybot/core/commands/registry.py b/07-event-driven/src/mybot/core/commands/registry.py new file mode 100644 index 0000000..d388a3d --- /dev/null +++ b/07-event-driven/src/mybot/core/commands/registry.py @@ -0,0 +1,76 @@ +"""Command registry for managing slash commands.""" + +from typing import TYPE_CHECKING + +from mybot.core.commands.base import Command + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + + +class CommandRegistry: + """Registry for slash commands.""" + + def __init__(self) -> None: + self._commands: dict[str, Command] = {} + + def register(self, cmd: Command) -> None: + """Register a command and its aliases.""" + self._commands[cmd.name] = cmd + for alias in cmd.aliases: + self._commands[alias] = cmd + + def list_commands(self) -> list[Command]: + """Return list of unique commands (deduplicated by name).""" + seen = set() + commands = [] + for cmd in self._commands.values(): + if cmd.name not in seen: + seen.add(cmd.name) + commands.append(cmd) + return commands + + def resolve(self, input: str) -> tuple[Command, str] | None: + """Parse input and return (command, args) if it matches.""" + if not input.startswith("/"): + return None + + parts = input[1:].split(None, 1) + if not parts: + return None + + cmd_name = parts[0].lower() + args = parts[1] if len(parts) > 1 else "" + + cmd = self._commands.get(cmd_name) + if cmd: + return (cmd, args) + return None + + async def dispatch(self, input: str, session: "AgentSession") -> str | None: + """Parse and execute a slash command.""" + resolved = self.resolve(input) + if not resolved: + return None + + cmd, args = resolved + return await cmd.execute(args, session) + + @classmethod + def with_builtins(cls) -> "CommandRegistry": + """Create registry with built-in commands registered.""" + from mybot.core.commands.handlers import ( + HelpCommand, + SkillsCommand, + SessionCommand, + CompactCommand, + ContextCommand, + ) + + registry = cls() + registry.register(HelpCommand()) + registry.register(SkillsCommand()) + registry.register(SessionCommand()) + registry.register(CompactCommand()) + registry.register(ContextCommand()) + return registry diff --git a/07-event-driven/src/mybot/core/context.py b/07-event-driven/src/mybot/core/context.py new file mode 100644 index 0000000..7c26bd2 --- /dev/null +++ b/07-event-driven/src/mybot/core/context.py @@ -0,0 +1,25 @@ +from mybot.core.agent_loader import AgentLoader +from mybot.core.commands.registry import CommandRegistry +from mybot.core.history import HistoryStore +from mybot.core.skill_loader import SkillLoader +from mybot.core.eventbus import EventBus +from mybot.utils.config import Config + + +class SharedContext: + """Global shared state for the application.""" + + config: Config + history_store: HistoryStore + agent_loader: AgentLoader + skill_loader: SkillLoader + command_registry: CommandRegistry + eventbus: EventBus + + def __init__(self, config: Config) -> None: + self.config = config + self.history_store = HistoryStore.from_config(config) + self.agent_loader = AgentLoader.from_config(config) + self.skill_loader = SkillLoader.from_config(config) + self.command_registry = CommandRegistry.with_builtins() + self.eventbus = EventBus(self) diff --git a/07-event-driven/src/mybot/core/context_guard.py b/07-event-driven/src/mybot/core/context_guard.py new file mode 100644 index 0000000..6ccd828 --- /dev/null +++ b/07-event-driven/src/mybot/core/context_guard.py @@ -0,0 +1,145 @@ +"""Context guard for proactive context window management.""" + +from dataclasses import dataclass +from typing import TYPE_CHECKING, cast + +from litellm import token_counter +from litellm.types.completion import ( + ChatCompletionMessageParam as Message, + ChatCompletionAssistantMessageParam, + ChatCompletionToolMessageParam, +) + +from mybot.core.session_state import SessionState + +if TYPE_CHECKING: + from mybot.core.context import SharedContext + + +# Default max size for tool result content before truncation +MAX_TOOL_RESULT_CHARS = 10000 + + +@dataclass +class ContextGuard: + """Manages context window size with proactive compaction.""" + + shared_context: "SharedContext" + token_threshold: int = 160000 # 80% of 200k context + max_tool_result_chars: int = MAX_TOOL_RESULT_CHARS + + def estimate_tokens(self, state: "SessionState") -> int: + """Estimate token count for session state.""" + if not state.messages: + return 0 + return token_counter( + model=state.agent.agent_def.llm.model, messages=state.build_messages() + ) + + async def check_and_compact( + self, + state: "SessionState", + ) -> "SessionState": + """Check token count, compact if needed.""" + token_count = self.estimate_tokens(state) + + if token_count < self.token_threshold: + return state + + # First try truncating large tool results + state.messages = self._truncate_large_tool_results(state.messages) + token_count = self.estimate_tokens(state) + + if token_count < self.token_threshold: + return state + + # If still over threshold, compact via summarization + return await self._compact_messages(state) + + def _compress_message_count(self, state: "SessionState") -> int: + """Calculate how many messages to compress.""" + keep_count = max(4, int(len(state.messages) * 0.2)) + compress_count = max(2, int(len(state.messages) * 0.5)) + return min(compress_count, len(state.messages) - keep_count) + + def _truncate_large_tool_results(self, messages: list[Message]) -> list[Message]: + """Truncate oversized tool results to reduce context size.""" + result: list[Message] = [] + for msg in messages: + if msg.get("role") == "tool": + content = msg.get("content", "") + if ( + isinstance(content, str) + and len(content) > self.max_tool_result_chars + ): + original_size = len(content) + truncated = content[: self.max_tool_result_chars] + truncated_content = ( + f"{truncated}\n\n" + f"[Truncated - original size: {original_size} chars]" + ) + + msg = cast(ChatCompletionToolMessageParam, {**msg, "content": truncated_content}) + + result.append(msg) + return result + + def _serialize_messages_for_summary(self, messages: list[Message]) -> str: + """Serialize messages to plain text for summarization.""" + lines = [] + for msg in messages: + role = msg.get("role", "unknown") + content = msg.get("content", "") + # Handle tool calls in assistant messages + if role == "assistant" and msg.get("tool_calls"): + tool_names = [ + tc.get("function", {}).get("name", "unknown") + for tc in (cast(ChatCompletionAssistantMessageParam, msg)).get( + "tool_calls", [] + ) + ] + lines.append( + f"ASSISTANT: [used tools: {', '.join(tool_names)}] {content}" + ) + else: + lines.append(f"{role.upper()}: {content}") + return "\n".join(lines) + + async def _compact_messages( + self, + state: "SessionState", + ) -> "SessionState": + """Compact history by summarizing older messages.""" + compress_count = self._compress_message_count(state) + + old_messages = state.messages[:compress_count] + old_text = self._serialize_messages_for_summary(old_messages) + + summary_prompt = f"""Summarize the conversation so far. Keep it factual and concise. Focus on key decisions, facts, and user preferences discovered: + +{old_text}""" + + response, _ = await state.agent.llm.chat( + [{"role": "user", "content": summary_prompt}], + [], # No tools needed + ) + + # Build compacted message list + messages: list[Message] = [] + messages.append( + { + "role": "user", + "content": f"[Previous conversation summary]\n{response}", + } + ) + messages.append( + { + "role": "assistant", + "content": "Understood, I have the context.", + } + ) + messages.extend(state.messages[compress_count:]) + + # Update state in place + state.messages = messages + return state diff --git a/07-event-driven/src/mybot/core/eventbus.py b/07-event-driven/src/mybot/core/eventbus.py new file mode 100644 index 0000000..64a90e3 --- /dev/null +++ b/07-event-driven/src/mybot/core/eventbus.py @@ -0,0 +1,79 @@ +"""Central event bus for pub/sub event distribution.""" + +import asyncio +import logging +from collections import defaultdict +from typing import Awaitable, Callable, TypeVar + +from mybot.server.worker import Worker + +from .events import Event + +logger = logging.getLogger(__name__) + +E = TypeVar("E", bound=Event) +Handler = Callable[[Event], Awaitable[None]] + + +class EventBus(Worker): + """Central event bus with subscription support and async dispatch.""" + + def __init__(self, context) -> None: + super().__init__(context) + self.context = context + self._subscribers: dict[type[Event], list[Handler]] = defaultdict(list) + self._queue: asyncio.Queue[Event] = asyncio.Queue() + + def subscribe( + self, event_class: type[E], handler: Callable[[E], Awaitable[None]] + ) -> None: + """Subscribe a handler to an event class.""" + self._subscribers[event_class].append(handler) + logger.debug(f"Subscribed handler to {event_class.__name__} events") + + def unsubscribe(self, handler: Handler) -> None: + """Remove a handler from all subscriptions.""" + for event_class in self._subscribers: + if handler in self._subscribers[event_class]: + self._subscribers[event_class].remove(handler) + logger.debug(f"Unsubscribed handler from {event_class.__name__} events") + + async def publish(self, event: Event) -> None: + """Publish an event to the internal queue (non-blocking).""" + await self._queue.put(event) + logger.debug(f"Queued {event.__class__.__name__} event") + + async def run(self) -> None: + """Process events from queue, starting with recovery.""" + logger.info("EventBus started") + + # Process events from queue + try: + while True: + event = await self._queue.get() + try: + await self._dispatch(event) + except Exception as e: + logger.error(f"Error dispatching event: {e}") + finally: + self._queue.task_done() + except asyncio.CancelledError: + logger.info("EventBus stopping...") + raise + + async def _dispatch(self, event: Event) -> None: + """Persist if OUTBOUND, then notify subscribers.""" + await self._notify_subscribers(event) + logger.debug(f"Dispatched {event.__class__.__name__} event") + + async def _notify_subscribers(self, event: Event) -> None: + """Notify all subscribers of an event (waits for all handlers to complete).""" + handlers = self._subscribers.get(type(event), []) + if not handlers: + return + + tasks = [handler(event) for handler in handlers] + results = await asyncio.gather(*tasks, return_exceptions=True) + for result in results: + if isinstance(result, Exception): + logger.error(f"Error in event handler: {result}") \ No newline at end of file diff --git a/07-event-driven/src/mybot/core/events.py b/07-event-driven/src/mybot/core/events.py new file mode 100644 index 0000000..a72b404 --- /dev/null +++ b/07-event-driven/src/mybot/core/events.py @@ -0,0 +1,71 @@ +"""Event types and data classes for the event bus.""" + +import time +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class Event: + """Base class for all typed events.""" + + session_id: str + content: str + timestamp: float = field(default_factory=time.time) + + def to_dict(self) -> dict[str, Any]: + """Serialize event to dictionary, including type.""" + result: dict[str, Any] = {"type": self.__class__.__name__} + for field_name in self.__dataclass_fields__: + value = getattr(self, field_name) + result[field_name] = value + + return result + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "Event": + """Deserialize event from dictionary.""" + kwargs = {} + for k, v in data.items(): + if k == "type": + continue + elif k in cls.__dataclass_fields__: + kwargs[k] = v + return cls(**kwargs) + + +@dataclass +class InboundEvent(Event): + """Event for external work entering the system (platforms, cron, retry).""" + + retry_count: int = 0 + + +@dataclass +class OutboundEvent(Event): + """Event for agent responses to deliver to platforms.""" + + error: str | None = None + + +# Registry mapping event class names to event classes +_EVENT_CLASSES: dict[str, type[Event]] = { + "InboundEvent": InboundEvent, + "OutboundEvent": OutboundEvent, +} + + +def serialize_event(event: Event) -> dict[str, Any]: + """Serialize any event type to dict.""" + return event.to_dict() + + +def deserialize_event(data: dict[str, Any]) -> Event: + """Deserialize dict to appropriate event type.""" + event_type: str = data.get("type", "") + + event_class = _EVENT_CLASSES.get(event_type) + if event_class is None: + raise ValueError(f"Unknown event type: {event_type}") + + return event_class.from_dict(data) diff --git a/07-event-driven/src/mybot/core/history.py b/07-event-driven/src/mybot/core/history.py new file mode 100644 index 0000000..05d172a --- /dev/null +++ b/07-event-driven/src/mybot/core/history.py @@ -0,0 +1,213 @@ +"""JSONL file-based conversation history backend.""" + +from datetime import datetime +from pathlib import Path +from typing import Any, Literal, TYPE_CHECKING + +from pydantic import BaseModel, Field, field_validator + +from litellm.types.completion import ChatCompletionMessageParam as Message + +if TYPE_CHECKING: + from mybot.utils.config import Config + + +def _now_iso() -> str: + """Return current datetime as ISO format string.""" + return datetime.now().isoformat() + + +class HistorySession(BaseModel): + """Session metadata - stored in index.jsonl.""" + + id: str + agent_id: str + title: str | None = None + message_count: int = 0 + created_at: str + updated_at: str + + +class HistoryMessage(BaseModel): + """Single message - stored in session.jsonl.""" + + timestamp: str = Field(default_factory=_now_iso) + role: Literal["user", "assistant", "system", "tool"] + content: str + tool_calls: list[dict[str, Any]] | None = None + tool_call_id: str | None = None + + @classmethod + def from_message(cls, message: Message) -> "HistoryMessage": + """Create HistoryMessage from litellm Message format.""" + tool_calls = None + if message.get("tool_calls"): + tool_calls = [ + { + "id": tc.get("id"), + "type": tc.get("type", "function"), + "function": tc.get("function", {}), + } + for tc in message["tool_calls"] + ] + + tool_call_id = message.get("tool_call_id") + + return cls( + role=message["role"], + content=str(message.get("content", "")), + tool_calls=tool_calls, + tool_call_id=tool_call_id, + ) + + def to_message(self) -> Message: + """Convert HistoryMessage to litellm Message format.""" + base: dict[str, Any] = { + "role": self.role, + "content": self.content, + } + + if self.role == "assistant" and self.tool_calls: + return { + "role": "assistant", + "content": self.content, + "tool_calls": self.tool_calls, + } + + if self.role == "tool" and self.tool_call_id: + base["tool_call_id"] = self.tool_call_id + return base + + return base + + +class HistoryStore: + """JSONL file-based history storage.""" + + @staticmethod + def from_config(config: "Config") -> "HistoryStore": + return HistoryStore(config.history_path) + + def __init__(self, base_path: Path): + self.base_path = Path(base_path) + self.sessions_path = self.base_path / "sessions" + self.index_path = self.base_path / "index.jsonl" + + self.base_path.mkdir(parents=True, exist_ok=True) + self.sessions_path.mkdir(parents=True, exist_ok=True) + + def _session_path(self, session_id: str) -> Path: + """Get the file path for a session.""" + return self.sessions_path / f"{session_id}.jsonl" + + def _read_index(self) -> list[HistorySession]: + """Read all session entries from index.jsonl.""" + if not self.index_path.exists(): + return [] + + sessions = [] + with open(self.index_path) as f: + for line in f: + line = line.strip() + if line: + try: + sessions.append(HistorySession.model_validate_json(line)) + except Exception: + continue + return sessions + + def _write_index(self, sessions: list[HistorySession]) -> None: + """Write all session entries to index.jsonl.""" + with open(self.index_path, "w") as f: + for session in sessions: + f.write(session.model_dump_json() + "\n") + + def _find_session_index( + self, sessions: list[HistorySession], session_id: str + ) -> int: + """Find the index of a session in the list.""" + for i, s in enumerate(sessions): + if s.id == session_id: + return i + return -1 + + def create_session(self, agent_id: str, session_id: str) -> dict[str, Any]: + """Create a new conversation session.""" + now = _now_iso() + session = HistorySession( + id=session_id, + agent_id=agent_id, + title=None, + message_count=0, + created_at=now, + updated_at=now, + ) + + # Append to index + with open(self.index_path, "a") as f: + f.write(session.model_dump_json() + "\n") + + # Create session file + self._session_path(session_id).touch() + + return session.model_dump() + + def save_message(self, session_id: str, message: HistoryMessage) -> None: + """Save a message to history.""" + sessions = self._read_index() + idx = self._find_session_index(sessions, session_id) + if idx < 0: + raise ValueError(f"Session not found: {session_id}") + + session = sessions[idx] + + # Append message to session file + session_file = self._session_path(session_id) + with open(session_file, "a") as f: + f.write(message.model_dump_json() + "\n") + + # Update index + session.message_count += 1 + session.updated_at = _now_iso() + + # Auto-generate title from first user message + if session.title is None and message.role == "user": + title = message.content[:50] + if len(message.content) > 50: + title += "..." + session.title = title + + sessions.sort(key=lambda s: s.updated_at, reverse=True) + self._write_index(sessions) + + def list_sessions(self) -> list[HistorySession]: + """List all sessions, most recently updated first.""" + sessions = self._read_index() + sessions.sort(key=lambda s: s.updated_at, reverse=True) + return sessions + + def get_messages(self, session_id: str) -> list[HistoryMessage]: + """Get all messages for a session.""" + session_file = self._session_path(session_id) + if not session_file.exists(): + return [] + + messages: list[HistoryMessage] = [] + with open(session_file) as f: + for line in f: + line = line.strip() + if line: + try: + messages.append(HistoryMessage.model_validate_json(line)) + except Exception: + continue + + return messages + + def get_session_info(self, session_id: str) -> HistorySession | None: + """Get session metadata without loading messages.""" + sessions = self._read_index() + for session in sessions: + if session.id == session_id: + return session + return None diff --git a/07-event-driven/src/mybot/core/session_state.py b/07-event-driven/src/mybot/core/session_state.py new file mode 100644 index 0000000..e5c511d --- /dev/null +++ b/07-event-driven/src/mybot/core/session_state.py @@ -0,0 +1,35 @@ +"""Session state container with persistence helpers.""" + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from litellm.types.completion import ChatCompletionMessageParam as Message + +from mybot.core.history import HistoryMessage + +if TYPE_CHECKING: + from mybot.core.agent import Agent + from mybot.core.context import SharedContext + + +@dataclass +class SessionState: + """Pure conversation state + persistence.""" + + session_id: str + agent: "Agent" + messages: list[Message] + shared_context: "SharedContext" + + def add_message(self, message: Message) -> None: + """Add message to in-memory list + persist.""" + self.messages.append(message) + history_msg = HistoryMessage.from_message(message) + self.shared_context.history_store.save_message(self.session_id, history_msg) + + def build_messages(self) -> list[Message]: + """Build messages list with system prompt.""" + system_prompt = self.agent.agent_def.agent_md + messages: list[Message] = [{"role": "system", "content": system_prompt}] + messages.extend(self.messages) + return messages diff --git a/07-event-driven/src/mybot/core/skill_loader.py b/07-event-driven/src/mybot/core/skill_loader.py new file mode 100644 index 0000000..a590ff1 --- /dev/null +++ b/07-event-driven/src/mybot/core/skill_loader.py @@ -0,0 +1,69 @@ +"""Skill loader for discovering and loading skills.""" + +import logging +from typing import TYPE_CHECKING, Any + +from pydantic import BaseModel, ConfigDict, ValidationError + +from mybot.utils.def_loader import DefNotFoundError, discover_definitions + +if TYPE_CHECKING: + from mybot.utils.config import Config + +logger = logging.getLogger(__name__) + + +class SkillDef(BaseModel): + """Loaded skill definition.""" + + model_config = ConfigDict(extra="forbid") + + id: str + name: str + description: str + content: str + + +class SkillLoader: + """Load and manage skill definitions from filesystem.""" + + @staticmethod + def from_config(config: "Config") -> "SkillLoader": + """Create SkillLoader from config.""" + return SkillLoader(config) + + def __init__(self, config: "Config"): + self.config = config + + def discover_skills(self) -> list[SkillDef]: + """Scan skills directory and return list of valid SkillDef.""" + return discover_definitions( + self.config.skills_path, "SKILL.md", self._parse_skill_def + ) + + def _parse_skill_def( + self, def_id: str, frontmatter: dict[str, Any], body: str + ) -> SkillDef | None: + """Parse skill definition from frontmatter (callback for discover_definitions).""" + try: + return SkillDef( + id=def_id, + name=frontmatter["name"], # type: ignore[misc] + description=frontmatter["description"], # type: ignore[misc] + content=body.strip(), + ) + except ValidationError as e: + logger.warning(f"Invalid skill '{def_id}': {e}") + return None + except KeyError as e: + logger.warning(f"Missing required field in skill '{def_id}': {e}") + return None + + def load_skill(self, skill_id: str) -> SkillDef: + """Load full skill definition by ID.""" + skills = self.discover_skills() + for skill in skills: + if skill.id == skill_id: + return skill + + raise DefNotFoundError("skill", skill_id) diff --git a/07-event-driven/src/mybot/provider/__init__.py b/07-event-driven/src/mybot/provider/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/07-event-driven/src/mybot/provider/llm/__init__.py b/07-event-driven/src/mybot/provider/llm/__init__.py new file mode 100644 index 0000000..c1dfcd6 --- /dev/null +++ b/07-event-driven/src/mybot/provider/llm/__init__.py @@ -0,0 +1,5 @@ +"""LLM provider abstraction.""" + +from .base import LLMProvider, LLMToolCall + +__all__ = ["LLMProvider", "LLMToolCall"] diff --git a/07-event-driven/src/mybot/provider/llm/base.py b/07-event-driven/src/mybot/provider/llm/base.py new file mode 100644 index 0000000..8f52788 --- /dev/null +++ b/07-event-driven/src/mybot/provider/llm/base.py @@ -0,0 +1,85 @@ +"""Base LLM provider abstraction.""" + +from dataclasses import dataclass +from typing import Any, Optional, cast + +from litellm import acompletion, Choices +from litellm.types.completion import ChatCompletionMessageParam as Message + +from mybot.utils.config import LLMConfig + + +@dataclass +class LLMToolCall: + """A tool/function call from the LLM.""" + + id: str + name: str + arguments: str # JSON string + + +class LLMProvider: + """LLM provider using litellm for multi-provider support.""" + + def __init__( + self, + model: str, + api_key: str, + api_base: Optional[str] = None, + temperature: float = 0.7, + max_tokens: int = 2048, + **kwargs: Any, + ): + """Initialize LLM provider.""" + self.model = model + self.api_key = api_key + self.api_base = api_base + self.temperature = temperature + self.max_tokens = max_tokens + self._settings = kwargs + + @classmethod + def from_config(cls, config: LLMConfig) -> "LLMProvider": + """Create provider from LLMConfig.""" + return cls( + model=config.model, + api_key=config.api_key, + api_base=config.api_base, + temperature=config.temperature, + max_tokens=config.max_tokens, + ) + + async def chat( + self, + messages: list[Message], + tools: Optional[list[dict[str, Any]]] = None, + **kwargs: Any, + ) -> tuple[str, list[LLMToolCall]]: + """Default implementation using litellm. Subclasses can override.""" + request_kwargs: dict[str, Any] = { + "model": self.model, + "messages": messages, + "api_key": self.api_key, + } + + if self.api_base: + request_kwargs["api_base"] = self.api_base + if tools: + request_kwargs["tools"] = tools + request_kwargs.update(kwargs) + + response = await acompletion(**request_kwargs) + + message = cast(Choices, response.choices[0]).message + + return ( + message.content or "", + [ + LLMToolCall( + id=tc["id"], + name=tc["function"]["name"], + arguments=tc["function"]["arguments"], + ) + for tc in (message.tool_calls or []) + ], + ) diff --git a/07-event-driven/src/mybot/provider/web_read/__init__.py b/07-event-driven/src/mybot/provider/web_read/__init__.py new file mode 100644 index 0000000..3d8da67 --- /dev/null +++ b/07-event-driven/src/mybot/provider/web_read/__init__.py @@ -0,0 +1,5 @@ +"""Web read provider module.""" + +from .base import ReadResult, WebReadProvider + +__all__ = ["ReadResult", "WebReadProvider"] diff --git a/07-event-driven/src/mybot/provider/web_read/base.py b/07-event-driven/src/mybot/provider/web_read/base.py new file mode 100644 index 0000000..ce410e2 --- /dev/null +++ b/07-event-driven/src/mybot/provider/web_read/base.py @@ -0,0 +1,41 @@ +"""Base class for web read providers.""" + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +from pydantic import BaseModel + +if TYPE_CHECKING: + from mybot.utils.config import Config + + +class ReadResult(BaseModel): + """Normalized result from reading a web page.""" + + url: str + title: str + content: str # Markdown content + error: str | None = None + + +class WebReadProvider(ABC): + """Abstract base class for web page reading providers.""" + + @abstractmethod + async def read(self, url: str) -> ReadResult: + """Read a web page and return normalized content.""" + pass + + @staticmethod + def from_config(config: "Config") -> "WebReadProvider": + """Factory method to create provider from config.""" + if config.webread is None: + raise ValueError("Webread not configured") + + match config.webread.provider: + case "crawl4ai": + from .crawl4ai import Crawl4AIProvider + + return Crawl4AIProvider() + case _: + raise ValueError(f"Unknown webread provider: {config.webread.provider}") diff --git a/07-event-driven/src/mybot/provider/web_read/crawl4ai.py b/07-event-driven/src/mybot/provider/web_read/crawl4ai.py new file mode 100644 index 0000000..4feb85f --- /dev/null +++ b/07-event-driven/src/mybot/provider/web_read/crawl4ai.py @@ -0,0 +1,36 @@ +"""Crawl4AI provider for web page reading.""" + +from crawl4ai import AsyncWebCrawler + +from .base import WebReadProvider, ReadResult + + +class Crawl4AIProvider(WebReadProvider): + """Web read provider using Crawl4AI.""" + + def __init__(self): + """Initialize Crawl4AI provider.""" + pass + + async def read(self, url: str) -> ReadResult: + """Read a web page using Crawl4AI.""" + try: + async with AsyncWebCrawler(verbose=False) as crawler: + result = await crawler.arun(url=url) + + if not result.success: + raise Exception(result.error_message or "Failed to crawl page") + + return ReadResult( + url=url, + title=(result.metadata.get("title", "") if result.metadata else ""), + content=result.markdown or "", + error=None, + ) + except Exception as e: + return ReadResult( + url=url, + title="", + content="", + error=str(e), + ) diff --git a/07-event-driven/src/mybot/provider/web_search/__init__.py b/07-event-driven/src/mybot/provider/web_search/__init__.py new file mode 100644 index 0000000..38761eb --- /dev/null +++ b/07-event-driven/src/mybot/provider/web_search/__init__.py @@ -0,0 +1,5 @@ +"""Web search provider module.""" + +from .base import SearchResult, WebSearchProvider + +__all__ = ["SearchResult", "WebSearchProvider"] diff --git a/07-event-driven/src/mybot/provider/web_search/base.py b/07-event-driven/src/mybot/provider/web_search/base.py new file mode 100644 index 0000000..e5793c4 --- /dev/null +++ b/07-event-driven/src/mybot/provider/web_search/base.py @@ -0,0 +1,42 @@ +"""Base class for web search providers.""" + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +from pydantic import BaseModel + +if TYPE_CHECKING: + from mybot.utils.config import Config + + +class SearchResult(BaseModel): + """Normalized search result from any provider.""" + + title: str + url: str + snippet: str + + +class WebSearchProvider(ABC): + """Abstract base class for web search providers.""" + + @abstractmethod + async def search(self, query: str) -> list[SearchResult]: + """Search the web and return normalized results.""" + pass + + @staticmethod + def from_config(config: "Config") -> "WebSearchProvider": + """Factory method to create provider from config.""" + if config.websearch is None: + raise ValueError("Websearch not configured") + + match config.websearch.provider: + case "brave": + from .brave import BraveSearchProvider + + return BraveSearchProvider(config) + case _: + raise ValueError( + f"Unknown websearch provider: {config.websearch.provider}" + ) diff --git a/07-event-driven/src/mybot/provider/web_search/brave.py b/07-event-driven/src/mybot/provider/web_search/brave.py new file mode 100644 index 0000000..1941e48 --- /dev/null +++ b/07-event-driven/src/mybot/provider/web_search/brave.py @@ -0,0 +1,49 @@ +"""Brave Search API provider.""" + +from typing import TYPE_CHECKING +import httpx + +from .base import WebSearchProvider, SearchResult + +if TYPE_CHECKING: + from mybot.utils.config import Config + + +class BraveSearchProvider(WebSearchProvider): + """Web search provider using Brave Search API.""" + + BASE_URL = "https://api.search.brave.com/res/v1/web/search" + + def __init__(self, config: "Config"): + """Initialize Brave Search provider.""" + self.api_key = config.websearch.api_key + + async def search(self, query: str) -> list[SearchResult]: + """Search the web using Brave Search API.""" + async with httpx.AsyncClient() as client: + response = await client.get( + self.BASE_URL, + headers={ + "Accept": "application/json", + "X-Subscription-Token": self.api_key, + }, + params={ + "q": query, + "count": 10, + }, + timeout=30.0, + ) + response.raise_for_status() + data = response.json() + + results = [] + for item in data.get("web", {}).get("results", []): + results.append( + SearchResult( + title=item.get("title", ""), + url=item.get("url", ""), + snippet=item.get("description", ""), + ) + ) + + return results diff --git a/07-event-driven/src/mybot/server/__init__.py b/07-event-driven/src/mybot/server/__init__.py new file mode 100644 index 0000000..f8f119d --- /dev/null +++ b/07-event-driven/src/mybot/server/__init__.py @@ -0,0 +1,6 @@ +"""Server workers for event-driven architecture.""" + +from .worker import Worker, SubscriberWorker +from .agent_worker import AgentWorker + +__all__ = ["Worker", "SubscriberWorker", "AgentWorker"] diff --git a/07-event-driven/src/mybot/server/agent_worker.py b/07-event-driven/src/mybot/server/agent_worker.py new file mode 100644 index 0000000..3da4829 --- /dev/null +++ b/07-event-driven/src/mybot/server/agent_worker.py @@ -0,0 +1,123 @@ +"""Agent worker for executing agent jobs.""" + +import asyncio +import logging +from dataclasses import replace + +from .worker import SubscriberWorker +from mybot.core.agent import Agent +from mybot.core.events import ( + InboundEvent, + OutboundEvent, +) +from mybot.utils.def_loader import DefNotFoundError + + +# Maximum number of retry attempts for failed sessions +MAX_RETRIES = 3 + +logger = logging.getLogger(__name__) + + +class AgentWorker(SubscriberWorker): + """Dispatches events to session executors.""" + + def __init__(self, context): + super().__init__(context) + + # Auto-subscribe to events + self.context.eventbus.subscribe(InboundEvent, self.dispatch_event) + self.logger.info("AgentWorker subscribed to InboundEvent events") + + async def dispatch_event(self, event: InboundEvent) -> None: + """Create executor task for typed event.""" + # Get agent_id from session (single source of truth) + session_info = self.context.history_store.get_session_info(event.session_id) + if not session_info: + logger.error(f"Session not found: {event.session_id}") + return + + agent_id = session_info.agent_id + + try: + agent_def = self.context.agent_loader.load(agent_id) + except DefNotFoundError as e: + logger.error(f"Agent not found: {agent_id}: {e}") + + result_event = OutboundEvent( + session_id=event.session_id, + content="", + error=str(e), + ) + await self.context.eventbus.publish(result_event) + return + + asyncio.create_task(self.exec_session(event, agent_def)) + + async def exec_session(self, event: InboundEvent, agent_def) -> None: + session_id = event.session_id + + try: + agent = Agent(agent_def, self.context) + if session_id: + try: + session = agent.resume_session(session_id) + except ValueError: + logger.warning(f"Session {session_id} not found, creating new") + session = agent.new_session(session_id=session_id) + else: + session = agent.new_session() + session_id = session.session_id + + # Check for slash command FIRST + if event.content.startswith("/"): + result = await self.context.command_registry.dispatch( + event.content, session + ) + if result: + # Emit response and skip agent chat + await self._emit_response(event, result, session, agent_def.id) + logger.info(f"Command completed: {session_id}") + return + + response = await session.chat(event.content) + logger.info(f"Session completed: {session_id}") + + result_event = OutboundEvent( + session_id=event.session_id, + content=response, + ) + await self.context.eventbus.publish(result_event) + + except Exception as e: + logger.error(f"Session failed: {e}") + + if event.retry_count < MAX_RETRIES: + # Use dataclasses.replace() for retry logic + retry_event = replace( + event, + retry_count=event.retry_count + 1, + content=".", # Minimal message for retry + ) + await self.context.eventbus.publish(retry_event) + else: + result_event = OutboundEvent( + session_id=event.session_id, + content="", + error=str(e), + ) + await self.context.eventbus.publish(result_event) + + async def _emit_response( + self, + event: InboundEvent, + content: str, + session, + agent_id: str, + ) -> None: + """Emit response event with content.""" + result_event = OutboundEvent( + session_id=event.session_id, + content=content, + ) + await self.context.eventbus.publish(result_event) diff --git a/07-event-driven/src/mybot/server/worker.py b/07-event-driven/src/mybot/server/worker.py new file mode 100644 index 0000000..69593dd --- /dev/null +++ b/07-event-driven/src/mybot/server/worker.py @@ -0,0 +1,60 @@ +"""Base worker lifecycle management.""" + +import asyncio +import logging +from abc import ABC, abstractmethod + + +class Worker(ABC): + """Base class for all workers with lifecycle management.""" + + def __init__(self, context): + self.context = context + self.logger = logging.getLogger(f"mybot.server.{self.__class__.__name__}") + self._task: asyncio.Task | None = None + + @abstractmethod + async def run(self) -> None: + """Main worker loop. Runs until cancelled.""" + pass + + def start(self) -> asyncio.Task: + """Start the worker as an asyncio Task.""" + self._task = asyncio.create_task(self.run()) + return self._task + + def is_running(self) -> bool: + """Check if worker is actively running.""" + return self._task is not None and not self._task.done() + + def has_crashed(self) -> bool: + """Check if worker crashed (done but not cancelled).""" + return ( + self._task is not None and self._task.done() and not self._task.cancelled() + ) + + def get_exception(self) -> BaseException | None: + """Get the exception if worker crashed, None otherwise.""" + if self.has_crashed() and self._task is not None: + return self._task.exception() + return None + + async def stop(self) -> None: + """Gracefully stop the worker.""" + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + + +class SubscriberWorker(Worker): + """Worker that only subscribes to events, no active loop.""" + + async def run(self) -> None: + """Wait for cancellation - actual work happens in event handlers.""" + try: + await asyncio.Future() + except asyncio.CancelledError: + pass diff --git a/07-event-driven/src/mybot/tools/__init__.py b/07-event-driven/src/mybot/tools/__init__.py new file mode 100644 index 0000000..843290d --- /dev/null +++ b/07-event-driven/src/mybot/tools/__init__.py @@ -0,0 +1,7 @@ +"""Tools module for agent capabilities.""" + +from mybot.tools.base import BaseTool, tool +from mybot.tools.builtin_tools import bash, edit_file, read_file, write_file +from mybot.tools.registry import ToolRegistry + +__all__ = ["BaseTool", "tool", "ToolRegistry", "read_file", "write_file", "edit_file", "bash"] diff --git a/07-event-driven/src/mybot/tools/base.py b/07-event-driven/src/mybot/tools/base.py new file mode 100644 index 0000000..b2c7701 --- /dev/null +++ b/07-event-driven/src/mybot/tools/base.py @@ -0,0 +1,63 @@ +"""Base tool interface and decorator.""" + +import asyncio +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, Any, Callable + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + + +class BaseTool(ABC): + """Abstract base class for all tools.""" + + name: str + description: str + parameters: dict[str, Any] # JSON Schema for function calling + + @abstractmethod + async def execute(self, session: "AgentSession", **kwargs: Any) -> str: + """Execute the tool.""" + + def get_tool_schema(self) -> dict[str, Any]: + """Get the tool/function schema for LiteLLM.""" + return { + "type": "function", + "function": { + "name": self.name, + "description": self.description, + "parameters": self.parameters, + }, + } + + +def tool(name: str, description: str, parameters: dict[str, Any]) -> Callable: + """Decorator to register a function as a tool.""" + + def decorator(func: Callable) -> "FunctionTool": + return FunctionTool(name, description, parameters, func) + + return decorator + + +class FunctionTool(BaseTool): + """A tool created from a function using the @tool decorator.""" + + def __init__( + self, + name: str, + description: str, + parameters: dict[str, Any], + func: Callable, + ): + self.name = name + self.description = description + self.parameters = parameters + self._func = func + + async def execute(self, session: "AgentSession", **kwargs: Any) -> str: + """Execute the underlying function.""" + result = self._func(session=session, **kwargs) + if asyncio.iscoroutine(result): + result = await result + return str(result) diff --git a/07-event-driven/src/mybot/tools/builtin_tools.py b/07-event-driven/src/mybot/tools/builtin_tools.py new file mode 100644 index 0000000..d327e14 --- /dev/null +++ b/07-event-driven/src/mybot/tools/builtin_tools.py @@ -0,0 +1,133 @@ +"""Built-in tools for agent capabilities.""" + +import asyncio +from pathlib import Path +from typing import TYPE_CHECKING + +from mybot.tools.base import tool + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + + +# Filesystem tools + + +@tool( + name="read", + description="Read the contents of a text file", + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "Path to the file to read"}, + }, + "required": ["path"], + }, +) +async def read_file(path: str, session: "AgentSession") -> str: + """Read and return the contents of a file at the given path.""" + try: + return Path(path).read_text() + except FileNotFoundError: + return f"Error: File not found: {path}" + except PermissionError: + return f"Error: Permission denied reading: {path}" + except IsADirectoryError: + return f"Error: Path is a directory, not a file: {path}" + except Exception as e: + return f"Error reading file: {e}" + + +@tool( + name="write", + description="Write content to a file", + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "Path to the file to write"}, + "content": { + "type": "string", + "description": "Content to write to the file", + }, + }, + "required": ["path", "content"], + }, +) +async def write_file(path: str, content: str, session: "AgentSession") -> str: + """Write content to a file at the given path.""" + try: + Path(path).write_text(content) + return f"Successfully wrote to: {path}" + except PermissionError: + return f"Error: Permission denied writing to: {path}" + except IsADirectoryError: + return f"Error: Path is a directory, not a file: {path}" + except Exception as e: + return f"Error writing file: {e}" + + +@tool( + name="edit", + description="Edit a file by replacing a string with new content", + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "Path to the file to edit"}, + "old_text": {"type": "string", "description": "The text to replace"}, + "new_text": { + "type": "string", + "description": "The new text to replace with", + }, + }, + "required": ["path", "old_text", "new_text"], + }, +) +async def edit_file( + path: str, old_text: str, new_text: str, session: "AgentSession" +) -> str: + """Edit a file by replacing old_text with new_text.""" + try: + content = Path(path).read_text() + if old_text not in content: + return f"Error: '{old_text}' not found in {path}" + new_content = content.replace(old_text, new_text) + Path(path).write_text(new_content) + return f"Successfully edited {path}" + except FileNotFoundError: + return f"Error: File not found: {path}" + except PermissionError: + return f"Error: Permission denied editing: {path}" + except Exception as e: + return f"Error editing file: {e}" + + +# Shell tool + + +@tool( + name="bash", + description="Execute a bash shell command", + parameters={ + "type": "object", + "properties": { + "command": {"type": "string", "description": "The bash command to execute"}, + }, + "required": ["command"], + }, +) +async def bash(command: str, session: "AgentSession") -> str: + """Execute a bash command and return the output.""" + try: + process = await asyncio.create_subprocess_shell( + command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await process.communicate() + output = stdout.decode() if stdout else "" + error = stderr.decode() if stderr else "" + if output and error: + return f"{output}\n{error}" + return output or error or "Command completed with no output" + except Exception as e: + return f"Error executing command: {e}" diff --git a/07-event-driven/src/mybot/tools/registry.py b/07-event-driven/src/mybot/tools/registry.py new file mode 100644 index 0000000..2424099 --- /dev/null +++ b/07-event-driven/src/mybot/tools/registry.py @@ -0,0 +1,56 @@ +"""Tool registry for managing available tools.""" + +from typing import TYPE_CHECKING, Any + +from mybot.tools.base import BaseTool +from mybot.tools.builtin_tools import bash, edit_file, read_file, write_file + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + + +class ToolRegistry: + """Registry for all available tools.""" + + def __init__(self) -> None: + """Initialize an empty tool registry.""" + self._tools: dict[str, BaseTool] = {} + + def register(self, tool: BaseTool) -> None: + """Register a tool.""" + self._tools[tool.name] = tool + + def get(self, name: str) -> BaseTool | None: + """Get a tool by name.""" + return self._tools.get(name) + + def list_all(self) -> list[BaseTool]: + """List all registered tools.""" + return list(self._tools.values()) + + def get_tool_schemas(self) -> list[dict[str, Any]]: + """Get tool schemas for all registered tools.""" + return [tool.get_tool_schema() for tool in self._tools.values()] + + async def execute_tool( + self, name: str, session: "AgentSession", **kwargs: Any + ) -> str: + """Execute a tool by name.""" + tool = self.get(name) + if tool is None: + raise ValueError(f"Tool not found: {name}") + + return await tool.execute(session=session, **kwargs) + + @classmethod + def with_builtins(cls) -> "ToolRegistry": + """Create a ToolRegistry with builtin tools already registered.""" + + registry = cls() + + registry.register(read_file) + registry.register(write_file) + registry.register(edit_file) + registry.register(bash) + + return registry diff --git a/07-event-driven/src/mybot/tools/skill_tool.py b/07-event-driven/src/mybot/tools/skill_tool.py new file mode 100644 index 0000000..b4a37e0 --- /dev/null +++ b/07-event-driven/src/mybot/tools/skill_tool.py @@ -0,0 +1,51 @@ +"""Skill tool factory for creating dynamic skill tool.""" + +from typing import TYPE_CHECKING + +from mybot.tools.base import tool + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + from mybot.core.skill_loader import SkillLoader + + +def create_skill_tool(skill_loader: "SkillLoader"): + """Factory function to create skill tool with dynamic schema.""" + skill_metadata = skill_loader.discover_skills() + + if not skill_metadata: + return None + + # Build XML description of available skills + skills_xml = "\n" + for meta in skill_metadata: + skills_xml += f' {meta.description}\n' + skills_xml += "" + + # Build enum of skill IDs + skill_enum = [meta.id for meta in skill_metadata] + + @tool( + name="skill", + description=f"Load and invoke a specialized skill. {skills_xml}", + parameters={ + "type": "object", + "properties": { + "skill_name": { + "type": "string", + "enum": skill_enum, + "description": "The name of the skill to load", + } + }, + "required": ["skill_name"], + }, + ) + async def skill_tool(skill_name: str, session: "AgentSession") -> str: + """Load and return skill content.""" + try: + skill_def = skill_loader.load_skill(skill_name) + return skill_def.content + except Exception: + return f"Error: Skill '{skill_name}' not found. It may have been removed or is unavailable." + + return skill_tool diff --git a/07-event-driven/src/mybot/tools/webread_tool.py b/07-event-driven/src/mybot/tools/webread_tool.py new file mode 100644 index 0000000..7160b2c --- /dev/null +++ b/07-event-driven/src/mybot/tools/webread_tool.py @@ -0,0 +1,47 @@ +"""Webread tool factory.""" + +from typing import TYPE_CHECKING + +from mybot.tools.base import BaseTool, tool +from mybot.provider.web_read import WebReadProvider + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + from mybot.core.context import SharedContext + + +def create_webread_tool(context: "SharedContext") -> BaseTool | None: + """Factory to create webread tool with injected context.""" + if not context.config.webread: + return None + + provider = WebReadProvider.from_config(context.config) + + @tool( + name="webread", + description=( + "Read and extract content from a web page. " + "Returns the page content as markdown." + ), + parameters={ + "type": "object", + "properties": { + "url": { + "type": "string", + "description": "The URL to read", + } + }, + "required": ["url"], + }, + ) + async def webread(url: str, session: "AgentSession") -> str: + """Read a web page and return markdown content.""" + + result = await provider.read(url) + + if result.error: + return f"Error reading {url}: {result.error}" + + return f"**{result.title}**\n\n{result.content}" + + return webread diff --git a/07-event-driven/src/mybot/tools/websearch_tool.py b/07-event-driven/src/mybot/tools/websearch_tool.py new file mode 100644 index 0000000..bc3a319 --- /dev/null +++ b/07-event-driven/src/mybot/tools/websearch_tool.py @@ -0,0 +1,50 @@ +"""Websearch tool factory.""" + +from typing import TYPE_CHECKING + +from mybot.tools.base import BaseTool, tool +from mybot.provider.web_search import WebSearchProvider + +if TYPE_CHECKING: + from mybot.core.agent import AgentSession + from mybot.core.context import SharedContext + + +def create_websearch_tool(context: "SharedContext") -> BaseTool | None: + """Factory to create websearch tool with injected context.""" + if not context.config.websearch: + return None + + provider = WebSearchProvider.from_config(context.config) + + @tool( + name="websearch", + description=( + "Search the web for information. " + "Returns a list of results with titles, URLs, and snippets." + ), + parameters={ + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "The search query", + } + }, + "required": ["query"], + }, + ) + async def websearch(query: str, session: "AgentSession") -> str: + """Search the web and return formatted results.""" + + results = await provider.search(query) + + if not results: + return "No results found." + + output = [] + for i, r in enumerate(results, 1): + output.append(f"{i}. **{r.title}**\n {r.url}\n {r.snippet}") + return "\n\n".join(output) + + return websearch diff --git a/07-event-driven/src/mybot/utils/__init__.py b/07-event-driven/src/mybot/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/07-event-driven/src/mybot/utils/config.py b/07-event-driven/src/mybot/utils/config.py new file mode 100644 index 0000000..f1f74f0 --- /dev/null +++ b/07-event-driven/src/mybot/utils/config.py @@ -0,0 +1,81 @@ +"""Configuration management.""" + +from pathlib import Path +from typing import Any, Literal + +import yaml +from pydantic import BaseModel, Field, field_validator, model_validator + + +class LLMConfig(BaseModel): + """LLM provider configuration.""" + + provider: str + model: str + api_key: str + api_base: str | None = None + temperature: float = Field(default=0.7, ge=0.0, le=2.0) + max_tokens: int = Field(default=2048, gt=0) + + @field_validator("api_base") + @classmethod + def api_base_must_be_url(cls, v: str | None) -> str | None: + if v is not None and not v.startswith(("http://", "https://")): + raise ValueError("api_base must be a valid URL") + return v + + +class BraveWebSearchConfig(BaseModel): + """Configuration for web search provider.""" + + provider: Literal["brave"] = "brave" + api_key: str + + +class Crawl4AIWebReadConfig(BaseModel): + """Configuration for web read provider.""" + + provider: Literal["crawl4ai"] = "crawl4ai" + + +class Config(BaseModel): + """Main configuration for step 06.""" + + workspace: Path + llm: LLMConfig + default_agent: str + agents_path: Path = Field(default=Path("agents")) + skills_path: Path = Field(default=Path("skills")) + history_path: Path = Field(default=Path(".sessions")) + websearch: BraveWebSearchConfig | None = None + webread: Crawl4AIWebReadConfig | None = None + + @model_validator(mode="after") + def resolve_paths(self) -> "Config": + """Resolve relative paths to absolute using workspace.""" + for field_name in ( + "agents_path", + "skills_path", + "history_path", + ): + path = getattr(self, field_name) + if not path.is_absolute(): + setattr(self, field_name, self.workspace / path) + return self + + @classmethod + def load(cls, workspace_dir: Path) -> "Config": + """Load configuration from workspace directory.""" + config_data = cls._load_config(workspace_dir) + config_data["workspace"] = workspace_dir + return cls.model_validate(config_data) + + @classmethod + def _load_config(cls, workspace_dir: Path) -> dict[str, Any]: + """Load config from YAML file.""" + config_file = workspace_dir / "config.user.yaml" + if not config_file.exists(): + raise FileNotFoundError(f"Config file not found: {config_file}") + + with open(config_file) as f: + return yaml.safe_load(f) or {} diff --git a/07-event-driven/src/mybot/utils/def_loader.py b/07-event-driven/src/mybot/utils/def_loader.py new file mode 100644 index 0000000..6175837 --- /dev/null +++ b/07-event-driven/src/mybot/utils/def_loader.py @@ -0,0 +1,105 @@ +"""Shared utilities for loading definition files (agents, skills, crons).""" + +import logging +from pathlib import Path +from typing import Any, Callable, TypeVar + +import yaml + +T = TypeVar("T") +logger = logging.getLogger(__name__) + + +class DefNotFoundError(Exception): + """Definition folder or file doesn't exist.""" + + def __init__(self, kind: str, def_id: str): + super().__init__(f"{kind.capitalize()} not found: {def_id}") + self.kind = kind + self.def_id = def_id + + +class InvalidDefError(Exception): + """Definition file is malformed.""" + + def __init__(self, kind: str, def_id: str, reason: str): + super().__init__(f"Invalid {kind} '{def_id}': {reason}") + self.kind = kind + self.def_id = def_id + self.reason = reason + + +def parse_definition( + content: str, + def_id: str, + parse_fn: Callable[[str, dict[str, Any], str], T], +) -> T: + """Parse YAML frontmatter + markdown body with type conversion.""" + # Find frontmatter delimiters + if not content.startswith("---\n"): + body = content + return parse_fn(def_id, {}, body) + + end_delimiter = content.find("\n---\n", 4) + if end_delimiter == -1: + body = content + return parse_fn(def_id, {}, body) + + frontmatter_text = content[4:end_delimiter] + body = content[end_delimiter + 5 :] + + raw_dict = yaml.safe_load(frontmatter_text) or {} + return parse_fn(def_id, raw_dict, body) + + +def discover_definitions( + path: Path, + filename: str, + parse_fn: Callable[[str, dict[str, Any], str], T | None], +) -> list[T]: + """Scan directory for definition files.""" + if not path.exists(): + logger.warning(f"Definitions directory not found: {path}") + return [] + + results = [] + for def_dir in path.iterdir(): + if not def_dir.is_dir(): + continue + + def_file = def_dir / filename + if not def_file.exists(): + logger.warning(f"No {filename} found in {def_dir.name}") + continue + + try: + content = def_file.read_text() + result = parse_definition(content, def_dir.name, parse_fn) + if result is not None: + results.append(result) + except Exception as e: + logger.warning(f"Failed to parse {def_dir.name}: {e}") + continue + + return results + + +def write_definition( + def_id: str, + frontmatter: dict[str, Any], + body: str, + base_path: Path, + filename: str, +) -> Path: + """Write a definition file with YAML frontmatter and markdown body.""" + def_dir = base_path / def_id + def_dir.mkdir(parents=True, exist_ok=True) + + # Build file content with YAML frontmatter + yaml_content = yaml.dump(frontmatter, default_flow_style=False, sort_keys=False) + content = f"---\n{yaml_content}---\n\n{body.strip()}\n" + + def_file = def_dir / filename + def_file.write_text(content) + + return def_file