The Autonomous Multi-Agent AI System is a hierarchical, supervisor-led orchestration platform built on LangGraph. The system enables complex task execution through specialized agents that communicate via a centralized shared state. The architecture supports dynamic task decomposition, human-in-the-loop validation, and extensible agent/tool integration.
- Centralized State Management: All agent communication occurs through a shared state object, eliminating direct agent-to-agent coupling
- Supervisor-Led Orchestration: A single Supervisor Agent controls task decomposition and routing decisions
- Specialization: Each worker agent focuses on a specific capability domain
- Human Oversight: Critical checkpoints require explicit user approval before proceeding
- Extensibility: New agents and tools can be added without modifying core orchestration logic
- Orchestration: LangGraph for state management and agent coordination
- Backend: FastAPI for REST API layer
- Frontend: Streamlit or React for user interface
- LLM Integration: OpenAI/Anthropic APIs for agent reasoning
- Storage: PostgreSQL for persistent data, Vector DB (Pinecone/Weaviate) for RAG
- Deployment: Docker containers, Kubernetes for orchestration
- Authentication: JWT tokens with role-based access control
graph TD
User[User/Frontend] --> API[FastAPI Backend]
API --> Auth[Authentication Layer]
Auth --> LG[LangGraph Orchestration]
LG --> SS[Shared State]
LG --> Supervisor[Supervisor Agent]
Supervisor --> SS
Supervisor --> Research[Research Agent]
Supervisor --> DataTool[Data/Tool Agent]
Supervisor --> Analysis[Analysis Agent]
Supervisor --> Writer[Writer Agent]
Supervisor --> Reviewer[Reviewer Agent]
Research --> SS
DataTool --> SS
Analysis --> SS
Writer --> SS
Reviewer --> SS
Research --> VectorDB[(Vector Store)]
DataTool --> Tools[Tool Registry]
DataTool --> DB[(Database)]
SS --> AuditLog[(Audit Log)]
Reviewer --> ApprovalGate{Human Approval Gate}
ApprovalGate --> User
stateDiagram-v2
[*] --> TaskSubmission
TaskSubmission --> Decomposition: User submits task
Decomposition --> Routing: Supervisor creates subtasks
Routing --> WorkerExecution: Assign to worker
WorkerExecution --> Review: Worker completes
Review --> QualityCheck: Reviewer validates
QualityCheck --> ApprovalGate: Quality OK
QualityCheck --> WorkerExecution: Quality issues (rework)
ApprovalGate --> UserDecision: Present to user
UserDecision --> NextSubtask: Approved
UserDecision --> WorkerExecution: Changes requested
NextSubtask --> Routing: More subtasks
NextSubtask --> Completion: All subtasks done
Completion --> [*]
The system follows a layered architecture pattern:
- Presentation Layer: Web UI (Streamlit/React) for user interaction
- API Layer: FastAPI endpoints for task submission, status retrieval, and approval actions
- Security Layer: JWT authentication and RBAC authorization
- Orchestration Layer: LangGraph state machine managing agent coordination
- Agent Layer: Specialized worker agents and supervisor
- Tool Layer: Pluggable tools for data processing, code execution, and external integrations
- Data Layer: PostgreSQL for persistent storage, Vector DB for semantic search
- Infrastructure Layer: Docker containers, Kubernetes, cloud services
The Shared State is the central communication mechanism for all agents. It is implemented as a TypedDict in Python.
from typing import TypedDict, List, Optional, Literal
class SharedState(TypedDict):
# Core task information
user_task: str
task_id: str
user_id: str
# Execution control
current_subtask: Optional[str]
subtask_queue: List[str]
next_agent: Optional[str]
task_status: Literal["pending", "in_progress", "completed", "error"]
# Agent outputs
worker_output: Optional[dict]
reviewer_feedback: Optional[str]
# User interaction
user_decision: Optional[Literal["approved", "rejected", "changes_requested"]]
user_feedback: Optional[str]
# History and audit
execution_history: List[dict]
error_log: List[dict]
# Metadata
created_at: str
updated_at: strThe Supervisor Agent is responsible for task decomposition and routing decisions.
Interface:
class SupervisorAgent:
def decompose_task(self, state: SharedState) -> List[str]:
"""
Decomposes the user task into subtasks.
Args:
state: Current shared state containing user_task
Returns:
List of subtask descriptions
"""
pass
def route_next(self, state: SharedState) -> str:
"""
Determines which agent should handle the next subtask.
Args:
state: Current shared state
Returns:
Agent identifier (e.g., "research", "data_tool", "analysis", "writer", "reviewer")
"""
pass
def should_terminate(self, state: SharedState) -> bool:
"""
Determines if the workflow should terminate.
Args:
state: Current shared state
Returns:
True if workflow is complete, False otherwise
"""
passImplementation Notes:
- Uses LLM to analyze task complexity and determine subtask breakdown
- Maintains a routing table mapping subtask types to appropriate agents
- Considers reviewer feedback and user decisions when routing
- Terminates when subtask_queue is empty and all outputs are approved
All worker agents share a common interface but implement specialized logic.
Base Interface:
from abc import ABC, abstractmethod
class WorkerAgent(ABC):
@abstractmethod
def execute(self, state: SharedState) -> dict:
"""
Executes the current subtask and returns results.
Args:
state: Current shared state containing current_subtask
Returns:
Dictionary containing execution results
"""
pass
@abstractmethod
def get_capabilities(self) -> List[str]:
"""
Returns list of capability tags for routing.
Returns:
List of capability identifiers
"""
passCapabilities: ["research", "information_retrieval", "knowledge_search"]
Implementation:
- Performs semantic search against vector store when available
- Falls back to web search or document retrieval
- Structures retrieved information with source citations
- Returns formatted research results in worker_output
class ResearchAgent(WorkerAgent):
def __init__(self, vector_store: Optional[VectorStore] = None):
self.vector_store = vector_store
def execute(self, state: SharedState) -> dict:
query = state["current_subtask"]
if self.vector_store:
results = self.vector_store.similarity_search(query, k=5)
else:
results = self._web_search(query)
return {
"research_results": results,
"sources": [r.metadata for r in results],
"summary": self._summarize(results)
}Capabilities: ["code_execution", "database_query", "data_processing", "api_call"]
Implementation:
- Executes Python code in sandboxed environment
- Queries databases using SQLAlchemy
- Processes CSV/JSON data
- Calls external APIs
- Handles tool registration and execution
class DataToolAgent(WorkerAgent):
def __init__(self, tool_registry: ToolRegistry):
self.tool_registry = tool_registry
self.sandbox = CodeSandbox()
def execute(self, state: SharedState) -> dict:
subtask = state["current_subtask"]
tool_name = self._identify_tool(subtask)
if tool_name == "python":
return self._execute_code(subtask)
elif tool_name in self.tool_registry:
tool = self.tool_registry.get(tool_name)
return tool.execute(subtask)
else:
raise ValueError(f"Unknown tool: {tool_name}")
def _execute_code(self, code: str) -> dict:
try:
result = self.sandbox.run(code, timeout=30)
return {"success": True, "output": result}
except Exception as e:
return {"success": False, "error": str(e)}Capabilities: ["analysis", "reasoning", "insight_extraction", "data_interpretation"]
Implementation:
- Performs structured reasoning on provided data
- Extracts insights and patterns
- Generates structured analysis outputs
- Maintains data provenance
class AnalysisAgent(WorkerAgent):
def execute(self, state: SharedState) -> dict:
data = state["worker_output"]
subtask = state["current_subtask"]
analysis_prompt = self._build_analysis_prompt(subtask, data)
insights = self.llm.generate(analysis_prompt)
return {
"insights": insights,
"data_sources": self._extract_sources(data),
"confidence": self._calculate_confidence(insights),
"structured_output": self._structure_insights(insights)
}Capabilities: ["content_generation", "report_writing", "summarization", "formatting"]
Implementation:
- Generates business-ready content
- Formats output according to specified template
- Ensures coherence and completeness
- Integrates information from previous agents
class WriterAgent(WorkerAgent):
def execute(self, state: SharedState) -> dict:
context = self._gather_context(state)
output_format = self._determine_format(state["current_subtask"])
content = self.llm.generate(
prompt=self._build_writing_prompt(context, output_format),
temperature=0.7
)
return {
"content": content,
"format": output_format,
"word_count": len(content.split()),
"sections": self._extract_sections(content)
}Capabilities: ["quality_review", "validation", "consistency_check"]
Implementation:
- Evaluates output quality against criteria
- Checks logical consistency
- Validates completeness
- Triggers human approval gate
- Provides specific feedback for improvements
class ReviewerAgent(WorkerAgent):
def execute(self, state: SharedState) -> dict:
output = state["worker_output"]
subtask = state["current_subtask"]
quality_score = self._evaluate_quality(output, subtask)
consistency_check = self._check_consistency(output, state["execution_history"])
completeness = self._check_completeness(output, subtask)
if quality_score >= 0.8 and consistency_check and completeness:
return {
"approved": True,
"quality_score": quality_score,
"trigger_human_approval": True
}
else:
return {
"approved": False,
"quality_score": quality_score,
"feedback": self._generate_feedback(output, subtask),
"trigger_human_approval": False
}The Tool Registry enables dynamic tool registration and execution.
class Tool(ABC):
@abstractmethod
def execute(self, input_data: dict) -> dict:
"""Execute the tool with given input."""
pass
@abstractmethod
def get_schema(self) -> dict:
"""Return JSON schema for tool input."""
pass
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, Tool] = {}
def register(self, name: str, tool: Tool):
"""Register a new tool."""
self._tools[name] = tool
def get(self, name: str) -> Tool:
"""Retrieve a registered tool."""
return self._tools.get(name)
def list_tools(self) -> List[str]:
"""List all registered tool names."""
return list(self._tools.keys())The LangGraph state machine orchestrates the execution flow.
from langgraph.graph import StateGraph, END
def create_workflow() -> StateGraph:
workflow = StateGraph(SharedState)
# Add nodes
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("research", research_node)
workflow.add_node("data_tool", data_tool_node)
workflow.add_node("analysis", analysis_node)
workflow.add_node("writer", writer_node)
workflow.add_node("reviewer", reviewer_node)
workflow.add_node("human_approval", human_approval_node)
# Add conditional edges
workflow.add_conditional_edges(
"supervisor",
route_to_agent,
{
"research": "research",
"data_tool": "data_tool",
"analysis": "analysis",
"writer": "writer",
"reviewer": "reviewer",
"end": END
}
)
# Worker agents route to reviewer
for agent in ["research", "data_tool", "analysis", "writer"]:
workflow.add_edge(agent, "reviewer")
# Reviewer routes to human approval or back to supervisor
workflow.add_conditional_edges(
"reviewer",
route_after_review,
{
"human_approval": "human_approval",
"supervisor": "supervisor"
}
)
# Human approval routes back to supervisor
workflow.add_edge("human_approval", "supervisor")
# Set entry point
workflow.set_entry_point("supervisor")
return workflow.compile()The FastAPI backend exposes REST endpoints for system interaction.
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
app = FastAPI()
security = HTTPBearer()
@app.post("/tasks")
async def submit_task(
task: TaskSubmission,
credentials: HTTPAuthorizationCredentials = Depends(security)
):
"""Submit a new task for execution."""
user = authenticate_token(credentials.credentials)
task_id = generate_task_id()
initial_state = {
"user_task": task.description,
"task_id": task_id,
"user_id": user.id,
"task_status": "pending",
"subtask_queue": [],
"execution_history": [],
"error_log": [],
"created_at": datetime.utcnow().isoformat()
}
# Start async execution
asyncio.create_task(execute_workflow(initial_state))
return {"task_id": task_id, "status": "submitted"}
@app.get("/tasks/{task_id}")
async def get_task_status(
task_id: str,
credentials: HTTPAuthorizationCredentials = Depends(security)
):
"""Retrieve task status and results."""
user = authenticate_token(credentials.credentials)
task = get_task_from_db(task_id)
if task.user_id != user.id and not user.is_admin:
raise HTTPException(status_code=403, detail="Access denied")
return task
@app.post("/tasks/{task_id}/approve")
async def approve_output(
task_id: str,
decision: ApprovalDecision,
credentials: HTTPAuthorizationCredentials = Depends(security)
):
"""Submit approval decision for human gate."""
user = authenticate_token(credentials.credentials)
update_state(task_id, {
"user_decision": decision.decision,
"user_feedback": decision.feedback
})
# Resume workflow
asyncio.create_task(resume_workflow(task_id))
return {"status": "decision_recorded"}from jose import JWTError, jwt
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def authenticate_token(token: str) -> User:
"""Validate JWT token and return user."""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id = payload.get("sub")
if user_id is None:
raise HTTPException(status_code=401, detail="Invalid token")
user = get_user_by_id(user_id)
return user
except JWTError:
raise HTTPException(status_code=401, detail="Invalid token")
def check_permission(user: User, action: str) -> bool:
"""Check if user has permission for action."""
role_permissions = {
"admin": ["*"],
"user": ["submit_task", "view_own_tasks", "approve_own_tasks"],
"viewer": ["view_own_tasks"]
}
user_perms = role_permissions.get(user.role, [])
return "*" in user_perms or action in user_permsclass AuditLogger:
def __init__(self, db_connection):
self.db = db_connection
def log_agent_decision(self, agent_name: str, state: SharedState, decision: dict):
"""Log agent routing or execution decision."""
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"task_id": state["task_id"],
"agent": agent_name,
"action": "decision",
"details": decision,
"state_snapshot": self._sanitize_state(state)
}
self.db.insert("audit_log", log_entry)
def log_user_action(self, user_id: str, task_id: str, action: str, details: dict):
"""Log user approval or rejection."""
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"task_id": task_id,
"user_id": user_id,
"action": action,
"details": details
}
self.db.insert("audit_log", log_entry)-- Users table
CREATE TABLE users (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
email VARCHAR(255) UNIQUE NOT NULL,
hashed_password VARCHAR(255) NOT NULL,
role VARCHAR(50) NOT NULL DEFAULT 'user',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Tasks table
CREATE TABLE tasks (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
user_id UUID REFERENCES users(id),
description TEXT NOT NULL,
status VARCHAR(50) NOT NULL,
state JSONB NOT NULL,
result JSONB,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
completed_at TIMESTAMP
);
-- Audit log table
CREATE TABLE audit_log (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
task_id UUID REFERENCES tasks(id),
user_id UUID REFERENCES users(id),
agent VARCHAR(100),
action VARCHAR(100) NOT NULL,
details JSONB,
state_snapshot JSONB,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Tool registry table
CREATE TABLE tools (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
name VARCHAR(100) UNIQUE NOT NULL,
description TEXT,
schema JSONB NOT NULL,
enabled BOOLEAN DEFAULT true,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Indexes
CREATE INDEX idx_tasks_user_id ON tasks(user_id);
CREATE INDEX idx_tasks_status ON tasks(status);
CREATE INDEX idx_audit_log_task_id ON audit_log(task_id);
CREATE INDEX idx_audit_log_timestamp ON audit_log(timestamp);from pydantic import BaseModel, Field
from typing import Optional, List, Literal
from datetime import datetime
from uuid import UUID
class User(BaseModel):
id: UUID
email: str
role: Literal["admin", "user", "viewer"]
created_at: datetime
class TaskSubmission(BaseModel):
description: str = Field(..., min_length=10, max_length=5000)
metadata: Optional[dict] = None
class Task(BaseModel):
id: UUID
user_id: UUID
description: str
status: Literal["pending", "in_progress", "completed", "error"]
state: dict
result: Optional[dict] = None
created_at: datetime
updated_at: datetime
completed_at: Optional[datetime] = None
class ApprovalDecision(BaseModel):
decision: Literal["approved", "rejected", "changes_requested"]
feedback: Optional[str] = None
class AuditLogEntry(BaseModel):
id: UUID
task_id: UUID
user_id: Optional[UUID] = None
agent: Optional[str] = None
action: str
details: dict
timestamp: datetime
class ToolDefinition(BaseModel):
name: str
description: str
schema: dict
enabled: bool = TrueA property is a characteristic or behavior that should hold true across all valid executions of a system—essentially, a formal statement about what the system should do. Properties serve as the bridge between human-readable specifications and machine-verifiable correctness guarantees.
Property 1: Task Decomposition Completeness
For any user task submitted to the system, the Supervisor Agent should decompose it into one or more subtasks (non-empty list).
Validates: Requirements 1.1
Property 2: Subtask Routing Correctness
For any subtask with a known type (research, data processing, analysis, writing), the Supervisor Agent should route it to the appropriate worker agent based on agent capabilities.
Validates: Requirements 1.2, 10.3
Property 3: State Update After Assignment
For any subtask assignment, the Shared State should be updated to contain the current_subtask and next_agent fields with the correct values.
Validates: Requirements 1.3
Property 4: Workflow Termination
For any workflow where the subtask queue is empty and all outputs are approved, the Supervisor Agent should set task_status to "completed" and terminate the execution loop.
Validates: Requirements 1.4, 9.3
Property 5: Worker Agent Output Generation
For any worker agent (Research, Data/Tool, Analysis, Writer) and any valid subtask, the agent should produce a non-empty output dictionary.
Validates: Requirements 2.1, 4.1, 5.1
Property 6: Worker Agent State Updates
For any worker agent execution, the agent should write its results to the Shared State worker_output field.
Validates: Requirements 2.3, 3.5, 4.3, 5.3
Property 7: Research Agent Output Structure
For any research subtask, the Research Agent output should contain research_results, sources, and summary fields.
Validates: Requirements 2.4
Property 8: Code Execution Capability
For any valid Python code string, the Data/Tool Agent should execute it and return a result dictionary with success status and output or error.
Validates: Requirements 3.1
Property 9: Data Processing Capability
For any valid CSV or JSON data, the Data/Tool Agent should process it and return extracted information.
Validates: Requirements 3.3
Property 10: Analysis Output Structure
For any analysis subtask, the Analysis Agent output should contain insights, data_sources, confidence, and structured_output fields.
Validates: Requirements 4.2, 4.4
Property 11: Writer Output Format Compliance
For any specified output format (report, summary, document) and context, the Writer Agent should generate content that matches the requested format structure.
Validates: Requirements 5.2
Property 12: Reviewer Evaluation Generation
For any worker output, the Reviewer Agent should produce an evaluation containing approved status, quality_score, and either trigger_human_approval or feedback fields.
Validates: Requirements 6.1, 6.4
Property 13: Reviewer Feedback on Quality Issues
For any worker output with quality_score below threshold (< 0.8), the Reviewer Agent should include specific feedback in the reviewer_feedback field.
Validates: Requirements 6.2
Property 14: Approval Gate Trigger
For any worker output with quality_score >= 0.8 and passing consistency checks, the Reviewer Agent should set trigger_human_approval to True.
Validates: Requirements 6.3
Property 15: User Decision State Update
For any user approval or rejection action, the system should update the Shared State user_decision field with the decision value and preserve it in execution_history.
Validates: Requirements 7.2, 7.5
Property 16: Change Request Routing
For any user decision of "changes_requested", the system should update the state with user feedback and route back to the appropriate worker agent.
Validates: Requirements 7.3
Property 17: State Initialization
For any task submission, the system should create a Shared State with all required fields: user_task, task_id, user_id, task_status, subtask_queue, execution_history, error_log, created_at.
Validates: Requirements 8.1, 9.1
Property 18: Loop Continuation
For any workflow state where task_status is "in_progress" or "pending", the Supervisor Agent should continue routing to agents until task_status changes to "completed" or "error".
Validates: Requirements 9.2
Property 19: Execution Order Preservation
For any completed workflow, the execution_history should show the sequence: decompose → route → execute → review → approve → decide, repeated for each subtask.
Validates: Requirements 9.4
Property 20: Error Handling and Status Update
For any error occurring during agent execution, the system should capture the error in error_log, update task_status to "error", and halt the execution loop.
Validates: Requirements 3.4, 9.5, 16.1
Property 21: Error Classification
For any error, the Supervisor Agent should classify it as either recoverable (retry) or non-recoverable (terminate) and take the appropriate action.
Validates: Requirements 16.2, 16.4
Property 22: Error Message Descriptiveness
For any non-recoverable error, the system should return an error response containing a descriptive message and sufficient context for debugging.
Validates: Requirements 16.3, 16.5
Property 23: Tool Registration and Execution
For any tool implementing the Tool interface, when registered with the Data/Tool Agent, the tool should be available for execution and errors should be handled with standardized results.
Validates: Requirements 11.1, 11.3
Property 24: JWT Token Validation
For any API request, the system should validate the JWT token and either process the request (valid token) or return 401 Unauthorized (invalid token).
Validates: Requirements 12.2, 12.3
Property 25: Role-Based Permission Enforcement
For any user action, the system should verify the user's role permits the action and either allow it (permitted) or return 403 Forbidden (not permitted).
Validates: Requirements 13.2, 13.3
Property 26: JSON Response Format
For any successful API request, the system should return a response in valid JSON format.
Validates: Requirements 12.5
Property 27: Comprehensive Action Logging
For any agent decision or user action, the system should create an audit log entry containing timestamp, task_id, agent/user identifier, action type, and relevant details.
Validates: Requirements 14.1, 14.2, 14.4
Property 28: Audit Log Persistence
For any audit log entry created, the system should persist it to the database immediately.
Validates: Requirements 14.3
The system distinguishes between several error categories:
-
Validation Errors: Invalid input data, malformed requests
- Response: 400 Bad Request with validation details
- Recovery: User must correct input
-
Authentication Errors: Invalid or expired JWT tokens
- Response: 401 Unauthorized
- Recovery: User must re-authenticate
-
Authorization Errors: Insufficient permissions
- Response: 403 Forbidden
- Recovery: User must request elevated permissions
-
Agent Execution Errors: Failures during agent processing
- Recoverable: Retry with exponential backoff (max 3 attempts)
- Non-recoverable: Terminate workflow with error status
- Examples of recoverable: Temporary API failures, rate limits
- Examples of non-recoverable: Invalid code syntax, missing required data
-
System Errors: Database failures, service unavailability
- Response: 500 Internal Server Error
- Recovery: Automatic retry for transient failures, alert for persistent failures
def handle_agent_error(error: Exception, state: SharedState, attempt: int) -> SharedState:
"""Handle errors during agent execution."""
error_entry = {
"timestamp": datetime.utcnow().isoformat(),
"agent": state["next_agent"],
"error_type": type(error).__name__,
"error_message": str(error),
"attempt": attempt,
"state_snapshot": sanitize_state(state)
}
state["error_log"].append(error_entry)
if is_recoverable(error) and attempt < MAX_RETRIES:
# Retry with exponential backoff
time.sleep(2 ** attempt)
return state
else:
# Non-recoverable or max retries exceeded
state["task_status"] = "error"
state["error_message"] = format_user_error(error)
return state
def is_recoverable(error: Exception) -> bool:
"""Determine if an error is recoverable."""
recoverable_types = [
TimeoutError,
ConnectionError,
RateLimitError,
TemporaryAPIError
]
return any(isinstance(error, t) for t in recoverable_types)All errors logged to the audit system include:
- Stack trace (for debugging)
- State snapshot at time of error
- Agent identifier
- Subtask being processed
- Attempt number
- Timestamp
The system requires both unit testing and property-based testing for comprehensive coverage:
- Unit tests: Verify specific examples, edge cases, and error conditions
- Property tests: Verify universal properties across all inputs
- Together they provide comprehensive coverage: unit tests catch concrete bugs, property tests verify general correctness
We will use Hypothesis (Python) for property-based testing. Each correctness property defined above will be implemented as a property-based test.
Configuration:
- Minimum 100 iterations per property test
- Each test tagged with:
# Feature: autonomous-multi-agent-system, Property N: [property text] - Tests organized by component (supervisor, workers, reviewer, API, etc.)
Example Property Test:
from hypothesis import given, strategies as st
import pytest
# Feature: autonomous-multi-agent-system, Property 1: Task Decomposition Completeness
@given(user_task=st.text(min_size=10, max_size=500))
@pytest.mark.property_test
def test_task_decomposition_completeness(user_task):
"""For any user task, supervisor should decompose into non-empty subtask list."""
supervisor = SupervisorAgent()
state = {"user_task": user_task, "task_id": "test-123"}
subtasks = supervisor.decompose_task(state)
assert isinstance(subtasks, list)
assert len(subtasks) > 0
assert all(isinstance(s, str) for s in subtasks)
# Feature: autonomous-multi-agent-system, Property 6: Worker Agent State Updates
@given(
agent_type=st.sampled_from(["research", "data_tool", "analysis", "writer"]),
subtask=st.text(min_size=10, max_size=200)
)
@pytest.mark.property_test
def test_worker_agent_state_updates(agent_type, subtask):
"""For any worker agent execution, results should be written to worker_output."""
agent = create_agent(agent_type)
state = {
"current_subtask": subtask,
"task_id": "test-123",
"worker_output": None
}
result = agent.execute(state)
assert result is not None
assert isinstance(result, dict)
# In actual implementation, agent would update state directly
state["worker_output"] = result
assert state["worker_output"] is not NoneUnit tests focus on:
- Specific examples of task decomposition
- Edge cases (empty inputs, very long inputs, special characters)
- Error conditions (invalid tokens, missing permissions, database failures)
- Integration between components (API → LangGraph → Agents)
- Mock external dependencies (LLM APIs, databases, vector stores)
Example Unit Tests:
def test_supervisor_routes_research_task():
"""Test that research-type subtasks route to research agent."""
supervisor = SupervisorAgent()
state = {
"current_subtask": "Find information about machine learning",
"subtask_queue": []
}
next_agent = supervisor.route_next(state)
assert next_agent == "research"
def test_invalid_jwt_returns_401():
"""Test that invalid JWT token returns 401 Unauthorized."""
client = TestClient(app)
response = client.post(
"/tasks",
json={"description": "Test task"},
headers={"Authorization": "Bearer invalid_token"}
)
assert response.status_code == 401
def test_reviewer_provides_feedback_on_low_quality():
"""Test that reviewer provides feedback when quality is low."""
reviewer = ReviewerAgent()
state = {
"worker_output": {"content": "x"}, # Very low quality
"current_subtask": "Write a comprehensive report"
}
result = reviewer.execute(state)
assert result["approved"] is False
assert "feedback" in result
assert len(result["feedback"]) > 0Integration tests verify end-to-end workflows:
- Complete task execution from submission to completion
- Human approval gate interaction
- Error recovery and retry logic
- Multi-agent collaboration
- Database persistence and retrieval
tests/
├── unit/
│ ├── test_supervisor.py
│ ├── test_research_agent.py
│ ├── test_data_tool_agent.py
│ ├── test_analysis_agent.py
│ ├── test_writer_agent.py
│ ├── test_reviewer_agent.py
│ ├── test_api.py
│ ├── test_auth.py
│ └── test_audit.py
├── property/
│ ├── test_orchestration_properties.py
│ ├── test_worker_properties.py
│ ├── test_review_properties.py
│ ├── test_api_properties.py
│ └── test_error_properties.py
├── integration/
│ ├── test_end_to_end_workflow.py
│ ├── test_approval_gate.py
│ ├── test_error_recovery.py
│ └── test_multi_agent_collaboration.py
└── conftest.py # Shared fixtures
For testing, we mock:
- LLM API calls (OpenAI/Anthropic) with deterministic responses
- Vector store queries with fixed result sets
- Database operations with in-memory SQLite
- External API calls with mock responses
- Time-dependent operations for deterministic testing
All tests run on every commit:
- Unit tests: Fast feedback (< 1 minute)
- Property tests: Comprehensive coverage (< 5 minutes)
- Integration tests: End-to-end validation (< 10 minutes)
CI pipeline fails if:
- Any test fails
- Code coverage drops below 80%
- Property tests find counterexamples
- Linting or type checking fails