diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bc14a47..eb0c26d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -111,7 +111,10 @@ jobs: - name: Build backend image uses: docker/build-push-action@v5 with: - context: ./backend + # backend/Dockerfile COPYs paths relative to the repo root + # (e.g. `COPY backend/app ./app`), matching how Railway builds it, + # so the build context must be the repo root, not ./backend. + context: . file: ./backend/Dockerfile push: false tags: personal-q-backend:test diff --git a/Dockerfile.celery-worker b/Dockerfile.celery-worker index c014d06..4faced0 100644 --- a/Dockerfile.celery-worker +++ b/Dockerfile.celery-worker @@ -3,7 +3,7 @@ FROM python:3.11-slim # Cache buster - change this value to force rebuild -ARG CACHE_BUST=2026-01-07-v6 +ARG CACHE_BUST=2026-06-16-v7-agent-sdk # Set working directory WORKDIR /app diff --git a/MIGRATION.md b/MIGRATION.md new file mode 100644 index 0000000..353216c --- /dev/null +++ b/MIGRATION.md @@ -0,0 +1,193 @@ +# Migration: CrewAI → Claude Agent Runtime (Anthropic SDK tool-use loop) + +> Status: **In progress on branch `claude/claude-sdk-agent-migration-lxxan0`.** +> This document is both the transition plan and the record of what has been done. + +## 1. Why migrate + +Personal-Q originally used **CrewAI** (which wraps **LiteLLM** + **LangChain**) +to orchestrate agent task execution. In practice this added a heavy, fast-moving +dependency stack on top of a backend that *already* talked to Claude directly +through the official **Anthropic SDK** (`backend/app/services/llm_service.py`). + +CrewAI was effectively a parallel, redundant execution path with real costs: + +- **Dependency weight & churn** — `crewai`, `litellm`, `langchain-anthropic`, + `langchain-core` pulled in a large transitive tree, slow Docker builds, and + frequent breaking releases (the repo already carried pins like + `crewai==0.203.1` in `requirements.txt` but `crewai>=0.86.0` in `pyproject.toml` + — an inconsistency waiting to break). +- **Indirection** — agent execution went app → CrewAI → LiteLLM → Anthropic, + making errors, token accounting, retries, and prompt-injection controls harder + to reason about than a direct SDK call. +- **Memory disabled / tools unused** — `Crew(memory=False)` was forced to avoid + OpenAI embeddings, and `create_agent_tools()` always returned `[]`. The crew + was, in effect, a single LLM call with extra layers. + +The replacement is a **self-hosted Claude agent runtime**: an agentic +**tool-use loop** built directly on the Anthropic SDK (Messages API). It keeps +everything in our own FastAPI/Celery infrastructure, removes the CrewAI/LangChain/ +LiteLLM stack, and gives us a clean place to add real tools. + +### Architecture chosen + +| Option | Decision | +|---|---| +| **Anthropic SDK tool-use loop** (self-hosted agentic loop) | ✅ **Chosen** — minimal deps, stays in our infra, fits Celery + the existing `llm_service` resilience patterns. | +| Claude Managed Agents (`client.beta.agents`/`sessions`, Anthropic-hosted containers) | ❌ Not now — server-managed, Anthropic-only, a larger architectural shift away from the Celery/worker model. Revisit if we need sandboxed code execution per task (see §8). | + +## 2. What the runtime does + +`backend/app/services/agent_runtime.py` → `AgentRuntime`: + +``` +execute_agent_task(db, agent, task_description, task_input) + └─ resolve & validate model (Anthropic only) + └─ build system prompt (role + agent.system_prompt, sanitized) + └─ build tools (empty today; extensible) + └─ run agentic loop: + while not done and iterations < MAX_AGENT_ITERATIONS: + resp = client.messages.create(model, system, messages, tools) + if resp.stop_reason == "tool_use": run tools, append results, continue + if resp.stop_reason == "pause_turn": re-send to resume + else: collect final text → done + └─ return {success, result, model_used, usage, iterations} +``` + +`execute_multi_agent_task(db, agents, tasks, process)` chains agents +**sequentially**, feeding each agent the prior agents' outputs as context +(reproducing CrewAI's sequential process; `hierarchical` adds a coordination +note). The public method signatures match the former `CrewService` so the +Celery worker swap was a one-line change. + +Key correctness properties carried over / improved: + +- **API keys come only from environment variables** (via `provider_registry`), never the DB. +- **Prompt-injection sanitization** reused from `app.security.prompt_sanitizer`. +- **Resilience**: tenacity retry with exponential backoff on transient errors + (`APIConnectionError`, `RateLimitError`, timeouts), per-call. +- **Token usage** is aggregated across loop iterations and returned to the worker. +- **Tool failures** are reported back to the model as `is_error` tool results + rather than crashing the task. + +## 3. Model-awareness (important behavioral change) + +Claude **4.6+ and Fable** models (`claude-opus-4-6/4-7/4-8`, `claude-sonnet-4-6`, +`claude-haiku-4-5`, `claude-fable-5`, `claude-mythos-5`) **reject** sampling +parameters (`temperature`/`top_p`/`top_k`) and the legacy +`thinking.budget_tokens` config — sending them returns HTTP 400. + +Both the new runtime and `llm_service` now **omit `temperature` for these model +families** and only send it for older models that accept it +(`_accepts_sampling_params`). This was a latent bug surfaced by moving the +default to a current model. + +The retired default `claude-3-5-sonnet-20241022` was replaced: + +- `settings.default_model` → **`claude-opus-4-8`**. +- `llm_service.validate_api_key` ping model → **`claude-haiku-4-5`** (was a retired model that would now 404). +- `provider_registry` gained current models: Opus 4.8 (recommended), Sonnet 4.6, Haiku 4.5, Fable 5. +- `model_validator` legacy map now remaps **retired snapshot IDs** (e.g. + `claude-3-5-sonnet-20241022`, `claude-3-7-sonnet-20250219`, + `claude-3-opus-20240229`) to current models, so existing seeded agents keep working. + +> **Scope note (multi-provider):** the OpenAI/Mistral entries remain in the +> registry for the settings UI and model metadata, but the **agent runtime +> executes Anthropic models only** and returns a clear error for other +> providers. CrewAI/LiteLLM was what made cross-provider *execution* work. If +> non-Anthropic execution is still required, see §8. + +## 4. Files changed + +**New** +- `backend/app/services/agent_runtime.py` — the Claude agent runtime. +- `backend/tests/unit/test_agent_runtime.py` — runtime unit tests (mocked SDK). +- `MIGRATION.md` — this document. + +**Modified** +- `backend/app/workers/tasks.py` — worker now calls `AgentRuntime.execute_agent_task`. +- `backend/app/services/__init__.py` — export `AgentRuntime`, drop `CrewService`. +- `backend/app/services/llm_service.py` — model-aware sampling params, current + validation-ping model, refreshed pricing table. +- `backend/app/services/model_validator.py` — current model aliases + retired-snapshot remaps. +- `backend/app/services/provider_registry.py` — current Claude models. +- `backend/config/settings.py` — `default_model = claude-opus-4-8`. +- `backend/app/main.py` — app description. +- `backend/requirements.txt`, `backend/pyproject.toml` — removed `crewai`, + `litellm`, `langchain-anthropic`, `langchain-core`; `anthropic>=0.39.0`. +- `backend/Dockerfile`, `Dockerfile.celery-worker` — dropped CrewAI build notes; bumped worker cache buster. +- `backend/tests/integration/test_api_key_config.py`, + `backend/tests/unit/test_websocket_broadcasts.py` — retargeted to `AgentRuntime`. +- `README.md`, `.github/workflows/claude-code-review.yml` — docs/architecture text. + +**Removed** +- `backend/app/services/crew_service.py` +- `backend/tests/unit/test_crew_service.py` + +## 5. Phased rollout + +1. **Phase 1 — Runtime in place (this PR).** New runtime, worker wired, CrewAI + removed from deps and code, tests updated. Behavior parity: single call per + task (no tools yet), multi-agent sequential chaining. +2. **Phase 2 — Verify in an environment with `ANTHROPIC_API_KEY`/`PERSONAL_Q_API_KEY`.** + Run a real task end-to-end through Celery; confirm token usage + result + persistence and WebSocket events. +3. **Phase 3 — Add real tools.** Register handlers in `TOOL_HANDLERS` and expose + schemas from `build_tools` (driven by `agent.tools_config`). Natural first + tools: web search/fetch (server-side), and the existing integrations + (Slack/Obsidian/MS Graph) as client-side tools. +4. **Phase 4 — Optional enhancements.** Adaptive thinking + `effort` for modern + models; streaming task output to the WebSocket; prompt caching for shared + system prompts; per-agent `max_tokens` streaming guard for large outputs. + +## 6. Testing + +- `backend/tests/unit/test_agent_runtime.py` — role/prompt helpers, + `_accepts_sampling_params`, model resolution (rejects non-Anthropic), happy + path, temperature inclusion/omission by model family, tool-use loop, unknown-tool + error handling, multi-agent chaining and validation. +- `backend/tests/integration/test_api_key_config.py` — graceful failure when the + key is missing, non-Anthropic rejection, multi-agent failure propagation. +- `backend/tests/unit/test_websocket_broadcasts.py` — task lifecycle events now + mock `AgentRuntime`. + +Run: `cd backend && pytest tests/unit/test_agent_runtime.py tests/integration/test_api_key_config.py tests/unit/test_websocket_broadcasts.py` + +## 7. Rollback + +The change is isolated to the execution layer. To roll back: +`git revert` this PR (restores `crew_service.py`, the deps, and the worker import). +No DB schema or API contract changed — task input/output shapes are unchanged, so +no migration is required either direction. + +## 8. Known follow-ups / out of scope + +- **Code Quality (black/isort) — FIXED in this PR.** The `ci.yml` lint job + installs an **unpinned** `black` (now 26.5.1) + `isort` and checks all of + `backend/app`. Pre-existing drift in files this migration didn't otherwise + touch (`routers/{auth,metrics,tasks,websocket,llm}.py`, `workers/celery_app.py`, + `services/memory_service.py`, `middleware/rate_limit.py`, `db/*`) was + reformatted repo-wide (`black --line-length=100 backend/app` + + `isort --profile black --line-length 100 backend/app`) so the lint job passes. + Recommended follow-up: **pin** `black`/`isort` versions in the workflow so an + unpinned upgrade can't re-break the check. +- **Build Docker Images — FIXED in this PR.** The failure was a pre-existing CI + misconfiguration, not a missing file: `backend/entrypoint.sh` (and every other + `COPY` source) exists, but `backend/Dockerfile` COPYs paths relative to the + **repo root** (`COPY backend/app ./app`, matching how Railway builds it) while + the `ci.yml` `build` job passed `context: ./backend`, so `COPY backend/entrypoint.sh` + resolved to `backend/backend/entrypoint.sh` → not found. Fixed by setting the + backend build step's `context: .` (the frontend step already used `context: .`). +- **`backend/uv.lock`** still references the removed packages. Regenerate with + `uv lock` (or delete if pip-only) so the lockfile matches `pyproject.toml`. + The Docker build uses `pip install -e .` from `pyproject.toml`, so this is a + hygiene item, not a build blocker. +- **Non-Anthropic execution** (OpenAI/Mistral) is no longer wired for task + execution. If needed, add provider-specific clients behind the runtime or + reintroduce a thin LiteLLM adapter for those providers only. +- **`anthropic` version**: floor kept at `>=0.39.0` for safety; bump to a recent + release to pick up newer features (adaptive thinking betas, server-side tools). +- **PROJECT_SPECIFICATION.md** still describes CrewAI in places (historical spec); + update if it is meant to be a living document. +- **Managed Agents** remains the path if we later want Anthropic-hosted, per-task + sandboxed tool execution. diff --git a/README.md b/README.md index 112bb59..b48ea1f 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Personal-Q AI Agent Management System -A comprehensive, locally-run AI agent management platform with CrewAI orchestration, Claude integration, and real-time monitoring. +A comprehensive, locally-run AI agent management platform with a Claude agent runtime (Anthropic SDK tool-use loop), Claude integration, and real-time monitoring. ![License](https://img.shields.io/badge/license-MIT-blue.svg) ![Python](https://img.shields.io/badge/python-3.11+-blue.svg) @@ -10,7 +10,7 @@ A comprehensive, locally-run AI agent management platform with CrewAI orchestrat ## Features ### Core Capabilities -- **Multi-Agent Orchestration**: CrewAI-powered agent collaboration with sequential and hierarchical workflows +- **Multi-Agent Orchestration**: Claude agent runtime with sequential and hierarchical workflows (Anthropic SDK tool-use loop) - **LLM Integration**: Claude (Anthropic) integration with streaming support - **Task Management**: Async task queue with Celery + Redis for background processing - **Memory & Context**: ChromaDB-based vector storage for semantic search and RAG @@ -80,8 +80,8 @@ See [Installation Guide](docs/INSTALLATION.md) for detailed setup instructions. │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ │ ┌──────┴──────────────────┴──────────────────┴───────┐ │ -│ │ CrewAI Orchestration │ │ -│ │ (Multi-Agent Task Execution) │ │ +│ │ Claude Agent Runtime │ │ +│ │ (Anthropic SDK tool-use loop · multi-agent) │ │ │ └──────────────────────┬───────────────────────────────┘ │ │ │ │ │ ┌──────────────────────┴───────────────────────────────┐ │ @@ -117,7 +117,7 @@ See [Installation Guide](docs/INSTALLATION.md) for detailed setup instructions. - SQLAlchemy (async ORM) - Alembic (migrations) - Celery + Redis (task queue) -- CrewAI (agent orchestration) +- Anthropic SDK (Claude agent runtime — tool-use loop) **Databases** - SQLite (structured data) @@ -213,7 +213,7 @@ MIT License - see [LICENSE](LICENSE) file for details. ## Acknowledgments - Built with [Claude Code](https://claude.com/claude-code) -- Powered by [CrewAI](https://github.com/joaomdmoura/crewAI) +- Powered by the [Anthropic SDK](https://github.com/anthropics/anthropic-sdk-python) (Claude agent runtime) - UI components from [shadcn/ui](https://ui.shadcn.com/) - LLM by [Anthropic Claude](https://www.anthropic.com/) diff --git a/backend/Dockerfile b/backend/Dockerfile index 5e8df26..ea2dbad 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -36,7 +36,6 @@ COPY backend/migrations ./migrations COPY backend/entrypoint.sh ./entrypoint.sh # Install package in editable mode with dependencies -# Increased timeout for CrewAI dependency resolution RUN pip install --no-cache-dir -e . --timeout 300 # Create data directories diff --git a/backend/app/db/__init__.py b/backend/app/db/__init__.py index 7324f7b..38cff93 100644 --- a/backend/app/db/__init__.py +++ b/backend/app/db/__init__.py @@ -14,6 +14,6 @@ "EncryptedString", ] -from .lance_client import LanceDBClient, get_lance_client from .database import AsyncSessionLocal, Base, close_db, engine, get_db, init_db from .encrypted_types import EncryptedString +from .lance_client import LanceDBClient, get_lance_client diff --git a/backend/app/db/init_db.py b/backend/app/db/init_db.py index 56bef23..cc4c451 100644 --- a/backend/app/db/init_db.py +++ b/backend/app/db/init_db.py @@ -7,13 +7,13 @@ import uuid from datetime import datetime +from app.db.database import AsyncSessionLocal, Base, engine from app.db.lance_client import ( AgentOutputSchema, ConversationSchema, DocumentSchema, lance_client, ) -from app.db.database import AsyncSessionLocal, Base, engine from app.models import Agent, AgentStatus, AgentType, APIKey from app.utils.datetime_utils import utcnow from sqlalchemy.ext.asyncio import AsyncSession diff --git a/backend/app/db/lance_client.py b/backend/app/db/lance_client.py index 09490a9..52f04d8 100644 --- a/backend/app/db/lance_client.py +++ b/backend/app/db/lance_client.py @@ -8,11 +8,10 @@ from typing import Optional import lancedb +from config.settings import settings from lancedb.embeddings import get_registry from lancedb.pydantic import LanceModel, Vector -from config.settings import settings - logger = logging.getLogger(__name__) # Initialize embedding model from registry diff --git a/backend/app/main.py b/backend/app/main.py index 1736cda..48fc2fc 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -111,7 +111,7 @@ async def lifespan(app: FastAPI): app = FastAPI( title=settings.app_name, version=settings.app_version, - description="AI Agent Management System with CrewAI orchestration", + description="AI Agent Management System with Claude agent runtime (Anthropic SDK tool-use)", lifespan=lifespan, docs_url=f"{settings.api_prefix}/docs", redoc_url=f"{settings.api_prefix}/redoc", diff --git a/backend/app/middleware/rate_limit.py b/backend/app/middleware/rate_limit.py index a70e0b2..dfee821 100644 --- a/backend/app/middleware/rate_limit.py +++ b/backend/app/middleware/rate_limit.py @@ -5,12 +5,11 @@ import logging +from config.settings import settings from fastapi import Request from slowapi import Limiter from slowapi.util import get_remote_address -from config.settings import settings - logger = logging.getLogger(__name__) diff --git a/backend/app/routers/auth.py b/backend/app/routers/auth.py index 267d8f9..3979881 100644 --- a/backend/app/routers/auth.py +++ b/backend/app/routers/auth.py @@ -129,9 +129,7 @@ def verify_access_token(token: str) -> Optional[dict]: # Verify email is in the allowed list if not settings.is_email_allowed(payload.get("email", "")): - logger.warning( - f"Token email not in allowed list: {payload.get('email')}" - ) + logger.warning(f"Token email not in allowed list: {payload.get('email')}") return None return payload @@ -166,10 +164,12 @@ async def login(request: Request): # Store state in Redis with TTL (HIGH-001 fix: Redis instead of in-memory) try: redis_client = get_redis_client() - state_data = json.dumps({ - "created_at": utcnow().isoformat(), - "redirect_uri": str(request.url_for("auth_callback")) - }) + state_data = json.dumps( + { + "created_at": utcnow().isoformat(), + "redirect_uri": str(request.url_for("auth_callback")), + } + ) redis_client.setex(f"{OAUTH_STATE_PREFIX}{state}", OAUTH_STATE_TTL, state_data) logger.info(f"Generated OAuth state token: {state[:8]}... (stored in Redis)") except redis.RedisError as e: @@ -326,8 +326,7 @@ async def logout(request: Request): if not csrf_header or csrf_cookie != csrf_header: logger.warning("Logout CSRF validation failed") raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="CSRF validation failed" + status_code=status.HTTP_403_FORBIDDEN, detail="CSRF validation failed" ) # Verify session exists (cookie must be present for logout) diff --git a/backend/app/routers/llm.py b/backend/app/routers/llm.py index 579025d..d16a0e5 100644 --- a/backend/app/routers/llm.py +++ b/backend/app/routers/llm.py @@ -6,13 +6,12 @@ import logging from typing import Dict -from fastapi import APIRouter, Depends - from app.dependencies.auth import get_current_user from app.schemas.llm import ProvidersResponse, ValidationResult from app.services.model_validator import model_validator from app.services.provider_registry import provider_registry from config.settings import settings +from fastapi import APIRouter, Depends logger = logging.getLogger(__name__) diff --git a/backend/app/routers/metrics.py b/backend/app/routers/metrics.py index 4af30a6..416b20a 100644 --- a/backend/app/routers/metrics.py +++ b/backend/app/routers/metrics.py @@ -79,10 +79,7 @@ async def get_agent_metrics( # Issue #114 fix: Use proper HTTP status code for not found if not agent: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="Agent not found" - ) + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Agent not found") # Get task breakdown pending_result = await db.execute( diff --git a/backend/app/routers/tasks.py b/backend/app/routers/tasks.py index c66d540..bdeb283 100644 --- a/backend/app/routers/tasks.py +++ b/backend/app/routers/tasks.py @@ -234,9 +234,7 @@ async def retry_task( """ Retry a pending or failed task by re-queuing it for execution. """ - result = await db.execute( - select(TaskModel).where(TaskModel.id == task_id) - ) + result = await db.execute(select(TaskModel).where(TaskModel.id == task_id)) task = result.scalar_one_or_none() if not task: diff --git a/backend/app/routers/websocket.py b/backend/app/routers/websocket.py index 01cc667..25c6953 100644 --- a/backend/app/routers/websocket.py +++ b/backend/app/routers/websocket.py @@ -137,12 +137,13 @@ async def websocket_endpoint(websocket: WebSocket): # Prevents resource exhaustion from clients that connect but never authenticate try: auth_data = await asyncio.wait_for( - websocket.receive_text(), - timeout=AUTH_TIMEOUT_SECONDS + websocket.receive_text(), timeout=AUTH_TIMEOUT_SECONDS ) except asyncio.TimeoutError: logger.warning("WebSocket: Authentication timeout - closing connection") - await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="Authentication timeout") + await websocket.close( + code=status.WS_1008_POLICY_VIOLATION, reason="Authentication timeout" + ) return # Validate message size @@ -157,14 +158,22 @@ async def websocket_endpoint(websocket: WebSocket): except json.JSONDecodeError: logger.warning("WebSocket: Invalid JSON in authentication message") await websocket.send_json({"error": "Invalid JSON"}) - await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="Invalid authentication format") + await websocket.close( + code=status.WS_1008_POLICY_VIOLATION, reason="Invalid authentication format" + ) return # Verify this is an authentication message if auth_message.get("action") != "authenticate": - logger.warning(f"WebSocket: First message must be authentication, got: {auth_message.get('action')}") - await websocket.send_json({"error": "First message must be authentication", "expected_action": "authenticate"}) - await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="Authentication required") + logger.warning( + f"WebSocket: First message must be authentication, got: {auth_message.get('action')}" + ) + await websocket.send_json( + {"error": "First message must be authentication", "expected_action": "authenticate"} + ) + await websocket.close( + code=status.WS_1008_POLICY_VIOLATION, reason="Authentication required" + ) return # Issue #110 fix: Support both token-based and cookie-based authentication @@ -185,8 +194,12 @@ async def websocket_endpoint(websocket: WebSocket): if not user: logger.warning("WebSocket: Authentication failed - invalid token") - await websocket.send_json({"error": "Authentication failed", "reason": "Invalid or expired token"}) - await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="Authentication failed") + await websocket.send_json( + {"error": "Authentication failed", "reason": "Invalid or expired token"} + ) + await websocket.close( + code=status.WS_1008_POLICY_VIOLATION, reason="Authentication failed" + ) return # Authentication successful @@ -203,7 +216,9 @@ async def websocket_endpoint(websocket: WebSocket): # Validate message size (MEDIUM-003 fix) if len(data) > MAX_MESSAGE_SIZE: logger.warning(f"WebSocket: Message too large ({len(data)} bytes)") - await websocket.send_json({"error": "Message too large", "max_size": MAX_MESSAGE_SIZE}) + await websocket.send_json( + {"error": "Message too large", "max_size": MAX_MESSAGE_SIZE} + ) continue try: @@ -232,7 +247,9 @@ async def websocket_endpoint(websocket: WebSocket): await websocket.send_json({"error": "Invalid JSON"}) except WebSocketDisconnect: - logger.info(f"WebSocket disconnected for user: {user.get('email') if user else 'unauthenticated'}") + logger.info( + f"WebSocket disconnected for user: {user.get('email') if user else 'unauthenticated'}" + ) manager.disconnect(websocket) except Exception as e: logger.error(f"WebSocket error: {e}") diff --git a/backend/app/services/__init__.py b/backend/app/services/__init__.py index bdf9652..a430f1f 100644 --- a/backend/app/services/__init__.py +++ b/backend/app/services/__init__.py @@ -4,16 +4,16 @@ __all__ = [ "AgentService", - "CrewService", + "AgentRuntime", "LLMService", "MemoryService", "CacheService", "EncryptionService", ] +from .agent_runtime import AgentRuntime from .agent_service import AgentService from .cache_service import CacheService, cache_service -from .crew_service import CrewService from .encryption_service import EncryptionService, encryption_service from .llm_service import LLMService, get_llm_service from .memory_service import MemoryService, get_memory_service diff --git a/backend/app/services/agent_runtime.py b/backend/app/services/agent_runtime.py new file mode 100644 index 0000000..d084238 --- /dev/null +++ b/backend/app/services/agent_runtime.py @@ -0,0 +1,525 @@ +""" +ABOUTME: Claude agent runtime — a self-hosted tool-use agentic loop on the Anthropic SDK. +ABOUTME: Replaces the CrewAI/LiteLLM orchestration layer with the Messages API + tool use. + +This module is the execution engine for agents. Instead of delegating to CrewAI +(which wrapped LiteLLM and LangChain), it drives Claude directly through the +official Anthropic SDK using an agentic loop: + + while not done: + response = client.messages.create(model, system, messages, tools) + if response wants a tool -> run the tool, append the result, continue + else -> done + +Tools are optional. With an empty tool set the loop degenerates to a single +Messages API call, which is the common case today. The loop is structured so +that real tools (web search, integrations, etc.) can be added later by +registering a handler in ``TOOL_HANDLERS`` and exposing a schema from +``build_tools``. + +SECURITY: API keys are ONLY read from environment variables (via the provider +registry) — NEVER from the database. +""" + +import logging +from typing import Any, Awaitable, Callable, Dict, List, Optional + +import httpx +from anthropic import ( + APIConnectionError, + APIError, + AsyncAnthropic, + RateLimitError, +) +from app.models.agent import Agent, AgentType +from app.security.prompt_sanitizer import PromptSanitizer +from app.services.model_validator import ModelValidator, model_validator +from app.services.provider_registry import provider_registry +from config.settings import settings +from sqlalchemy.ext.asyncio import AsyncSession +from tenacity import ( + before_sleep_log, + retry, + retry_if_exception_type, + stop_after_attempt, + wait_exponential, +) + +logger = logging.getLogger(__name__) + +# Maximum number of model<->tool round trips before we stop the loop. With no +# tools registered this is effectively a single call; the cap protects against +# runaway tool loops once tools are added. +MAX_AGENT_ITERATIONS = 10 + +# Timeout configuration (seconds) — mirrors LLMService. +TIMEOUT_CONNECT = 5.0 +TIMEOUT_READ = 60.0 +TIMEOUT_WRITE = 10.0 +TIMEOUT_POOL = 5.0 + +# Model families that REJECT sampling params (temperature/top_p/top_k) and the +# legacy `thinking.budget_tokens` config. Sending those to these models returns +# a 400. We therefore omit temperature for any model whose id starts with one of +# these prefixes. See the Anthropic model migration guide. +_NO_SAMPLING_PARAM_PREFIXES = ( + "claude-opus-4-6", + "claude-opus-4-7", + "claude-opus-4-8", + "claude-sonnet-4-6", + "claude-haiku-4-5", + "claude-fable-5", + "claude-mythos-5", +) + + +# A tool handler takes the tool input dict and returns a string result. +ToolHandler = Callable[[Dict[str, Any]], Awaitable[str]] + +# Registry of built-in tool handlers, keyed by tool name. Empty by default. +# Register entries here (and expose a matching schema from ``build_tools``) to +# give agents real capabilities. +TOOL_HANDLERS: Dict[str, ToolHandler] = {} + + +def _accepts_sampling_params(model: str) -> bool: + """Return True if the model accepts temperature/top_p (older Claude models).""" + return not any(model.startswith(prefix) for prefix in _NO_SAMPLING_PARAM_PREFIXES) + + +class AgentRuntime: + """Executes agent tasks via the Anthropic SDK tool-use loop.""" + + # Maps the application's agent types to a short role descriptor that is + # prepended to the agent's own system prompt. + _ROLE_BY_TYPE = { + AgentType.CONVERSATIONAL: "a conversational customer support specialist", + AgentType.ANALYTICAL: "a data analyst and researcher", + AgentType.CREATIVE: "a creative content writer", + AgentType.AUTOMATION: "an automation and workflow specialist", + } + + # ------------------------------------------------------------------ # + # Configuration helpers + # ------------------------------------------------------------------ # + @staticmethod + def _role_for(agent: Agent) -> str: + return AgentRuntime._ROLE_BY_TYPE.get(agent.agent_type, "a general-purpose assistant") + + @staticmethod + def _build_system_prompt(agent: Agent) -> str: + """Compose the system prompt from the agent's role and configured prompt.""" + role = AgentRuntime._role_for(agent) + base = agent.system_prompt or "" + # Validate/sanitize the operator-provided system prompt. + safe_prompt = PromptSanitizer.validate_agent_prompt(base) if base else "" + header = f"You are {role}." + if safe_prompt: + return f"{header}\n\n{safe_prompt}" + return header + + @staticmethod + def build_tools(agent: Agent) -> List[Dict[str, Any]]: + """ + Build the Anthropic tool schema list for an agent. + + Returns an empty list today (no tools wired up). To add a tool: + 1. Append its JSON-schema definition here (optionally gated on + ``agent.tools_config``). + 2. Register an async handler in ``TOOL_HANDLERS`` under the same name. + """ + # tools_config is reserved for per-agent tool enablement. + return [] + + @staticmethod + def _resolve_model( + model_string: str, validator: ModelValidator = None + ) -> tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]: + """ + Validate the model string and return (model_id, api_key, error). + + Only the Anthropic provider is supported by this runtime. Non-Anthropic + providers return a descriptive error rather than silently failing. + + SECURITY: API keys come ONLY from environment variables. + """ + validator = validator or model_validator + validation = validator.validate_model(model_string, check_configured=True) + + if not validation.is_valid: + return None, None, {"error": validation.error} + + if validation.provider != "anthropic": + return ( + None, + None, + { + "error": ( + f"Provider '{validation.provider}' is not supported by the Claude " + f"agent runtime. Configure an Anthropic model (e.g. " + f"'anthropic/claude-opus-4-8')." + ) + }, + ) + + api_key = provider_registry.get_api_key("anthropic") + if not api_key: + provider_config = provider_registry.get_provider("anthropic") + env_var = provider_config.api_key_env if provider_config else "ANTHROPIC_API_KEY" + return ( + None, + None, + { + "error": ( + f"API key not configured for provider 'anthropic'. " + f"Set {env_var} (or PERSONAL_Q_API_KEY) environment variable." + ) + }, + ) + + # validation.model is the bare model id (no provider prefix), which is + # exactly what the Anthropic Messages API expects. + return validation.model, api_key, None + + @staticmethod + def _client(api_key: str) -> AsyncAnthropic: + http_client = httpx.AsyncClient( + timeout=httpx.Timeout( + connect=TIMEOUT_CONNECT, + read=TIMEOUT_READ, + write=TIMEOUT_WRITE, + pool=TIMEOUT_POOL, + ) + ) + return AsyncAnthropic(api_key=api_key, http_client=http_client) + + # ------------------------------------------------------------------ # + # Core agentic loop + # ------------------------------------------------------------------ # + @staticmethod + @retry( + retry=retry_if_exception_type((APIConnectionError, RateLimitError, httpx.TimeoutException)), + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=1, max=10), + before_sleep=before_sleep_log(logger, logging.WARNING), + reraise=True, + ) + async def _create_message( + client: AsyncAnthropic, + *, + model: str, + max_tokens: int, + temperature: Optional[float], + system: str, + messages: List[Dict[str, Any]], + tools: List[Dict[str, Any]], + ): + """Single Messages API call with retry on transient errors.""" + kwargs: Dict[str, Any] = { + "model": model, + "max_tokens": max_tokens, + "system": system, + "messages": messages, + } + # Only send temperature to models that accept it. + if temperature is not None and _accepts_sampling_params(model): + kwargs["temperature"] = temperature + if tools: + kwargs["tools"] = tools + return await client.messages.create(**kwargs) + + @staticmethod + async def _run_loop( + client: AsyncAnthropic, + *, + model: str, + max_tokens: int, + temperature: Optional[float], + system: str, + initial_user_content: str, + tools: List[Dict[str, Any]], + ) -> Dict[str, Any]: + """ + Drive the agentic loop until the model stops requesting tools. + + Returns a dict with the final text, token usage, and iteration count. + """ + messages: List[Dict[str, Any]] = [{"role": "user", "content": initial_user_content}] + total_input = 0 + total_output = 0 + + for iteration in range(1, MAX_AGENT_ITERATIONS + 1): + response = await AgentRuntime._create_message( + client, + model=model, + max_tokens=max_tokens, + temperature=temperature, + system=system, + messages=messages, + tools=tools, + ) + + if response.usage: + total_input += response.usage.input_tokens or 0 + total_output += response.usage.output_tokens or 0 + + # Server-side tools may pause; re-send to resume. + if response.stop_reason == "pause_turn": + messages.append({"role": "assistant", "content": response.content}) + continue + + tool_use_blocks = [b for b in response.content if b.type == "tool_use"] + + if response.stop_reason != "tool_use" or not tool_use_blocks: + # Done — gather the final text. + text = "".join( + block.text for block in response.content if block.type == "text" + ).strip() + return { + "text": text, + "usage": {"input_tokens": total_input, "output_tokens": total_output}, + "iterations": iteration, + "stop_reason": response.stop_reason, + } + + # Execute requested tools and feed the results back. + messages.append({"role": "assistant", "content": response.content}) + tool_results = [] + for block in tool_use_blocks: + handler = TOOL_HANDLERS.get(block.name) + if handler is None: + result_content = f"Error: tool '{block.name}' is not available." + is_error = True + else: + try: + result_content = await handler(block.input) + is_error = False + except Exception as exc: # noqa: BLE001 - report tool failure to model + logger.warning("Tool '%s' failed: %s", block.name, exc) + result_content = f"Error executing tool '{block.name}': {exc}" + is_error = True + tool_results.append( + { + "type": "tool_result", + "tool_use_id": block.id, + "content": result_content, + "is_error": is_error, + } + ) + messages.append({"role": "user", "content": tool_results}) + + # Loop budget exhausted. + return { + "text": "", + "usage": {"input_tokens": total_input, "output_tokens": total_output}, + "iterations": MAX_AGENT_ITERATIONS, + "stop_reason": "max_iterations", + "error": "Agent stopped after reaching the maximum number of tool iterations.", + } + + # ------------------------------------------------------------------ # + # Public API (matches the former CrewService surface) + # ------------------------------------------------------------------ # + @staticmethod + async def execute_agent_task( + db: AsyncSession, + agent: Agent, + task_description: str, + task_input: Dict[str, Any] = None, + ) -> Dict[str, Any]: + """ + Execute a task with a single agent using the Claude tool-use loop. + + Args: + db: Database session (unused today; kept for signature compatibility + and future tool handlers that need DB access). + agent: Agent to execute the task. + task_description: The task prompt. + task_input: Optional structured input merged into the prompt. + + Returns: + Result dict: {success, result/error, agent_id, task_description, + model_used, usage, iterations}. + """ + model_string = agent.model or settings.default_model + model_id, api_key, error = AgentRuntime._resolve_model(model_string) + + if error: + return { + "success": False, + "error": error["error"], + "agent_id": agent.id, + "task_description": task_description, + } + + try: + user_content = AgentRuntime._compose_task_prompt(task_description, task_input) + except ValueError as exc: + logger.warning("Rejected task prompt for agent '%s': %s", agent.name, exc) + return { + "success": False, + "error": "Invalid task content detected.", + "agent_id": agent.id, + "task_description": task_description, + } + + system = AgentRuntime._build_system_prompt(agent) + tools = AgentRuntime.build_tools(agent) + temperature = ( + agent.temperature if agent.temperature is not None else settings.default_temperature + ) + max_tokens = agent.max_tokens or settings.default_max_tokens + + client = AgentRuntime._client(api_key) + try: + logger.info("Executing task for agent '%s' with model '%s'", agent.name, model_id) + loop_result = await AgentRuntime._run_loop( + client, + model=model_id, + max_tokens=max_tokens, + temperature=temperature, + system=system, + initial_user_content=user_content, + tools=tools, + ) + except RateLimitError as exc: + logger.error("Rate limited executing agent task: %s", exc) + return AgentRuntime._error_result(agent, task_description, str(exc)) + except (APIConnectionError, httpx.TimeoutException) as exc: + logger.error("Connection/timeout executing agent task: %s", exc) + return AgentRuntime._error_result(agent, task_description, str(exc)) + except APIError as exc: + logger.error("Anthropic API error executing agent task: %s", exc) + return AgentRuntime._error_result(agent, task_description, str(exc)) + except Exception as exc: # noqa: BLE001 - surface unexpected failures + logger.error("Task execution failed: %s", exc, exc_info=True) + return AgentRuntime._error_result(agent, task_description, str(exc)) + finally: + await client.close() + + if loop_result.get("error"): + return AgentRuntime._error_result( + agent, + task_description, + loop_result["error"], + extra={ + "iterations": loop_result.get("iterations"), + }, + ) + + return { + "success": True, + "result": loop_result["text"], + "agent_id": agent.id, + "task_description": task_description, + "model_used": model_id, + "usage": loop_result["usage"], + "iterations": loop_result["iterations"], + } + + @staticmethod + async def execute_multi_agent_task( + db: AsyncSession, + agents: List[Agent], + task_descriptions: List[str], + process: str = "sequential", + ) -> Dict[str, Any]: + """ + Execute tasks across multiple agents. + + The agents collaborate by chaining: each agent receives the prior + agents' outputs as context, then performs its own task. This reproduces + CrewAI's "sequential" process. "hierarchical" currently runs the same + sequential chain with a coordination note — true delegation can be added + later via a delegation tool. + + Args: + db: Database session. + agents: Agents to run, in order. + task_descriptions: One task per agent. + process: "sequential" or "hierarchical". + + Returns: + Result dict: {success, result, agents, process, model_used, usage}. + """ + if len(agents) != len(task_descriptions): + raise ValueError("Number of agents must match number of tasks") + + agent_summaries = [{"id": a.id, "name": a.name} for a in agents] + transcript: List[str] = [] + total_input = 0 + total_output = 0 + last_model: Optional[str] = None + + coordination_note = ( + "You are part of a team working through tasks in sequence. " + "Build on the prior team members' results below.\n\n" + if process == "hierarchical" + else "" + ) + + for agent, task_desc in zip(agents, task_descriptions): + context = "" + if transcript: + context = ( + "Results from previous team members:\n" + + "\n\n".join(transcript) + + "\n\n---\n\n" + ) + combined = f"{coordination_note}{context}Your task: {task_desc}" + + result = await AgentRuntime.execute_agent_task(db, agent, combined) + if not result["success"]: + return { + "success": False, + "error": result["error"], + "agents": agent_summaries, + "process": process, + "failed_agent_id": agent.id, + } + + last_model = result.get("model_used", last_model) + usage = result.get("usage", {}) + total_input += usage.get("input_tokens", 0) + total_output += usage.get("output_tokens", 0) + transcript.append(f"[{agent.name}]\n{result['result']}") + + return { + "success": True, + "result": transcript[-1].split("\n", 1)[-1] if transcript else "", + "transcript": transcript, + "agents": agent_summaries, + "process": process, + "model_used": last_model, + "usage": {"input_tokens": total_input, "output_tokens": total_output}, + } + + # ------------------------------------------------------------------ # + # Small helpers + # ------------------------------------------------------------------ # + @staticmethod + def _compose_task_prompt(task_description: str, task_input: Optional[Dict[str, Any]]) -> str: + """Sanitize and assemble the user prompt from the task and optional input.""" + safe_task = PromptSanitizer.sanitize_prompt(task_description, raise_on_detection=True) + if task_input: + # Render structured input as readable context. + input_lines = "\n".join(f"- {k}: {v}" for k, v in task_input.items()) + return f"{safe_task}\n\nAdditional input:\n{input_lines}" + return safe_task + + @staticmethod + def _error_result( + agent: Agent, + task_description: str, + error: str, + extra: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + result = { + "success": False, + "error": error, + "agent_id": agent.id, + "task_description": task_description, + } + if extra: + result.update(extra) + return result diff --git a/backend/app/services/crew_service.py b/backend/app/services/crew_service.py deleted file mode 100644 index 9aa3a19..0000000 --- a/backend/app/services/crew_service.py +++ /dev/null @@ -1,325 +0,0 @@ -""" -ABOUTME: CrewAI service for multi-agent orchestration and collaboration. -ABOUTME: Supports multiple LLM providers via ModelValidator and ProviderRegistry. -ABOUTME: API keys are ONLY read from environment variables - NEVER from database. -""" - -import logging -from typing import Any, Dict, List - -from app.models.agent import Agent, AgentType -from app.services.model_validator import ModelValidator, model_validator -from app.services.provider_registry import provider_registry -from config.settings import settings -from sqlalchemy.ext.asyncio import AsyncSession - -logger = logging.getLogger(__name__) - -# Try to import CrewAI, but provide stubs if not available -try: - from crewai import Agent as CrewAgent - from crewai import Crew, LLM, Process - from crewai import Task as CrewTask - - CREWAI_AVAILABLE = True -except ImportError: - CREWAI_AVAILABLE = False - # Provide stub classes for type checking - CrewAgent = None - CrewTask = None - Crew = None - Process = None - LLM = None - - -class CrewService: - """Service for CrewAI agent orchestration with multi-provider support.""" - - @staticmethod - def _map_agent_type_to_role(agent_type: AgentType) -> str: - """Map agent type to CrewAI role description.""" - role_mapping = { - AgentType.CONVERSATIONAL: "Customer Support Specialist", - AgentType.ANALYTICAL: "Data Analyst and Researcher", - AgentType.CREATIVE: "Creative Content Writer", - AgentType.AUTOMATION: "Automation and Workflow Specialist", - } - return role_mapping.get(agent_type, "General Purpose Assistant") - - @staticmethod - def _create_crew_agent(agent: Agent, llm_instance: Any) -> CrewAgent: - """ - Create a CrewAI agent from database agent model. - - Args: - agent: Database agent model - llm_instance: LLM instance for the agent - - Returns: - CrewAI Agent instance - """ - role = CrewService._map_agent_type_to_role(agent.agent_type) - - return CrewAgent( - role=role, - goal=agent.description, - backstory=agent.system_prompt, - verbose=True, - allow_delegation=True, - llm=llm_instance, - ) - - @staticmethod - def _resolve_llm( - model_string: str, validator: ModelValidator = None - ) -> tuple[Any, str, Dict[str, Any]]: - """ - Resolve and create an LLM instance from a model string. - - SECURITY: API keys are ONLY read from environment variables. - They are NEVER stored in the database. - - Args: - model_string: Model string (e.g., "anthropic/claude-3-5-sonnet-20241022" or "GPT-4") - validator: ModelValidator instance (optional) - - Returns: - Tuple of (LLM instance, normalized model string, error dict or None) - """ - validator = validator or model_validator - - # Validate and normalize the model string - validation = validator.validate_model(model_string, check_configured=True) - - if not validation.is_valid: - return None, None, {"error": validation.error} - - # Get API key from environment (NEVER from database) - api_key = provider_registry.get_api_key(validation.provider) - - if not api_key: - provider_config = provider_registry.get_provider(validation.provider) - env_var = provider_config.api_key_env if provider_config else "API_KEY" - return None, None, { - "error": f"API key not configured for provider '{validation.provider}'. " - f"Set {env_var} environment variable." - } - - logger.info( - f"Creating LLM for provider '{validation.provider}' " - f"with model '{validation.model}'" - ) - - return validation.normalized, api_key, None - - @staticmethod - async def execute_agent_task( - db: AsyncSession, - agent: Agent, - task_description: str, - task_input: Dict[str, Any] = None, - ) -> Dict[str, Any]: - """ - Execute a task with a single agent. - - Supports multiple LLM providers (Anthropic, OpenAI, Mistral). - - Args: - db: Database session - agent: Agent to execute task - task_description: Task description - task_input: Task input data - - Returns: - Task execution result - """ - if not CREWAI_AVAILABLE: - return { - "success": False, - "error": "CrewAI is not available. Please install crewai and crewai-tools packages.", - "agent_id": agent.id, - "task_description": task_description, - } - - # Resolve model and get API key (from environment only) - model_string = agent.model or settings.default_model - normalized_model, api_key, error = CrewService._resolve_llm(model_string) - - if error: - return { - "success": False, - "error": error["error"], - "agent_id": agent.id, - "task_description": task_description, - } - - # Create LLM instance - llm = LLM( - model=normalized_model, - api_key=api_key, - temperature=agent.temperature or settings.default_temperature, - max_tokens=agent.max_tokens or settings.default_max_tokens, - ) - - # Create CrewAI agent - crew_agent = CrewService._create_crew_agent(agent, llm) - - # Create task - crew_task = CrewTask( - description=task_description, - agent=crew_agent, - expected_output="Detailed result of the task execution", - ) - - # Create crew with single agent (memory disabled to avoid OpenAI embeddings) - crew = Crew( - agents=[crew_agent], - tasks=[crew_task], - process=Process.sequential, - verbose=True, - memory=False, # Disable memory to avoid OpenAI embeddings requirement - ) - - # Execute - try: - logger.info( - f"Executing task for agent '{agent.name}' with model '{normalized_model}'" - ) - result = crew.kickoff() - - return { - "success": True, - "result": str(result), - "agent_id": agent.id, - "task_description": task_description, - "model_used": normalized_model, - } - - except Exception as e: - logger.error(f"Task execution failed: {e}", exc_info=True) - return { - "success": False, - "error": str(e), - "agent_id": agent.id, - "task_description": task_description, - } - - @staticmethod - async def execute_multi_agent_task( - db: AsyncSession, - agents: List[Agent], - task_descriptions: List[str], - process: str = "sequential", - ) -> Dict[str, Any]: - """ - Execute tasks with multiple agents collaborating. - - Supports multiple LLM providers (Anthropic, OpenAI, Mistral). - - Args: - db: Database session - agents: List of agents to collaborate - task_descriptions: List of task descriptions (one per agent) - process: "sequential" or "hierarchical" - - Returns: - Execution result - """ - if not CREWAI_AVAILABLE: - return { - "success": False, - "error": "CrewAI is not available. Please install crewai and crewai-tools packages.", - "agents": [{"id": a.id, "name": a.name} for a in agents], - } - - if len(agents) != len(task_descriptions): - raise ValueError("Number of agents must match number of tasks") - - # Use first agent's model as default for multi-agent crew - primary_agent = agents[0] - model_string = primary_agent.model or settings.default_model - normalized_model, api_key, error = CrewService._resolve_llm(model_string) - - if error: - return { - "success": False, - "error": error["error"], - "agents": [{"id": a.id, "name": a.name} for a in agents], - } - - # Create shared LLM instance - llm = LLM( - model=normalized_model, - api_key=api_key, - temperature=primary_agent.temperature or settings.default_temperature, - max_tokens=primary_agent.max_tokens or settings.default_max_tokens, - ) - - # Create CrewAI agents - crew_agents = [CrewService._create_crew_agent(agent, llm) for agent in agents] - - # Create tasks - crew_tasks = [ - CrewTask( - description=task_desc, - agent=crew_agents[i], - expected_output="Detailed task result", - ) - for i, task_desc in enumerate(task_descriptions) - ] - - # Determine process type - process_type = ( - Process.hierarchical if process == "hierarchical" else Process.sequential - ) - - # Create crew (memory disabled to avoid OpenAI embeddings requirement) - crew = Crew( - agents=crew_agents, - tasks=crew_tasks, - process=process_type, - verbose=True, - memory=False, - ) - - # Execute - try: - logger.info( - f"Executing multi-agent task with {len(agents)} agents, " - f"model '{normalized_model}', process '{process}'" - ) - result = crew.kickoff() - - return { - "success": True, - "result": str(result), - "agents": [{"id": a.id, "name": a.name} for a in agents], - "process": process, - "model_used": normalized_model, - } - - except Exception as e: - logger.error(f"Multi-agent task execution failed: {e}", exc_info=True) - return { - "success": False, - "error": str(e), - "agents": [{"id": a.id, "name": a.name} for a in agents], - } - - @staticmethod - def create_agent_tools(agent: Agent) -> List[Any]: - """ - Create tools for an agent based on its configuration. - - Args: - agent: Agent model - - Returns: - List of CrewAI tools - """ - tools = [] - - # Based on agent.tools_config, create appropriate tools - # This will be implemented when we add external integrations - - return tools diff --git a/backend/app/services/llm_service.py b/backend/app/services/llm_service.py index f3293f1..6998b9b 100644 --- a/backend/app/services/llm_service.py +++ b/backend/app/services/llm_service.py @@ -25,6 +25,23 @@ # Opens after 5 failures, stays open for 60s llm_breaker = CircuitBreaker(fail_max=5, reset_timeout=60, name="llm_service") +# Claude 4.6+ / Fable models reject sampling params (temperature/top_p/top_k). +# Sending them returns a 400, so omit temperature for these model families. +_NO_SAMPLING_PARAM_PREFIXES = ( + "claude-opus-4-6", + "claude-opus-4-7", + "claude-opus-4-8", + "claude-sonnet-4-6", + "claude-haiku-4-5", + "claude-fable-5", + "claude-mythos-5", +) + + +def _accepts_sampling_params(model: str) -> bool: + """Return True if the model accepts temperature/top_p (older Claude models).""" + return not any(model.startswith(prefix) for prefix in _NO_SAMPLING_PARAM_PREFIXES) + def get_anthropic_api_key() -> str: """ @@ -163,14 +180,17 @@ async def generate( f"Generating with model {model}, temp={temperature}, max_tokens={max_tokens}" ) - response = await self.async_client.messages.create( - model=model, - max_tokens=max_tokens, - temperature=temperature, - system=sanitized_system if sanitized_system else "", - messages=[{"role": "user", "content": sanitized_prompt}], + create_kwargs = { + "model": model, + "max_tokens": max_tokens, + "system": sanitized_system if sanitized_system else "", + "messages": [{"role": "user", "content": sanitized_prompt}], **kwargs, - ) + } + if _accepts_sampling_params(model): + create_kwargs["temperature"] = temperature + + response = await self.async_client.messages.create(**create_kwargs) logger.info( f"LLM generation successful: {response.usage.input_tokens} in, {response.usage.output_tokens} out" @@ -260,14 +280,17 @@ async def generate_stream( try: logger.debug(f"Streaming with model {model}") - async with self.async_client.messages.stream( - model=model, - max_tokens=max_tokens, - temperature=temperature, - system=sanitized_system if sanitized_system else "", - messages=[{"role": "user", "content": sanitized_prompt}], + stream_kwargs = { + "model": model, + "max_tokens": max_tokens, + "system": sanitized_system if sanitized_system else "", + "messages": [{"role": "user", "content": sanitized_prompt}], **kwargs, - ) as stream: + } + if _accepts_sampling_params(model): + stream_kwargs["temperature"] = temperature + + async with self.async_client.messages.stream(**stream_kwargs) as stream: async for text in stream.text_stream: yield text @@ -293,9 +316,9 @@ async def validate_api_key(self, api_key: str) -> bool: api_key=api_key, http_client=httpx.AsyncClient(timeout=httpx.Timeout(connect=5.0, read=10.0)), ) - # Make a minimal test request + # Make a minimal test request with a current, low-cost model response = await test_client.messages.create( - model="claude-3-5-sonnet-20241022", + model="claude-haiku-4-5", max_tokens=10, messages=[{"role": "user", "content": "test"}], ) @@ -332,11 +355,14 @@ def estimate_cost(self, input_tokens: int, output_tokens: int, model: str = None """ model = model or settings.default_model - # Pricing as of 2024 (per million tokens) + # Pricing per million tokens (current Claude models + legacy fallbacks) pricing = { + "claude-fable-5": {"input": 10.00, "output": 50.00}, + "claude-opus-4-8": {"input": 5.00, "output": 25.00}, + "claude-sonnet-4-6": {"input": 3.00, "output": 15.00}, + "claude-haiku-4-5": {"input": 1.00, "output": 5.00}, + # Legacy "claude-3-5-sonnet-20241022": {"input": 3.00, "output": 15.00}, - "claude-3-opus-20240229": {"input": 15.00, "output": 75.00}, - "claude-3-sonnet-20240229": {"input": 3.00, "output": 15.00}, "claude-3-haiku-20240307": {"input": 0.25, "output": 1.25}, } diff --git a/backend/app/services/memory_service.py b/backend/app/services/memory_service.py index 7a2c483..3c304cb 100644 --- a/backend/app/services/memory_service.py +++ b/backend/app/services/memory_service.py @@ -31,12 +31,8 @@ def __init__(self): self.conversations_table = self.lance.get_or_create_table( "conversations", ConversationSchema ) - self.outputs_table = self.lance.get_or_create_table( - "agent_outputs", AgentOutputSchema - ) - self.documents_table = self.lance.get_or_create_table( - "documents", DocumentSchema - ) + self.outputs_table = self.lance.get_or_create_table("agent_outputs", AgentOutputSchema) + self.documents_table = self.lance.get_or_create_table("documents", DocumentSchema) async def store_conversation( self, agent_id: str, message: str, role: str = "user", metadata: Dict[str, Any] = None @@ -159,7 +155,11 @@ def _search(): # Format results conversations = [] for row in results: - metadata = {"agent_id": row["agent_id"], "role": row["role"], "timestamp": row["timestamp"]} + metadata = { + "agent_id": row["agent_id"], + "role": row["role"], + "timestamp": row["timestamp"], + } if row.get("metadata_json"): metadata.update(json.loads(row["metadata_json"])) conversations.append( @@ -294,7 +294,11 @@ def _get_history(): # Format results conversations = [] for row in results: - metadata = {"agent_id": row["agent_id"], "role": row["role"], "timestamp": row["timestamp"]} + metadata = { + "agent_id": row["agent_id"], + "role": row["role"], + "timestamp": row["timestamp"], + } if row.get("metadata_json"): metadata.update(json.loads(row["metadata_json"])) conversations.append( diff --git a/backend/app/services/model_validator.py b/backend/app/services/model_validator.py index c7ca2f3..3fe7d9b 100644 --- a/backend/app/services/model_validator.py +++ b/backend/app/services/model_validator.py @@ -44,19 +44,32 @@ class ModelValidator: "GPT-4o": "openai/gpt-4o", "GPT-3.5": "openai/gpt-3.5-turbo", "GPT-3.5-Turbo": "openai/gpt-3.5-turbo", + # Anthropic current names + "claude-opus-4.8": "anthropic/claude-opus-4-8", + "claude-opus-4-8": "anthropic/claude-opus-4-8", + "claude-sonnet-4.6": "anthropic/claude-sonnet-4-6", + "claude-sonnet-4-6": "anthropic/claude-sonnet-4-6", + "claude-haiku-4.5": "anthropic/claude-haiku-4-5", + "claude-haiku-4-5": "anthropic/claude-haiku-4-5", # Anthropic legacy names - "claude-3-opus": "anthropic/claude-3-opus-20240229", - "claude-3-sonnet": "anthropic/claude-3-sonnet-20240229", + "claude-3-opus": "anthropic/claude-opus-4-8", # retired -> current Opus + "claude-3-sonnet": "anthropic/claude-sonnet-4-6", # retired -> current Sonnet "claude-3-haiku": "anthropic/claude-3-haiku-20240307", - "claude-3.5-sonnet": "anthropic/claude-3-5-sonnet-20241022", - "claude-3.5-haiku": "anthropic/claude-3-5-haiku-20241022", - "claude-3.7-sonnet": "anthropic/claude-3-7-sonnet-20250219", + "claude-3.5-sonnet": "anthropic/claude-sonnet-4-6", # retired -> current Sonnet + "claude-3.5-haiku": "anthropic/claude-haiku-4-5", # retired -> current Haiku + "claude-3.7-sonnet": "anthropic/claude-sonnet-4-6", # retired -> current Sonnet "claude-sonnet-4": "anthropic/claude-sonnet-4-20250514", "claude-opus-4": "anthropic/claude-opus-4-20250514", + # Retired snapshot IDs remapped to current models (seeded agents keep working) + "claude-3-5-sonnet-20241022": "anthropic/claude-sonnet-4-6", + "claude-3-5-sonnet-20240620": "anthropic/claude-sonnet-4-6", + "claude-3-7-sonnet-20250219": "anthropic/claude-sonnet-4-6", + "claude-3-5-haiku-20241022": "anthropic/claude-haiku-4-5", + "claude-3-opus-20240229": "anthropic/claude-opus-4-8", # Common informal names - "Claude-3": "anthropic/claude-3-5-sonnet-20241022", - "Claude-3.5": "anthropic/claude-3-5-sonnet-20241022", - "Claude": "anthropic/claude-3-5-sonnet-20241022", + "Claude-3": "anthropic/claude-sonnet-4-6", + "Claude-3.5": "anthropic/claude-sonnet-4-6", + "Claude": "anthropic/claude-opus-4-8", # Mistral legacy names "mistral-large": "mistral/mistral-large-latest", "mistral-medium": "mistral/mistral-medium-latest", @@ -175,9 +188,7 @@ def validate_model(self, model: str, check_configured: bool = True) -> Validatio # Step 4: Validate provider exists provider_config = self.registry.get_provider(provider) if not provider_config: - available_providers = ", ".join( - p.name for p in self.registry.list_providers() - ) + available_providers = ", ".join(p.name for p in self.registry.list_providers()) return ValidationResult( is_valid=False, provider=provider, diff --git a/backend/app/services/provider_registry.py b/backend/app/services/provider_registry.py index 5a12090..3767481 100644 --- a/backend/app/services/provider_registry.py +++ b/backend/app/services/provider_registry.py @@ -93,7 +93,52 @@ def _initialize_providers(self) -> None: api_key_env="ANTHROPIC_API_KEY", fallback_env="PERSONAL_Q_API_KEY", models=[ - # Claude 4 Series (Latest) + # Claude 4.6+ / Fable (current — tool-use agent runtime) + ModelInfo( + id="claude-opus-4-8", + display_name="Claude Opus 4.8", + context_window=1000000, + max_output_tokens=128000, + supports_vision=True, + supports_tools=True, + cost_per_1k_input=0.005, + cost_per_1k_output=0.025, + is_recommended=True, + ), + ModelInfo( + id="claude-sonnet-4-6", + display_name="Claude Sonnet 4.6", + context_window=1000000, + max_output_tokens=64000, + supports_vision=True, + supports_tools=True, + cost_per_1k_input=0.003, + cost_per_1k_output=0.015, + is_recommended=False, + ), + ModelInfo( + id="claude-haiku-4-5", + display_name="Claude Haiku 4.5", + context_window=200000, + max_output_tokens=64000, + supports_vision=True, + supports_tools=True, + cost_per_1k_input=0.001, + cost_per_1k_output=0.005, + is_recommended=False, + ), + ModelInfo( + id="claude-fable-5", + display_name="Claude Fable 5", + context_window=1000000, + max_output_tokens=128000, + supports_vision=True, + supports_tools=True, + cost_per_1k_input=0.010, + cost_per_1k_output=0.050, + is_recommended=False, + ), + # Claude 4 Series (legacy — accept sampling params) ModelInfo( id="claude-sonnet-4-20250514", display_name="Claude Sonnet 4", @@ -103,7 +148,7 @@ def _initialize_providers(self) -> None: supports_tools=True, cost_per_1k_input=0.003, cost_per_1k_output=0.015, - is_recommended=True, + is_recommended=False, ), ModelInfo( id="claude-opus-4-20250514", diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index 35031ee..fd348e6 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -17,10 +17,15 @@ def _init_database_sync(): Called at module load time to ensure tables exist before any tasks run. """ try: - from app.db.database import Base - from app.models import Agent, Task, Activity, APIKey, Schedule # noqa: F401 - Import to register models - import sqlalchemy + from app.db.database import Base + from app.models import ( # noqa: F401 - Import to register models + Activity, + Agent, + APIKey, + Schedule, + Task, + ) # Determine sync URL based on database type db_url = settings.database_url diff --git a/backend/app/workers/tasks.py b/backend/app/workers/tasks.py index a188779..c98cf79 100644 --- a/backend/app/workers/tasks.py +++ b/backend/app/workers/tasks.py @@ -10,7 +10,7 @@ from app.models.agent import Agent from app.models.task import Task as TaskModel from app.models.task import TaskStatus -from app.services.crew_service import CrewService +from app.services.agent_runtime import AgentRuntime from app.utils.datetime_utils import utcnow_naive from app.workers.celery_app import celery_app from celery import Task @@ -78,8 +78,8 @@ async def execute_agent_task(self, task_id: str): ) try: - # Execute with CrewAI - result = await CrewService.execute_agent_task( + # Execute with the Claude agent runtime (Anthropic SDK tool-use loop) + result = await AgentRuntime.execute_agent_task( db, agent, task.description or task.title, task.input_data ) diff --git a/backend/config/settings.py b/backend/config/settings.py index 2e20fc6..d14a535 100644 --- a/backend/config/settings.py +++ b/backend/config/settings.py @@ -67,7 +67,7 @@ def validate_debug_mode(cls, v: bool, info) -> bool: memory_retention_days: int = 90 # LLM Defaults - default_model: str = "claude-3-5-sonnet-20241022" + default_model: str = "claude-opus-4-8" default_temperature: float = 0.7 default_max_tokens: int = 4096 diff --git a/backend/pyproject.toml b/backend/pyproject.toml index aa533bb..462c8a8 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -46,10 +46,6 @@ dependencies = [ "cryptography==43.0.3", "slowapi==0.1.9", "itsdangerous>=2.1.2", - "langchain-anthropic>=0.1.23,<0.3.0", - "langchain-core>=0.2.26,<0.3.0", - "crewai>=0.86.0", - "litellm>=1.50.0", ] [project.optional-dependencies] diff --git a/backend/requirements.txt b/backend/requirements.txt index 1370566..070f0bc 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -12,13 +12,10 @@ aiosqlite==0.20.0 # ChromaDB for vector storage chromadb==0.5.18 -# LLM Integration -anthropic==0.39.0 - -# CrewAI - Multi-agent orchestration -crewai==0.203.1 -langchain-anthropic==0.3.0 # Required for ChatAnthropic integration -langchain-core==0.3.0 # Core LangChain components +# LLM Integration / Agent runtime +# The Claude agent runtime uses the Anthropic SDK directly (Messages API + +# tool-use loop). No CrewAI / LangChain / LiteLLM required. +anthropic>=0.39.0 # Task Queue celery==5.4.0 diff --git a/backend/tests/integration/test_api_key_config.py b/backend/tests/integration/test_api_key_config.py index 4bddbad..1a03f75 100644 --- a/backend/tests/integration/test_api_key_config.py +++ b/backend/tests/integration/test_api_key_config.py @@ -4,7 +4,7 @@ Tests the environment-based API key configuration: - get_anthropic_api_key() function in llm_service.py - /api-key-status endpoint in settings.py -- CrewService behavior with/without API key +- AgentRuntime behavior with/without API key """ import pytest @@ -108,140 +108,133 @@ async def test_api_key_status_endpoint_empty_string(self, test_app): assert data["configured"] is False -class TestCrewServiceApiKeyHandling: - """Integration tests for CrewService API key handling.""" +class TestAgentRuntimeApiKeyHandling: + """Integration tests for AgentRuntime API key handling.""" - @pytest.mark.asyncio - async def test_crew_service_fails_gracefully_without_api_key(self): - """Test CrewService returns error dict when API key not configured.""" - from app.services.crew_service import CrewService, CREWAI_AVAILABLE + @staticmethod + def _make_agent(**overrides): from app.models.agent import Agent, AgentType, AgentStatus - if not CREWAI_AVAILABLE: - pytest.skip("CrewAI not available") - - # Mock get_anthropic_api_key to raise ValueError - with patch("app.services.crew_service.get_anthropic_api_key") as mock_get_key: - mock_get_key.side_effect = ValueError( - "PERSONAL_Q_API_KEY environment variable is not set. " - "This is required for agent execution." - ) - - agent = Agent( - id="test-agent-no-key", - name="Test Agent", - description="Test agent for API key failure", - agent_type=AgentType.CONVERSATIONAL, - model="claude-3-5-sonnet-20241022", - system_prompt="You are a test agent.", - temperature=0.7, - max_tokens=2048, - status=AgentStatus.ACTIVE, - ) + defaults = dict( + id="test-agent-no-key", + name="Test Agent", + description="Test agent for API key failure", + agent_type=AgentType.CONVERSATIONAL, + model="anthropic/claude-opus-4-8", + system_prompt="You are a test agent.", + temperature=0.7, + max_tokens=2048, + status=AgentStatus.ACTIVE, + ) + defaults.update(overrides) + return Agent(**defaults) - mock_db = Mock() - result = await CrewService.execute_agent_task( - db=mock_db, + @pytest.mark.asyncio + async def test_agent_runtime_fails_gracefully_without_api_key(self): + """AgentRuntime returns an error dict when no Anthropic API key is configured.""" + from app.services.agent_runtime import AgentRuntime + from app.schemas.llm import ValidationResult + + agent = self._make_agent() + + # Model validates as a configured Anthropic model, but the key lookup + # returns None — exercising the runtime's explicit "key missing" branch. + with patch( + "app.services.agent_runtime.model_validator.validate_model", + return_value=ValidationResult( + is_valid=True, + provider="anthropic", + model="claude-opus-4-8", + normalized="anthropic/claude-opus-4-8", + ), + ), patch( + "app.services.agent_runtime.provider_registry.get_api_key", + return_value=None, + ): + result = await AgentRuntime.execute_agent_task( + db=Mock(), agent=agent, task_description="Test task that should fail", ) - # Verify graceful failure - assert result["success"] is False - assert "PERSONAL_Q_API_KEY" in result["error"] - assert "not set" in result["error"] - assert result["agent_id"] == "test-agent-no-key" + assert result["success"] is False + assert "anthropic" in result["error"].lower() + assert "api key" in result["error"].lower() + assert result["agent_id"] == "test-agent-no-key" @pytest.mark.asyncio - async def test_crew_service_multi_agent_fails_gracefully_without_api_key(self): - """Test CrewService multi-agent returns error when API key not configured.""" - from app.services.crew_service import CrewService, CREWAI_AVAILABLE - from app.models.agent import Agent, AgentType, AgentStatus - - if not CREWAI_AVAILABLE: - pytest.skip("CrewAI not available") - - with patch("app.services.crew_service.get_anthropic_api_key") as mock_get_key: - mock_get_key.side_effect = ValueError( - "PERSONAL_Q_API_KEY environment variable is not set." + async def test_agent_runtime_rejects_non_anthropic_provider(self): + """AgentRuntime returns a clear error for non-Anthropic providers.""" + from app.services.agent_runtime import AgentRuntime + from app.schemas.llm import ValidationResult + + agent = self._make_agent(model="openai/gpt-4o") + + with patch( + "app.services.agent_runtime.model_validator.validate_model", + return_value=ValidationResult( + is_valid=True, + provider="openai", + model="gpt-4o", + normalized="openai/gpt-4o", + ), + ): + result = await AgentRuntime.execute_agent_task( + db=Mock(), + agent=agent, + task_description="Test task", ) - agents = [ - Agent( - id="agent-1", - name="Agent 1", - description="First test agent", - agent_type=AgentType.ANALYTICAL, - model="claude-3-5-sonnet-20241022", - system_prompt="Test agent 1", - temperature=0.7, - max_tokens=2048, - status=AgentStatus.ACTIVE, - ), - Agent( - id="agent-2", - name="Agent 2", - description="Second test agent", - agent_type=AgentType.CREATIVE, - model="claude-3-5-sonnet-20241022", - system_prompt="Test agent 2", - temperature=0.8, - max_tokens=2048, - status=AgentStatus.ACTIVE, - ), - ] - - mock_db = Mock() - result = await CrewService.execute_multi_agent_task( - db=mock_db, + assert result["success"] is False + assert "not supported" in result["error"].lower() + + @pytest.mark.asyncio + async def test_agent_runtime_multi_agent_fails_gracefully_without_api_key(self): + """Multi-agent execution returns an error when the API key is missing.""" + from app.services.agent_runtime import AgentRuntime + from app.schemas.llm import ValidationResult + + agents = [ + self._make_agent(id="agent-1", name="Agent 1"), + self._make_agent(id="agent-2", name="Agent 2"), + ] + + with patch( + "app.services.agent_runtime.model_validator.validate_model", + return_value=ValidationResult( + is_valid=True, + provider="anthropic", + model="claude-opus-4-8", + normalized="anthropic/claude-opus-4-8", + ), + ), patch( + "app.services.agent_runtime.provider_registry.get_api_key", + return_value=None, + ): + result = await AgentRuntime.execute_multi_agent_task( + db=Mock(), agents=agents, task_descriptions=["Task 1", "Task 2"], ) - # Verify graceful failure - assert result["success"] is False - assert "PERSONAL_Q_API_KEY" in result["error"] - assert len(result["agents"]) == 2 + assert result["success"] is False + assert "api key" in result["error"].lower() + assert result["failed_agent_id"] == "agent-1" @pytest.mark.asyncio - async def test_crew_service_error_message_content(self): - """Test that CrewService error message provides actionable information.""" - from app.services.crew_service import CrewService, CREWAI_AVAILABLE - from app.models.agent import Agent, AgentType, AgentStatus + async def test_agent_runtime_multi_agent_task_count_mismatch(self): + """Mismatched agent/task counts raise ValueError.""" + from app.services.agent_runtime import AgentRuntime - if not CREWAI_AVAILABLE: - pytest.skip("CrewAI not available") - - with patch("app.services.crew_service.get_anthropic_api_key") as mock_get_key: - error_message = ( - "PERSONAL_Q_API_KEY environment variable is not set. " - "This is required for agent execution." - ) - mock_get_key.side_effect = ValueError(error_message) - - agent = Agent( - id="test-agent", - name="Test Agent", - description="Test", - agent_type=AgentType.CONVERSATIONAL, - model="claude-3-5-sonnet-20241022", - system_prompt="Test", - temperature=0.7, - max_tokens=2048, - status=AgentStatus.ACTIVE, - ) + agents = [self._make_agent(id="agent-1", name="Agent 1")] - mock_db = Mock() - result = await CrewService.execute_agent_task( - db=mock_db, - agent=agent, - task_description="Test task", + with pytest.raises(ValueError, match="Number of agents must match number of tasks"): + await AgentRuntime.execute_multi_agent_task( + db=Mock(), + agents=agents, + task_descriptions=["Task 1", "Task 2"], ) - # Error message should be actionable - assert "PERSONAL_Q_API_KEY" in result["error"] - assert "required" in result["error"] - class TestApiKeyConfigIntegration: """End-to-end integration tests for API key configuration.""" diff --git a/backend/tests/unit/test_agent_runtime.py b/backend/tests/unit/test_agent_runtime.py new file mode 100644 index 0000000..ea42679 --- /dev/null +++ b/backend/tests/unit/test_agent_runtime.py @@ -0,0 +1,266 @@ +""" +Unit tests for the Claude agent runtime (Anthropic SDK tool-use loop). +""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock, patch + +import pytest +from app.models.agent import Agent, AgentStatus, AgentType +from app.schemas.llm import ValidationResult +from app.services import agent_runtime +from app.services.agent_runtime import AgentRuntime, _accepts_sampling_params + + +def _make_agent(**overrides) -> Agent: + defaults = dict( + id="agent-1", + name="Test Agent", + description="A test agent", + agent_type=AgentType.ANALYTICAL, + model="anthropic/claude-opus-4-8", + system_prompt="You analyze data.", + temperature=0.5, + max_tokens=1024, + status=AgentStatus.ACTIVE, + ) + defaults.update(overrides) + return Agent(**defaults) + + +def _text_response(text: str, *, input_tokens=10, output_tokens=5): + return SimpleNamespace( + stop_reason="end_turn", + content=[SimpleNamespace(type="text", text=text)], + usage=SimpleNamespace(input_tokens=input_tokens, output_tokens=output_tokens), + ) + + +def _tool_use_response(tool_id: str, name: str, tool_input: dict): + return SimpleNamespace( + stop_reason="tool_use", + content=[ + SimpleNamespace(type="tool_use", id=tool_id, name=name, input=tool_input) + ], + usage=SimpleNamespace(input_tokens=8, output_tokens=4), + ) + + +def _patch_model_resolution(provider="anthropic", model="claude-opus-4-8", api_key="sk-test"): + """Context managers that make _resolve_model succeed for the given provider.""" + return ( + patch( + "app.services.agent_runtime.model_validator.validate_model", + return_value=ValidationResult( + is_valid=True, + provider=provider, + model=model, + normalized=f"{provider}/{model}", + ), + ), + patch( + "app.services.agent_runtime.provider_registry.get_api_key", + return_value=api_key, + ), + ) + + +class TestHelpers: + def test_role_mapping(self): + assert "analyst" in AgentRuntime._role_for(_make_agent()).lower() + assert "support" in AgentRuntime._role_for( + _make_agent(agent_type=AgentType.CONVERSATIONAL) + ).lower() + + def test_system_prompt_includes_role_and_prompt(self): + prompt = AgentRuntime._build_system_prompt(_make_agent()) + assert "analyst" in prompt.lower() + assert "You analyze data." in prompt + + def test_build_tools_empty_by_default(self): + assert AgentRuntime.build_tools(_make_agent()) == [] + + def test_accepts_sampling_params(self): + # Modern models reject temperature/top_p + assert _accepts_sampling_params("claude-opus-4-8") is False + assert _accepts_sampling_params("claude-sonnet-4-6") is False + assert _accepts_sampling_params("claude-fable-5") is False + # Legacy models accept them + assert _accepts_sampling_params("claude-sonnet-4-20250514") is True + assert _accepts_sampling_params("claude-3-haiku-20240307") is True + + +class TestResolveModel: + def test_rejects_non_anthropic_provider(self): + validate, _ = _patch_model_resolution(provider="openai", model="gpt-4o") + with validate: + model_id, api_key, error = AgentRuntime._resolve_model("openai/gpt-4o") + assert model_id is None + assert error is not None + assert "not supported" in error["error"].lower() + + def test_returns_bare_model_id_for_anthropic(self): + validate, key = _patch_model_resolution() + with validate, key: + model_id, api_key, error = AgentRuntime._resolve_model("anthropic/claude-opus-4-8") + assert error is None + assert model_id == "claude-opus-4-8" + assert api_key == "sk-test" + + +class TestExecuteAgentTask: + @pytest.mark.asyncio + async def test_happy_path_single_call(self): + validate, key = _patch_model_resolution() + mock_client = Mock() + mock_client.messages.create = AsyncMock(return_value=_text_response("The answer is 42.")) + mock_client.close = AsyncMock() + + with validate, key, patch.object( + agent_runtime, "AsyncAnthropic", return_value=mock_client + ): + result = await AgentRuntime.execute_agent_task( + db=Mock(), agent=_make_agent(), task_description="What is the answer?" + ) + + assert result["success"] is True + assert result["result"] == "The answer is 42." + assert result["model_used"] == "claude-opus-4-8" + assert result["iterations"] == 1 + assert result["usage"]["input_tokens"] == 10 + + @pytest.mark.asyncio + async def test_omits_temperature_for_modern_models(self): + validate, key = _patch_model_resolution() + mock_client = Mock() + mock_client.messages.create = AsyncMock(return_value=_text_response("ok")) + mock_client.close = AsyncMock() + + with validate, key, patch.object( + agent_runtime, "AsyncAnthropic", return_value=mock_client + ): + await AgentRuntime.execute_agent_task( + db=Mock(), agent=_make_agent(), task_description="hi" + ) + + _, kwargs = mock_client.messages.create.call_args + assert "temperature" not in kwargs # opus-4-8 rejects it + + @pytest.mark.asyncio + async def test_includes_temperature_for_legacy_models(self): + validate, key = _patch_model_resolution(model="claude-sonnet-4-20250514") + mock_client = Mock() + mock_client.messages.create = AsyncMock(return_value=_text_response("ok")) + mock_client.close = AsyncMock() + + agent = _make_agent(model="anthropic/claude-sonnet-4-20250514", temperature=0.3) + with validate, key, patch.object( + agent_runtime, "AsyncAnthropic", return_value=mock_client + ): + await AgentRuntime.execute_agent_task( + db=Mock(), agent=agent, task_description="hi" + ) + + _, kwargs = mock_client.messages.create.call_args + assert kwargs.get("temperature") == 0.3 + + @pytest.mark.asyncio + async def test_tool_use_loop(self): + validate, key = _patch_model_resolution() + mock_client = Mock() + mock_client.messages.create = AsyncMock( + side_effect=[ + _tool_use_response("tu-1", "echo", {"value": "hello"}), + _text_response("Tool said: hello"), + ] + ) + mock_client.close = AsyncMock() + + handler = AsyncMock(return_value="hello") + + with validate, key, patch.object( + agent_runtime, "AsyncAnthropic", return_value=mock_client + ), patch.object( + AgentRuntime, + "build_tools", + return_value=[{"name": "echo", "description": "echo", "input_schema": {}}], + ), patch.dict(agent_runtime.TOOL_HANDLERS, {"echo": handler}, clear=False): + result = await AgentRuntime.execute_agent_task( + db=Mock(), agent=_make_agent(), task_description="use the echo tool" + ) + + assert result["success"] is True + assert result["result"] == "Tool said: hello" + assert result["iterations"] == 2 + handler.assert_awaited_once_with({"value": "hello"}) + + @pytest.mark.asyncio + async def test_unknown_tool_reports_error_to_model(self): + validate, key = _patch_model_resolution() + mock_client = Mock() + mock_client.messages.create = AsyncMock( + side_effect=[ + _tool_use_response("tu-1", "missing_tool", {}), + _text_response("Recovered without the tool."), + ] + ) + mock_client.close = AsyncMock() + + with validate, key, patch.object( + agent_runtime, "AsyncAnthropic", return_value=mock_client + ), patch.object( + AgentRuntime, + "build_tools", + return_value=[{"name": "missing_tool", "description": "x", "input_schema": {}}], + ): + result = await AgentRuntime.execute_agent_task( + db=Mock(), agent=_make_agent(), task_description="try a tool" + ) + + assert result["success"] is True + # Second call's messages should contain a tool_result flagged as error. + second_call_kwargs = mock_client.messages.create.call_args_list[1].kwargs + tool_result_msg = second_call_kwargs["messages"][-1] + assert tool_result_msg["content"][0]["is_error"] is True + + +class TestMultiAgentTask: + @pytest.mark.asyncio + async def test_task_count_mismatch_raises(self): + with pytest.raises(ValueError, match="must match"): + await AgentRuntime.execute_multi_agent_task( + db=Mock(), + agents=[_make_agent()], + task_descriptions=["a", "b"], + ) + + @pytest.mark.asyncio + async def test_sequential_chaining(self): + agents = [ + _make_agent(id="a1", name="Researcher"), + _make_agent(id="a2", name="Writer"), + ] + + async def fake_execute(db, agent, task_description, task_input=None): + return { + "success": True, + "result": f"{agent.name} output", + "agent_id": agent.id, + "task_description": task_description, + "model_used": "claude-opus-4-8", + "usage": {"input_tokens": 5, "output_tokens": 3}, + "iterations": 1, + } + + with patch.object(AgentRuntime, "execute_agent_task", side_effect=fake_execute): + result = await AgentRuntime.execute_multi_agent_task( + db=Mock(), + agents=agents, + task_descriptions=["research", "write"], + process="sequential", + ) + + assert result["success"] is True + assert len(result["transcript"]) == 2 + assert result["usage"]["input_tokens"] == 10 + assert result["process"] == "sequential" diff --git a/backend/tests/unit/test_crew_service.py b/backend/tests/unit/test_crew_service.py deleted file mode 100644 index b0985ad..0000000 --- a/backend/tests/unit/test_crew_service.py +++ /dev/null @@ -1,311 +0,0 @@ -""" -Unit tests for CrewAI service. -""" - -import pytest -from unittest.mock import Mock, AsyncMock, patch -from app.services.crew_service import CrewService, CREWAI_AVAILABLE -from app.models.agent import Agent, AgentType, AgentStatus - - -class TestCrewService: - """Tests for CrewAI Service.""" - - def test_crewai_availability(self): - """Test CrewAI import status.""" - # Will be True if installation succeeds - assert isinstance(CREWAI_AVAILABLE, bool) - # After our implementation, this should be True - assert CREWAI_AVAILABLE is True, "CrewAI should be available after installation" - - @pytest.mark.skipif(not CREWAI_AVAILABLE, reason="CrewAI not available") - def test_agent_type_to_role_mapping(self): - """Test agent type to role mapping.""" - role_conversational = CrewService._map_agent_type_to_role(AgentType.CONVERSATIONAL) - assert "Support" in role_conversational or "Customer" in role_conversational - - role_analytical = CrewService._map_agent_type_to_role(AgentType.ANALYTICAL) - assert "Analyst" in role_analytical or "Data" in role_analytical - - role_creative = CrewService._map_agent_type_to_role(AgentType.CREATIVE) - assert "Creative" in role_creative or "Content" in role_creative - - role_automation = CrewService._map_agent_type_to_role(AgentType.AUTOMATION) - assert "Automation" in role_automation or "Workflow" in role_automation - - @pytest.mark.skipif(not CREWAI_AVAILABLE, reason="CrewAI not available") - @patch("app.services.crew_service.CrewAgent") - def test_create_crew_agent(self, mock_crew_agent_class): - """Test creating a CrewAI agent from database model.""" - # Create test agent - agent = Agent( - id="test-123", - name="Test Agent", - description="Test description", - agent_type=AgentType.ANALYTICAL, - model="claude-3-5-sonnet-20241022", - system_prompt="You are a test agent.", - temperature=0.7, - max_tokens=2048, - status=AgentStatus.ACTIVE - ) - - # Mock LLM instance with required attributes for CrewAI - mock_llm = Mock() - mock_llm.supports_stop_words = True - mock_llm.model_name = "claude-3-5-sonnet-20241022" - - # Mock CrewAgent to avoid internal initialization issues - mock_crew_agent = Mock() - mock_crew_agent.role = "Data Analyst and Researcher" - mock_crew_agent.goal = agent.description - mock_crew_agent.backstory = agent.system_prompt - mock_crew_agent.verbose = True - mock_crew_agent.allow_delegation = True - mock_crew_agent_class.return_value = mock_crew_agent - - # Create CrewAI agent - crew_agent = CrewService._create_crew_agent(agent, mock_llm) - - # Verify agent creation - assert crew_agent is not None - assert "Analyst" in crew_agent.role or "Data" in crew_agent.role - assert crew_agent.goal == agent.description - assert crew_agent.backstory == agent.system_prompt - assert crew_agent.verbose is True - assert crew_agent.allow_delegation is True - - @pytest.mark.skipif(not CREWAI_AVAILABLE, reason="CrewAI not available") - @pytest.mark.asyncio - async def test_execute_agent_task_without_api_key(self): - """Test task execution fails gracefully without API key.""" - from app.services.llm_service import llm_service - - # Ensure API key is not set - original_key = llm_service.api_key - llm_service.api_key = None - - try: - agent = Agent( - id="test-123", - name="Test Agent", - description="Test", - agent_type=AgentType.CONVERSATIONAL, - model="claude-3-5-sonnet-20241022", - system_prompt="Test", - temperature=0.7, - max_tokens=2048, - status=AgentStatus.ACTIVE - ) - - result = await CrewService.execute_agent_task( - db=Mock(), - agent=agent, - task_description="Test task" - ) - - # Should return error about missing API key - assert result["success"] is False - assert "API key not configured" in result["error"] - finally: - # Restore original key - llm_service.api_key = original_key - - @pytest.mark.skipif(not CREWAI_AVAILABLE, reason="CrewAI not available") - @pytest.mark.asyncio - async def test_execute_multi_agent_task_validation(self): - """Test multi-agent task validation.""" - from app.services.llm_service import llm_service - - # Ensure API key is not set - original_key = llm_service.api_key - llm_service.api_key = None - - try: - agents = [ - Agent( - id="agent-1", - name="Agent 1", - description="Test", - agent_type=AgentType.ANALYTICAL, - model="claude-3-5-sonnet-20241022", - system_prompt="Test", - temperature=0.7, - max_tokens=2048, - status=AgentStatus.ACTIVE - ) - ] - - tasks = ["Task 1", "Task 2"] # Mismatch: 1 agent, 2 tasks - - with pytest.raises(ValueError, match="Number of agents must match number of tasks"): - await CrewService.execute_multi_agent_task( - db=Mock(), - agents=agents, - task_descriptions=tasks - ) - finally: - llm_service.api_key = original_key - - @pytest.mark.skipif(not CREWAI_AVAILABLE, reason="CrewAI not available") - def test_create_agent_tools(self): - """Test creating agent tools.""" - agent = Agent( - id="test-123", - name="Test Agent", - description="Test", - agent_type=AgentType.AUTOMATION, - model="claude-3-5-sonnet-20241022", - system_prompt="Test", - temperature=0.7, - max_tokens=2048, - status=AgentStatus.ACTIVE - ) - - tools = CrewService.create_agent_tools(agent) - - # Should return empty list for now (tools not yet implemented) - assert isinstance(tools, list) - - def test_crewai_not_available_fallback(self): - """Test that service handles CrewAI not being available.""" - if not CREWAI_AVAILABLE: - # This test only makes sense if CrewAI is not available - # In our case, after implementation, CREWAI_AVAILABLE should be True - # But we keep this test for completeness - pytest.skip("CrewAI is available - this test checks unavailable scenario") - - # If we reach here, CrewAI is available, which is the expected state - assert CREWAI_AVAILABLE is True - - -class TestCrewServiceIntegration: - """Integration tests for CrewAI service with mock LLM.""" - - @pytest.mark.skipif(not CREWAI_AVAILABLE, reason="CrewAI not available") - @pytest.mark.asyncio - @patch("app.services.crew_service.CrewAgent") - @patch("app.services.crew_service.ChatAnthropic") - @patch("app.services.crew_service.llm_service") - async def test_execute_agent_task_with_mocked_llm(self, mock_llm_service, mock_chat_anthropic, mock_crew_agent_class): - """Test executing a task with mocked LLM.""" - # Configure mocks - mock_llm_service.api_key = "test-api-key" - - # Mock LLM instance with all required attributes - mock_llm = Mock() - mock_llm.supports_stop_words = True - mock_llm.model_name = "claude-3-5-sonnet-20241022" - mock_llm.temperature = 0.7 - mock_llm.max_tokens = 2048 - - # Make ChatAnthropic constructor return the mock LLM - mock_chat_anthropic.return_value = mock_llm - - # Mock CrewAgent to avoid initialization errors - mock_crew_agent = Mock() - mock_crew_agent_class.return_value = mock_crew_agent - - # Mock Crew execution - with patch("app.services.crew_service.Crew") as mock_crew_class: - with patch("app.services.crew_service.CrewTask") as mock_task_class: - mock_crew = Mock() - mock_crew.kickoff.return_value = "Task completed successfully" - mock_crew_class.return_value = mock_crew - - agent = Agent( - id="test-123", - name="Test Agent", - description="Test agent for testing", - agent_type=AgentType.CONVERSATIONAL, - model="claude-3-5-sonnet-20241022", - system_prompt="You are a helpful test agent.", - temperature=0.7, - max_tokens=2048, - status=AgentStatus.ACTIVE - ) - - result = await CrewService.execute_agent_task( - db=Mock(), - agent=agent, - task_description="Complete a test task" - ) - - # Verify success - assert result["success"] is True - assert "result" in result - assert result["agent_id"] == "test-123" - - @pytest.mark.skipif(not CREWAI_AVAILABLE, reason="CrewAI not available") - @pytest.mark.asyncio - @patch("app.services.crew_service.CrewAgent") - @patch("app.services.crew_service.ChatAnthropic") - @patch("app.services.crew_service.llm_service") - async def test_execute_multi_agent_task_sequential(self, mock_llm_service, mock_chat_anthropic, mock_crew_agent_class): - """Test executing multi-agent task in sequential mode.""" - # Configure mocks - mock_llm_service.api_key = "test-api-key" - - # Mock LLM instance with all required attributes - mock_llm = Mock() - mock_llm.supports_stop_words = True - mock_llm.model_name = "claude-3-5-sonnet-20241022" - mock_llm.temperature = 0.7 - mock_llm.max_tokens = 2048 - - # Make ChatAnthropic constructor return the mock LLM - mock_chat_anthropic.return_value = mock_llm - - # Mock CrewAgent to avoid initialization errors - mock_crew_agent = Mock() - mock_crew_agent_class.return_value = mock_crew_agent - - # Mock Crew execution - with patch("app.services.crew_service.Crew") as mock_crew_class: - with patch("app.services.crew_service.CrewTask") as mock_task_class: - mock_crew = Mock() - mock_crew.kickoff.return_value = "All tasks completed" - mock_crew_class.return_value = mock_crew - - agents = [ - Agent( - id="agent-1", - name="Research Agent", - description="Research topics", - agent_type=AgentType.ANALYTICAL, - model="claude-3-5-sonnet-20241022", - system_prompt="You research topics.", - temperature=0.7, - max_tokens=2048, - status=AgentStatus.ACTIVE - ), - Agent( - id="agent-2", - name="Writing Agent", - description="Write content", - agent_type=AgentType.CREATIVE, - model="claude-3-5-sonnet-20241022", - system_prompt="You write content.", - temperature=0.9, - max_tokens=2048, - status=AgentStatus.ACTIVE - ) - ] - - tasks = [ - "Research AI trends", - "Write article about findings" - ] - - result = await CrewService.execute_multi_agent_task( - db=Mock(), - agents=agents, - task_descriptions=tasks, - process="sequential" - ) - - # Verify success - assert result["success"] is True - assert "result" in result - assert len(result["agents"]) == 2 - assert result["process"] == "sequential" diff --git a/backend/tests/unit/test_websocket_broadcasts.py b/backend/tests/unit/test_websocket_broadcasts.py index 19a786b..3f96971 100644 --- a/backend/tests/unit/test_websocket_broadcasts.py +++ b/backend/tests/unit/test_websocket_broadcasts.py @@ -33,9 +33,9 @@ async def test_task_started_broadcast(db_session, sample_agent): # Mock the broadcast_event function with patch("app.routers.websocket.broadcast_event", new_callable=AsyncMock) as mock_broadcast: - # Mock CrewService to prevent actual execution + # Mock AgentRuntime to prevent actual execution with patch( - "app.workers.tasks.CrewService.execute_agent_task", new_callable=AsyncMock + "app.workers.tasks.AgentRuntime.execute_agent_task", new_callable=AsyncMock ) as mock_crew: mock_crew.return_value = {"success": True, "output": "Test output"} @@ -81,9 +81,9 @@ async def test_task_completed_broadcast(db_session, sample_agent): # Mock the broadcast_event function with patch("app.routers.websocket.broadcast_event", new_callable=AsyncMock) as mock_broadcast: - # Mock CrewService to return success + # Mock AgentRuntime to return success with patch( - "app.workers.tasks.CrewService.execute_agent_task", new_callable=AsyncMock + "app.workers.tasks.AgentRuntime.execute_agent_task", new_callable=AsyncMock ) as mock_crew: mock_crew.return_value = {"success": True, "output": "Completed successfully"} @@ -128,9 +128,9 @@ async def test_task_failed_broadcast_on_error(db_session, sample_agent): # Mock the broadcast_event function with patch("app.routers.websocket.broadcast_event", new_callable=AsyncMock) as mock_broadcast: - # Mock CrewService to return failure + # Mock AgentRuntime to return failure with patch( - "app.workers.tasks.CrewService.execute_agent_task", new_callable=AsyncMock + "app.workers.tasks.AgentRuntime.execute_agent_task", new_callable=AsyncMock ) as mock_crew: mock_crew.return_value = {"success": False, "error": "Execution failed"} @@ -174,9 +174,9 @@ async def test_task_failed_broadcast_on_exception(db_session, sample_agent): # Mock the broadcast_event function with patch("app.routers.websocket.broadcast_event", new_callable=AsyncMock) as mock_broadcast: - # Mock CrewService to raise exception + # Mock AgentRuntime to raise exception with patch( - "app.workers.tasks.CrewService.execute_agent_task", new_callable=AsyncMock + "app.workers.tasks.AgentRuntime.execute_agent_task", new_callable=AsyncMock ) as mock_crew: mock_crew.side_effect = Exception("Unexpected error")