jpmschweitzerandClaude Opus 4.5
3b58fa4f8b
Structure webber into three independent subprojects: - webber-api/: FastAPI backend server with all agent code - webber-cli/: Standalone CLI client (renamed from cli/ to webber_cli/) - webber-sandbox/: Test project for functional testing Key changes: - Each subproject has its own .venv (Python 3.12+) - Added sandbox.sh for managing test project templates - Created sandbox-templates/ with calculator-cli and empty starter - Updated CI/CD for prefixed tags (api/v*, cli/v*) - Added comprehensive AGENTS.md with operational instructions - Added gitignore filtering to glob and grep tools - Created pyproject.toml for each subproject Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
"""
|
|
Agentic conversation loop for interactive CLI.
|
|
"""
|
|
from typing import Any
|
|
|
|
from rich.console import Console
|
|
|
|
from src.cli.session.context import SessionState
|
|
from src.cli.ui.display import format_response
|
|
from src.domains.agents.base import BaseAgent
|
|
from src.shared.logging import logged, trace_span, get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class AgenticLoop:
|
|
"""
|
|
Main conversation loop for interactive CLI sessions.
|
|
|
|
Manages state, executes agent turns, and handles display.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
agent: BaseAgent,
|
|
console: Console,
|
|
working_dir: str,
|
|
):
|
|
"""
|
|
Initialize the agentic loop.
|
|
|
|
Args:
|
|
agent: The agent to use for responses
|
|
console: Rich console for output
|
|
working_dir: Working directory for exploration
|
|
"""
|
|
self.agent = agent
|
|
self.console = console
|
|
self.state = SessionState(working_dir=working_dir)
|
|
|
|
@logged()
|
|
async def run_turn(self, user_input: str) -> str:
|
|
"""
|
|
Execute a single conversation turn.
|
|
|
|
Args:
|
|
user_input: User's prompt/question
|
|
|
|
Returns:
|
|
Agent's response
|
|
"""
|
|
# Record user message
|
|
self.state.add_message("user", user_input)
|
|
|
|
async with trace_span("agentic_turn"):
|
|
try:
|
|
# Run the agent
|
|
response = await self.agent.run(
|
|
user_input,
|
|
working_dir=self.state.working_dir,
|
|
)
|
|
|
|
# Record assistant response
|
|
self.state.add_message("assistant", response)
|
|
|
|
return response
|
|
|
|
except Exception as e:
|
|
logger.exception(f"Agent error: {e}")
|
|
error_msg = f"Error: {e}"
|
|
self.state.add_message("assistant", error_msg)
|
|
raise
|
|
|
|
def display_response(self, response: str) -> None:
|
|
"""Display agent response with formatting."""
|
|
formatted = format_response(response)
|
|
self.console.print(formatted)
|
|
|
|
def display_welcome(self) -> None:
|
|
"""Display welcome message."""
|
|
from src.shared.config import get_settings
|
|
settings = get_settings()
|
|
|
|
self.console.print()
|
|
self.console.print(f"[title]{settings.app_name}[/] [dim]v{settings.app_version}[/]")
|
|
self.console.print(f"[dim]Working in:[/] [path]{self.state.working_dir}[/]")
|
|
self.console.print(f"[dim]Agent:[/] {self.agent.name} - {self.agent.description}")
|
|
self.console.print()
|
|
self.console.print("[dim]Type 'exit' or Ctrl+C to quit. Type 'clear' to reset history.[/]")
|
|
self.console.print()
|
|
|
|
def display_status(self) -> None:
|
|
"""Display session status."""
|
|
self.console.print(f"[dim]Messages: {self.state.message_count} | Tokens: ~{self.state.estimated_tokens}[/]")
|
|
|
|
@property
|
|
def working_dir(self) -> str:
|
|
"""Get current working directory."""
|
|
return self.state.working_dir
|
|
|
|
def set_working_dir(self, path: str) -> None:
|
|
"""Change working directory."""
|
|
self.state.working_dir = path
|
|
self.console.print(f"[info]Changed directory to:[/] [path]{path}[/]")
|