Spaces:
Running
Running
File size: 2,639 Bytes
7c07ade 20ba79b 62d32ab 7c07ade 62d32ab 7c07ade 62d32ab 7c07ade 62d32ab 7c07ade 20ba79b 62d32ab 7c07ade 62d32ab 20ba79b 62d32ab 20ba79b 62d32ab 20ba79b 7c07ade 62d32ab 20ba79b 62d32ab 20ba79b 62d32ab 20ba79b 62d32ab 20ba79b 62d32ab 20ba79b 62d32ab 20ba79b 62d32ab 20ba79b 62d32ab 7c07ade 62d32ab 20ba79b 62d32ab 20ba79b 62d32ab 7c07ade 62d32ab 7c07ade 62d32ab |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
# Phase 4 Implementation Spec: Orchestrator & UI
**Goal**: Connect the Brain and the Body, then give it a Face.
**Philosophy**: "Streaming is Trust."
**Estimated Effort**: 4-5 hours
**Prerequisite**: Phases 1-3 complete
---
## 1. The Slice Definition
This slice connects:
1. **Orchestrator**: The loop calling `SearchHandler` → `JudgeHandler`.
2. **UI**: Gradio app.
**Files**:
- `src/utils/models.py`: Add Orchestrator models
- `src/orchestrator.py`: Main logic
- `src/app.py`: UI
---
## 2. Models (`src/utils/models.py`)
Add to models file:
```python
from enum import Enum
class AgentState(str, Enum):
SEARCHING = "searching"
JUDGING = "judging"
COMPLETE = "complete"
ERROR = "error"
class AgentEvent(BaseModel):
state: AgentState
message: str
data: dict[str, Any] | None = None
```
---
## 3. Orchestrator (`src/orchestrator.py`)
```python
"""Main agent orchestrator."""
import structlog
from typing import AsyncGenerator
from src.shared.config import settings
from src.tools.search_handler import SearchHandler
from src.agent_factory.judges import JudgeHandler
from src.utils.models import AgentEvent, AgentState
logger = structlog.get_logger()
class Orchestrator:
def __init__(self):
self.search = SearchHandler(...)
self.judge = JudgeHandler()
async def run(self, question: str) -> AsyncGenerator[AgentEvent, None]:
"""Run the loop."""
yield AgentEvent(state=AgentState.SEARCHING, message="Starting...")
# ... while loop implementation ...
# ... yield events ...
```
---
## 4. UI (`src/app.py`)
```python
"""Gradio UI."""
import gradio as gr
from src.orchestrator import Orchestrator
async def chat(message, history):
agent = Orchestrator()
async for event in agent.run(message):
yield f"**[{event.state.value}]** {event.message}"
# ... gradio blocks setup ...
```
---
## 5. TDD Workflow
### Test File: `tests/unit/test_orchestrator.py`
```python
"""Unit tests for Orchestrator."""
import pytest
from unittest.mock import AsyncMock
class TestOrchestrator:
@pytest.mark.asyncio
async def test_run_loop(self, mocker):
from src.orchestrator import Orchestrator
# Mock handlers
# ... setup mocks ...
orch = Orchestrator()
events = [e async for e in orch.run("test")]
assert len(events) > 0
```
---
## 6. Implementation Checklist
- [ ] Update `src/utils/models.py`
- [ ] Implement `src/orchestrator.py`
- [ ] Implement `src/app.py`
- [ ] Write tests in `tests/unit/test_orchestrator.py`
- [ ] Run `uv run python src/app.py` |