# Create Chat Completion
Source: https://arkos.mintlify.app/api-reference/endpoint/create
Send messages to an ARKOS agent
## POST /v1/chat/completions
Create a chat completion with an ARKOS agent using the OpenAI-compatible API format.
### Request Body
```json theme={null}
{
"model": "arkos-default",
"messages": [
{
"role": "system",
"content": "You are a helpful assistant with persistent memory."
},
{
"role": "user",
"content": "Hello! Can you help me understand ARKOS?"
}
],
"temperature": 0.7,
"max_tokens": 1000
}
```
### Response
```json theme={null}
{
"id": "chatcmpl-123",
"object": "chat.completion",
"created": 1677652288,
"model": "arkos-default",
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": "Hello! I'd be happy to help you understand ARKOS..."
},
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": 42,
"completion_tokens": 128,
"total_tokens": 170
}
}
```
### Parameters
| Parameter | Type | Required | Description |
| ----------- | ------- | -------- | ----------------------------------------------------- |
| model | string | Yes | Model identifier (currently supports "arkos-default") |
| messages | array | Yes | Array of message objects with role and content |
| temperature | float | No | Sampling temperature (0.0 to 2.0) |
| max\_tokens | integer | No | Maximum tokens to generate |
### Features In Development
* **Coming soon!** Streaming responses
* **Coming soon!** Tool/function calling
* **Coming soon!** Memory context injection
* **Coming soon!** Multi-agent conversations
# Delete Agent
Source: https://arkos.mintlify.app/api-reference/endpoint/delete
Remove an ARKOS agent and its associated memory
## DELETE /v1/agents/
**Coming soon!** This endpoint will delete an ARKOS agent and optionally its associated memory data.
### Planned Features
* Safe agent termination
* Memory export before deletion
* Cascade deletion of related resources
* Confirmation requirements for data loss prevention
### Example Request
```bash theme={null}
curl -X DELETE "http://localhost:8000/v1/agents/test-agent" \
-H "Authorization: Bearer YOUR_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"preserve_memory": false,
"export_data": true
}'
```
### Example Response (Planned)
```json theme={null}
{
"id": "test-agent",
"status": "deleted",
"memory_exported": true,
"export_location": "/exports/test-agent_2024-01-15.json",
"deleted_at": "2024-01-15T14:35:00Z"
}
```
### Parameters
| Parameter | Type | Required | Description |
| ---------------- | ------- | -------- | ---------------------------------------------------- |
| agent\_id | string | Yes | The unique identifier of the agent to delete |
| preserve\_memory | boolean | No | Whether to preserve memory data after agent deletion |
| export\_data | boolean | No | Export agent data before deletion |
This endpoint is not yet implemented. Agent management features are planned for a future release.
# Get Agent Status
Source: https://arkos.mintlify.app/api-reference/endpoint/get
Retrieve the current status of an ARKOS agent
## GET /v1/agents/
**Coming soon!** This endpoint will retrieve the current status and configuration of a specific ARKOS agent.
### Planned Features
* Agent state information
* Memory statistics
* Active tool configurations
* Conversation history
* Performance metrics
### Example Request
```bash theme={null}
curl -X GET "http://localhost:8000/v1/agents/default" \
-H "Authorization: Bearer YOUR_API_KEY"
```
### Example Response (Planned)
```json theme={null}
{
"id": "default",
"name": "Default ARKOS Agent",
"status": "active",
"model": {
"provider": "openai",
"name": "gpt-4"
},
"memory": {
"total_entries": 1523,
"working_memory": 12,
"long_term_memory": 1511
},
"state": {
"current": "waiting_for_input",
"transitions": 45
},
"tools": [
"weather",
"calendar"
],
"created_at": "2024-01-01T00:00:00Z",
"last_active": "2024-01-15T14:30:00Z"
}
```
This endpoint is not yet implemented. The structure above represents the planned API design.
# Webhook Events
Source: https://arkos.mintlify.app/api-reference/endpoint/webhook
Receive real-time notifications about agent events
## POST /v1/webhooks
**Coming soon!** Configure webhooks to receive real-time notifications about agent events and state changes.
### Planned Event Types
* **agent.created** - New agent instance created
* **agent.state\_changed** - Agent state transition occurred
* **memory.consolidated** - Memory consolidation completed
* **tool.executed** - Tool function was called
* **error.occurred** - Error or exception in agent processing
### Example Webhook Configuration
```json theme={null}
{
"url": "https://your-app.com/webhooks/arkos",
"events": ["agent.state_changed", "tool.executed"],
"secret": "webhook_secret_key",
"active": true
}
```
### Example Webhook Payload (Planned)
```json theme={null}
{
"event": "agent.state_changed",
"timestamp": "2024-01-15T14:30:00Z",
"agent_id": "default",
"data": {
"previous_state": "waiting_for_input",
"current_state": "processing",
"transition_reason": "user_input_received"
},
"signature": "sha256=..."
}
```
### Security
Webhooks will include HMAC-SHA256 signatures for payload verification:
```python theme={null}
import hmac
import hashlib
def verify_webhook(payload, signature, secret):
expected = hmac.new(
secret.encode(),
payload.encode(),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(f"sha256={expected}", signature)
```
Webhook functionality is not yet implemented. This feature is planned for a future release to enable event-driven integrations.
# API Reference
Source: https://arkos.mintlify.app/api-reference/introduction
ARKOS API endpoints for agent interaction and management
## Overview
ARKOS provides an OpenAI-compatible API interface for interacting with intelligent agents. The API is built with FastAPI and provides endpoints for chat completions, agent management, and memory operations.
The full API implementation is currently under development. The OpenAI compatibility layer is functional in `base_module/app.py`.
## Base URL
```
http://localhost:1111
```
## Authentication
**Coming soon!** Authentication and API key management will be implemented in a future release.
## Available Endpoints
### Chat Completions
The chat completions endpoint provides OpenAI-compatible chat interactions:
* **POST** `/v1/chat/completions` - Send messages to the agent
ARKOS implements the OpenAI chat completions API format for easy integration
## Planned Endpoints
The following endpoints are planned for future releases:
### Agent Management
* **Coming soon!** Create, list, update, and delete agents
* **Coming soon!** Configure agent behaviors and state machines
### Memory Operations
* **Coming soon!** Store and retrieve memories
* **Coming soon!** Search semantic memory
* **Coming soon!** Memory consolidation controls
### Tool Integration
* **Coming soon!** Register MCP tools
* **Coming soon!** Execute tool functions
* **Coming soon!** Tool discovery and management
### System Configuration
* **Coming soon!** Model selection and configuration
* **Coming soon!** System health and metrics
* **Coming soon!** Log access and debugging
# Architecture Overview
Source: https://arkos.mintlify.app/architecture/overview
Understanding ARKOS's modular architecture and design principles
# ARKOS Architecture Overview
ARKOS employs a sophisticated modular architecture designed for extensibility, scalability, and cognitive modeling. This document provides a comprehensive overview of the system design and component interactions.
## Core Design Principles
Each component is self-contained with well-defined interfaces, enabling independent development and testing
Memory and state systems inspired by cognitive science principles for human-like reasoning
Plugin-based architecture allows easy addition of new capabilities without core modifications
Graceful degradation with fallback mechanisms ensures system reliability
## System Architecture
```mermaid theme={null}
graph TB
subgraph "User Interface Layer"
UI["Web UI
(Coming soon!)"]
API["REST API
(Coming soon!)"]
CLI[CLI Interface]
end
subgraph "Core Processing Layer"
BASE[Base Module]
AGENT["Agent Module
(Coming soon!)"]
STATE["State Module
(Coming soon!)"]
end
subgraph "Intelligence Layer"
MODEL[Model Module]
MEMORY["Memory Module
(Basic CSV only)"]
TOOL["Tool Module
(Coming soon!)"]
end
subgraph "Storage Layer"
SQL["SQLite
(Coming soon!)"]
REDIS["Redis
(Coming soon!)"]
VECTOR["FAISS Vector DB
(Coming soon!)"]
end
subgraph "External Services"
LLM[LLM Providers]
MCP["MCP Tools
(Coming soon!)"]
CAL["Calendar Services
(Coming soon!)"]
end
CLI --> BASE
BASE --> MODEL
BASE --> MEMORY
MODEL --> LLM
```
## Module Overview
### Base Module
The foundation layer that orchestrates all other modules and provides the main interface.
**Key Responsibilities:**
* Request routing and orchestration
* Module initialization and lifecycle management
* Configuration loading and validation
* Error handling and recovery
### Agent Module
**Coming soon!** Will implement intelligent agent logic with context awareness and decision-making capabilities.
**Planned Features:**
* Multi-turn conversation management
* Context building and maintenance
* Decision tree navigation
* Response generation coordination
### State Module
**Coming soon!** Will manage conversation flow using a sophisticated state machine architecture.
**Planned Components:**
* **State Registry**: Dynamic state type registration
* **State Handler**: State transition management
* **State Graph**: Visual representation and analysis
* **Transition Rules**: Conditional state changes
### Model Module
Provides unified interface to multiple LLM providers with intelligent routing.
**Capabilities:**
* Multi-provider support (OpenAI, Anthropic, SGLANG)
* Automatic load balancing and failover
* Response caching and optimization
* Token counting and cost tracking
### Memory Module
Currently implements basic CSV-based memory storage. Advanced cognitive-inspired memory system **Coming soon!**
**Current Implementation:**
* Basic CSV file storage
* Simple memory persistence
**Planned Memory Types (Coming soon!):**
* **Working Memory**: Active context (7±2 items)
* **Short-term Memory**: Recent interactions
* **Long-term Memory**: Persistent knowledge base
* **Episodic Memory**: Event sequences
* **Semantic Memory**: Facts and concepts
### Tool Module
**Coming soon!** Will enable integration with external services and APIs through MCP protocol.
**Planned Tools:**
* Calendar management
* Weather information
* Web search
* Custom tool development
## Data Flow
### Request Processing Pipeline
```mermaid theme={null}
sequenceDiagram
participant User
participant Base
participant State
participant Agent
participant Model
participant Memory
participant Tool
User->>Base: Input Message
Base->>State: Check Current State
State->>Agent: Process Input
Agent->>Memory: Retrieve Context
Memory-->>Agent: Relevant Memories
Agent->>Model: Generate Response
Model->>Model: Select Provider
Model-->>Agent: LLM Response
Agent->>Tool: Execute Tools (if needed)
Tool-->>Agent: Tool Results
Agent->>Memory: Store Interaction
Agent-->>Base: Final Response
Base-->>User: Output Message
```
## Component Communication
### Inter-Module Messaging
Modules communicate through well-defined interfaces using typed messages:
```python theme={null}
@dataclass
class ModuleMessage:
sender: str
receiver: str
message_type: MessageType
payload: Any
timestamp: datetime
correlation_id: str
```
### Event System
ARKOS uses an event-driven architecture for loose coupling:
```python theme={null}
class EventBus:
def publish(event: Event) -> None
def subscribe(event_type: str, handler: Callable) -> None
def unsubscribe(subscription_id: str) -> None
```
## Scalability Considerations
**Coming soon!** Advanced scalability features are planned for future releases.
### Planned Horizontal Scaling Features:
* **Stateless Design**: Most modules will be stateless, enabling horizontal scaling
* **Load Balancing**: Built-in request distribution across multiple instances
* **Distributed Memory**: Redis backend will support distributed memory storage
### Planned Performance Optimization Features:
* **Caching**: Multi-level caching for responses and computations
* **Batch Processing**: Efficient handling of multiple requests
* **Async Operations**: Non-blocking I/O for improved throughput
* **Connection Pooling**: Reusable connections to external services
## Security Architecture
**Coming soon!** Comprehensive security features are planned for future releases.
### Planned Authentication & Authorization Features:
* **API Key Management**: Secure storage and rotation
* **Role-Based Access Control**: Fine-grained permissions
* **Session Management**: Secure session handling
### Planned Data Protection Features:
* **Encryption at Rest**: Sensitive data encrypted in storage
* **Encryption in Transit**: TLS for all external communications
* **Input Validation**: Comprehensive input sanitization
* **Rate Limiting**: Protection against abuse
## Deployment Architecture
**Coming soon!** Container-based deployment and orchestration features are planned.
### Planned Container-Based Deployment:
```yaml theme={null}
# Future docker-compose.yml (Coming soon!)
services:
arkos-core:
image: arkos:latest
ports:
- "8080:8080"
environment:
- MEMORY_BACKEND=redis
- MODEL_PROVIDER=openai
redis:
image: redis:alpine
volumes:
- redis_data:/data
sglang:
image: sglang:latest
deploy:
resources:
reservations:
devices:
- capabilities: [gpu]
```
### Cloud Deployment Options
ECS/EKS with auto-scaling groups
GKE with Cloud Run integration
AKS with Azure Functions
Docker Compose or Kubernetes
## Monitoring & Observability
**Coming soon!** Comprehensive monitoring and observability features are planned.
### Planned Metrics Collection:
* **Prometheus**: System and application metrics
* **Grafana**: Visualization and dashboards
* **Custom Metrics**: Module-specific performance indicators
### Planned Logging Strategy:
```python theme={null}
# Future structured logging example (Coming soon!)
logger.info(
"model_response_generated",
extra={
"module": "model",
"provider": "openai",
"latency_ms": 1250,
"tokens": 500,
"cost_usd": 0.015
}
)
```
### Planned Distributed Tracing:
* **OpenTelemetry**: End-to-end request tracing
* **Correlation IDs**: Request tracking across modules
* **Performance Profiling**: Bottleneck identification
## Development Workflow
### Module Development
1. **Interface Definition**: Define module interface and contracts
2. **Implementation**: Develop module following SOLID principles
3. **Unit Testing**: Comprehensive test coverage
4. **Integration**: Test with other modules
5. **Documentation**: Update architecture docs
### Testing Strategy
Currently limited to basic model tests. Comprehensive testing **Coming soon!**
**Current Testing:**
* Basic model functionality tests
**Planned Testing (Coming soon!):**
* **Unit Tests**: Individual module testing
* **Integration Tests**: Module interaction testing
* **System Tests**: End-to-end scenarios
* **Performance Tests**: Load and stress testing
* **Chaos Engineering**: Failure scenario testing
## Future Architecture Considerations
### Planned Enhancements
These are architectural improvements planned for future releases:
1. **Microservices Migration**: Breaking modules into separate services
2. **GraphQL API**: More flexible API layer
3. **Event Sourcing**: Complete audit trail of all events
4. **CQRS Pattern**: Separate read/write models
5. **Service Mesh**: Istio/Linkerd for service communication
### Research Areas
* **Neuromorphic Computing**: Brain-inspired processing
* **Quantum Integration**: Quantum computing for optimization
* **Federated Learning**: Distributed model training
* **Edge Deployment**: Running on edge devices
## Best Practices
### Code Organization
```
module_name/
├── __init__.py # Module exports
├── core.py # Core logic
├── models.py # Data models
├── handlers.py # Request handlers
├── utils.py # Utility functions
├── config.py # Configuration
└── tests/ # Test files
```
### Design Patterns
* **Factory Pattern**: Dynamic object creation
* **Strategy Pattern**: Algorithm selection
* **Observer Pattern**: Event handling
* **Singleton Pattern**: Resource management
* **Chain of Responsibility**: Request processing
## Conclusion
ARKOS's architecture is designed to be:
* **Modular**: Easy to understand and extend
* **Scalable**: Grows with your needs
* **Reliable**: Fault-tolerant with fallbacks
* **Intelligent**: Cognitive-inspired design
This architecture enables building sophisticated AI agents that can learn, remember, and adapt to user needs while maintaining high performance and reliability.
# Development Setup
Source: https://arkos.mintlify.app/development/setup
Complete guide to setting up your ARKOS development environment
# Development Setup
This guide will help you set up a complete development environment for ARKOS, including all dependencies and configurations.
## System Requirements
### Minimum Requirements
* **OS**: Ubuntu 20.04+, macOS 11+, or Windows 10+ with WSL2
* **Python**: 3.8 or higher
* **RAM**: 8GB minimum
* **Storage**: 20GB free space
* **CPU**: 4 cores recommended
### Recommended Requirements
* **RAM**: 16GB or more
* **GPU**: NVIDIA GPU with 8GB+ VRAM (for local models)
* **Storage**: 50GB+ for model files
* **CPU**: 8+ cores for better performance
## Environment Setup
### 1. Python Environment
```bash theme={null}
# Create virtual environment
python3 -m venv arkos-env
# Activate environment
# On Linux/Mac:
source arkos-env/bin/activate
# On Windows:
arkos-env\Scripts\activate
# Upgrade pip
pip install --upgrade pip
```
```bash theme={null}
# Create conda environment
conda create -n arkos python=3.10
# Activate environment
conda activate arkos
# Install pip in conda env
conda install pip
```
### 2. Clone Repository
```bash theme={null}
# Clone the repository
git clone https://github.com/SGIARK/arkos.git
cd arkos
```
### 3. Install Dependencies
```bash theme={null}
pip install -r requirements.txt
```
## Dependencies Overview
### Current Dependencies (requirements.txt)
```txt theme={null}
openai>=1.61.0
pyyaml>=6.0.2
pydantic>=2.10.6
requests>=2.32.3
fastapi>=0.115.0
uvicorn>=0.32.0
psycopg2-binary>=2.9.11
vecs
mem0ai
sentence-transformers
python-dotenv>=1.0.0
aiohttp>=3.9.0
google-auth-oauthlib>=1.2.0
```
### Dependency Details
| Package | Purpose |
| ----------------------- | ---------------------------------------- |
| `openai` | AsyncOpenAI client for LLM communication |
| `pyyaml` | YAML configuration parsing |
| `pydantic` | Data validation and message schemas |
| `fastapi` + `uvicorn` | API server |
| `psycopg2-binary` | PostgreSQL database adapter |
| `mem0ai` | Vector memory with Supabase |
| `sentence-transformers` | Embeddings for memory |
| `python-dotenv` | Environment variable loading |
| `aiohttp` | Async HTTP client |
| `google-auth-oauthlib` | Google OAuth for calendar |
## Configuration Files
### 1. Environment Variables (.env)
Create a `.env` file in the project root:
```bash theme={null}
cp .env.example .env
```
Edit `.env` with your configuration:
```bash theme={null}
# Database Connection (REQUIRED)
# Format: postgresql://user:password@host:port/database
DB_URL=postgresql://postgres:your-password@localhost:54322/postgres
# Hugging Face Token (OPTIONAL - for gated models)
HF_TOKEN=
# MCP Server Credentials (OPTIONAL - for tool integrations)
GOOGLE_OAUTH_CREDENTIALS=
GOOGLE_CALENDAR_MCP_TOKEN_PATH=
BRAVE_API_KEY=
# OpenAI API Key (REQUIRED by mem0, but can be placeholder)
OPENAI_API_KEY=sk-placeholder
```
The `DB_URL` environment variable is **required**. ARKOS uses PostgreSQL for conversation context storage.
### 2. YAML Configuration
The main configuration file is `config_module/config.yaml`:
```yaml theme={null}
app:
host: "0.0.0.0"
port: 1111
reload: false
system_prompt: "You are a helpful AI assistant."
llm:
base_url: "http://localhost:30000/v1"
database:
url: "${DB_URL}"
memory:
user_id: "default_user"
use_long_term: false
state:
graph_path: "state_module/state_graph.yaml"
# MCP Server Configuration (optional)
mcp_servers:
google-calendar:
transport: stdio
command: npx
args: ["-y", "@anthropic/google-calendar-mcp"]
```
## Database Setup
### PostgreSQL (Required)
ARKOS requires PostgreSQL for conversation context storage.
#### Using Supabase (Recommended)
1. Create a [Supabase](https://supabase.com) project
2. Get your connection string from Settings > Database
3. Set `DB_URL` in your `.env` file
#### Using Local PostgreSQL
```bash theme={null}
# Install PostgreSQL
sudo apt-get install postgresql postgresql-contrib
# Create database
sudo -u postgres psql
CREATE DATABASE arkos;
CREATE USER arkos_user WITH PASSWORD 'your_password';
GRANT ALL PRIVILEGES ON DATABASE arkos TO arkos_user;
\q
```
### Database Schema
Create the required table:
```sql theme={null}
CREATE TABLE conversation_context (
id SERIAL PRIMARY KEY,
user_id VARCHAR(255) NOT NULL,
session_id VARCHAR(255) NOT NULL,
role VARCHAR(50) NOT NULL,
message TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_user_id ON conversation_context(user_id);
CREATE INDEX idx_session_id ON conversation_context(session_id);
```
## Running SGLANG
### Check if Already Running
On shared servers, the LLM server may already be running:
```bash theme={null}
# Check if port 30000 is in use
lsof -i :30000
# Or verify it's responding
curl http://localhost:30000/v1/models
```
### Start the LLM Server
```bash theme={null}
cd arkos
bash model_module/run.sh
```
This starts the SGLang server on **port 30000** with the Qwen 2.5-7B-Instruct model.
Only one instance can run on port 30000 at a time. Coordinate with your team on shared servers.
## Running the Application
### 1. Start the LLM Server
```bash theme={null}
# Terminal 1
bash model_module/run.sh
```
Wait for "server started" messages.
### 2. Start the API Server
```bash theme={null}
# Terminal 2
python base_module/app.py
```
This starts FastAPI on **port 1111**.
### 3. Start the CLI Interface (Optional)
```bash theme={null}
# Terminal 3
python base_module/main_interface.py
```
### Verify Setup
```bash theme={null}
# Health check
curl http://localhost:1111/health
# Test chat
curl -X POST http://localhost:1111/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{"model": "ark-agent", "messages": [{"role": "user", "content": "Hello!"}]}'
```
## IDE Configuration
### VS Code
Create `.vscode/settings.json`:
```json theme={null}
{
"python.defaultInterpreterPath": "${workspaceFolder}/arkos-env/bin/python",
"python.linting.enabled": true,
"python.linting.flake8Enabled": true,
"python.formatting.provider": "black",
"python.testing.pytestEnabled": true,
"python.testing.pytestArgs": ["tests"],
"editor.formatOnSave": true
}
```
### PyCharm
1. Set Python interpreter to virtual environment
2. Enable pytest as test runner
3. Configure code style with Black formatter
## Testing
### Running Tests
```bash theme={null}
# Run specific test files
python model_module/tests_arkmodel.py
python config_module/test_config_loader.py
python tool_module/test_tool_call.py
# Run with pytest (if available)
pytest tests/
```
### Test Configuration
```ini theme={null}
# pytest.ini (if using pytest)
[pytest]
testpaths = tests
python_files = test_*.py
asyncio_mode = auto
```
## Debugging
### Enable Debug Logging
```python theme={null}
import logging
logging.basicConfig(level=logging.DEBUG)
```
### VS Code Launch Configuration
Create `.vscode/launch.json`:
```json theme={null}
{
"version": "0.2.0",
"configurations": [
{
"name": "Debug API Server",
"type": "python",
"request": "launch",
"program": "${workspaceFolder}/base_module/app.py",
"console": "integratedTerminal"
},
{
"name": "Debug CLI",
"type": "python",
"request": "launch",
"program": "${workspaceFolder}/base_module/main_interface.py",
"console": "integratedTerminal"
}
]
}
```
## Common Issues
Add the project root to PYTHONPATH:
```bash theme={null}
export PYTHONPATH="${PYTHONPATH}:$(pwd)"
```
Verify your `DB_URL` in `.env`:
```bash theme={null}
psql $DB_URL -c "SELECT 1"
```
Check if SGLang is running:
```bash theme={null}
curl http://localhost:30000/v1/models
```
If not, start it with `bash model_module/run.sh`
Find and kill the process:
```bash theme={null}
lsof -i :1111
kill -9
```
Reduce batch size or use a smaller model:
```python theme={null}
os.environ['PYTORCH_CUDA_ALLOC_CONF'] = 'max_split_size_mb:512'
```
Ensure `.env` file exists and is in the project root:
```bash theme={null}
ls -la .env
cat .env
```
## Project Structure
```
arkos/
├── agent_module/ # Agent orchestration
│ ├── __init__.py
│ └── agent.py
├── base_module/ # API server and CLI
│ ├── __init__.py
│ ├── app.py # FastAPI server
│ ├── auth.py # OAuth routes
│ └── main_interface.py # CLI
├── config_module/ # Configuration
│ ├── __init__.py
│ ├── loader.py # Config loading
│ ├── config.yaml # Main config
│ └── state_graph.yaml # State machine
├── memory_module/ # Memory system
│ ├── __init__.py
│ └── memory.py
├── model_module/ # LLM interface
│ ├── __init__.py
│ ├── ArkModelNew.py
│ └── run.sh # SGLang startup
├── state_module/ # State machine
│ ├── __init__.py
│ ├── state.py
│ ├── state_handler.py
│ └── state_registry.py
├── tool_module/ # MCP tools
│ ├── __init__.py
│ ├── tool_call.py # MCP client
│ ├── token_store.py # OAuth tokens
│ └── transports/ # Transport layer
├── .env.example
├── .gitignore
├── README.md
└── requirements.txt
```
## Next Steps
Learn about testing strategies
Start using ARKOS
Understand the system design
Explore the APIs
# Testing Guide
Source: https://arkos.mintlify.app/development/testing
Comprehensive testing strategy for ARKOS with pytest
# Testing Guide
**Comprehensive testing is coming soon!** This guide outlines the planned testing strategy for ARKOS.
## Current Testing Status
ARKOS currently has only basic model functionality tests. A comprehensive testing suite with unit tests, integration tests, and system tests is planned for future releases.
## Planned Testing Philosophy (Coming soon!)
ARKOS will follow a test-driven development (TDD) approach with comprehensive coverage across all modules. We aim for >80% code coverage with meaningful tests.
## Current Test Structure
```
tests/
└── test_model_module.py # Basic model tests only
```
## Planned Test Structure (Coming soon!)
```
# Future comprehensive test structure (Coming soon!)
tests/
├── conftest.py # Shared fixtures and configuration
├── unit/ # Unit tests for individual components
│ ├── test_state/
│ ├── test_memory/
│ ├── test_model/
│ └── test_tool/
├── integration/ # Integration tests
│ ├── test_agent_flow.py
│ ├── test_memory_persistence.py
│ └── test_model_routing.py
├── system/ # End-to-end system tests
│ ├── test_complete_conversation.py
│ └── test_multi_agent.py
├── performance/ # Performance and load tests
│ ├── test_memory_performance.py
│ └── test_model_latency.py
└── fixtures/ # Test data and mocks
├── sample_conversations.json
└── mock_responses.yaml
```
## Setting Up Testing
### Current Testing Setup
**Basic testing only - comprehensive testing coming soon!**
```bash theme={null}
# Current basic testing (if available)
python -m pytest tests/test_model_module.py
```
### Planned Testing Setup (Coming soon!)
```bash theme={null}
# Future comprehensive testing setup (Coming soon!)
pip install pytest pytest-asyncio pytest-cov pytest-mock pytest-benchmark httpx
```
### Configuration
Create `pytest.ini` in project root:
```ini theme={null}
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
# Markers
markers =
unit: Unit tests
integration: Integration tests
system: System tests
slow: Slow tests
requires_gpu: Tests requiring GPU
requires_api_key: Tests requiring API keys
# Coverage
addopts =
--cov=.
--cov-report=html
--cov-report=term-missing
--verbose
# Async
asyncio_mode = auto
```
## Writing Tests (Coming soon!)
**The following test examples are planned for future implementation.**
### Planned Unit Tests
Future unit tests will test individual components in isolation:
```python theme={null}
# tests/unit/test_state/test_state_handler.py
import pytest
from state_module.state_handler import StateHandler
from state_module.state import StateContext
class TestStateHandler:
@pytest.fixture
def handler(self):
"""Create state handler for testing"""
return StateHandler(yaml_path="tests/fixtures/test_state_graph.yaml")
@pytest.fixture
def context(self):
"""Create test context"""
return StateContext(user_input="test input")
def test_initial_state(self, handler):
"""Test that handler starts with correct initial state"""
initial = handler.get_initial_state()
assert initial.name == "greeting"
assert not initial.is_terminal
def test_state_transition(self, handler, context):
"""Test state transitions work correctly"""
current_state = handler.get_initial_state()
next_state = handler.get_next_state(current_state.name, context)
assert next_state is not None
assert next_state.name == "waiting_for_input"
@pytest.mark.parametrize("input_text,expected_state", [
("hello", "greeting_response"),
("help", "help_state"),
("exit", "goodbye")
])
def test_conditional_transitions(self, handler, input_text, expected_state):
"""Test conditional state transitions"""
context = StateContext(user_input=input_text)
next_state = handler.get_next_state("router", context)
assert next_state.name == expected_state
```
### Planned Integration Tests (Coming soon!)
Future integration tests will test component interactions:
```python theme={null}
# tests/integration/test_agent_flow.py
import pytest
from unittest.mock import Mock, patch
from agent_module.agent import Agent
from model_module.ArkModelNew import ArkModel
from memory_module.memory import Memory
class TestAgentFlow:
@pytest.fixture
def agent(self):
"""Create agent with mocked dependencies"""
model = Mock(spec=ArkModel)
memory = Mock(spec=Memory)
return Agent(model=model, memory=memory)
@pytest.mark.asyncio
async def test_conversation_flow(self, agent):
"""Test complete conversation flow"""
# Mock model responses
agent.model.generate.return_value = Mock(
content="Hello! How can I help you?",
success=True
)
# Mock memory operations
agent.memory.search.return_value = []
# Process user message
response = await agent.process_message("Hello")
# Verify interactions
assert agent.model.generate.called
assert agent.memory.store.called
assert "Hello" in response
@patch('model_module.ArkModelNew.ArkModel')
@patch('memory_module.memory.Memory')
def test_error_recovery(self, mock_memory, mock_model):
"""Test agent recovers from errors gracefully"""
# Simulate model failure then success
mock_model.generate.side_effect = [
Exception("API Error"),
Mock(content="Recovered response")
]
agent = Agent(model=mock_model, memory=mock_memory)
response = agent.process_message("test")
assert response == "Recovered response"
assert mock_model.generate.call_count == 2
```
### Planned Async Tests (Coming soon!)
Future async tests will test asynchronous operations:
```python theme={null}
# tests/unit/test_model/test_async_generation.py
import pytest
import asyncio
from model_module.ArkModelNew import ArkModel
class TestAsyncGeneration:
@pytest.fixture
async def model(self):
"""Create model for async testing"""
return ArkModel(provider="mock")
@pytest.mark.asyncio
async def test_async_generate(self, model):
"""Test async text generation"""
response = await model.agenerate("Test prompt")
assert response.content is not None
assert response.success
@pytest.mark.asyncio
async def test_concurrent_requests(self, model):
"""Test multiple concurrent requests"""
prompts = ["Prompt 1", "Prompt 2", "Prompt 3"]
tasks = [model.agenerate(p) for p in prompts]
responses = await asyncio.gather(*tasks)
assert len(responses) == 3
assert all(r.success for r in responses)
@pytest.mark.asyncio
@pytest.mark.timeout(5)
async def test_streaming(self, model):
"""Test async streaming responses"""
chunks = []
async for chunk in model.astream("Generate story"):
chunks.append(chunk)
assert len(chunks) > 0
complete = "".join(chunks)
assert len(complete) > 0
```
## Testing Patterns
### Fixtures
Create reusable test components:
```python theme={null}
# tests/conftest.py
import pytest
from pathlib import Path
import tempfile
import shutil
@pytest.fixture(scope="session")
def test_data_dir():
"""Provide test data directory"""
return Path(__file__).parent / "fixtures"
@pytest.fixture
def temp_dir():
"""Create temporary directory for test"""
temp = tempfile.mkdtemp()
yield Path(temp)
shutil.rmtree(temp)
@pytest.fixture
def mock_llm_response():
"""Mock LLM response factory"""
def _make_response(content="Test response", tokens=10):
return {
"content": content,
"prompt_tokens": tokens,
"completion_tokens": tokens * 2,
"total_tokens": tokens * 3
}
return _make_response
@pytest.fixture(scope="function")
def clean_memory():
"""Provide clean memory instance"""
from memory_module.memory import Memory
memory = Memory(backend="memory") # In-memory for tests
yield memory
memory.clear() # Cleanup
```
### Mocking
Mock external dependencies:
```python theme={null}
# tests/unit/test_tool/test_weather_tool.py
import pytest
from unittest.mock import patch, Mock
from tool_module.weather import WeatherTool
class TestWeatherTool:
@patch('requests.get')
def test_get_weather(self, mock_get):
"""Test weather API call with mocked response"""
# Mock API response
mock_get.return_value = Mock(
status_code=200,
json=lambda: {
"temp": 72,
"condition": "Sunny",
"humidity": 45
}
)
tool = WeatherTool(api_key="test")
result = tool.get_current("Boston")
assert result.success
assert result.data["temp"] == 72
mock_get.assert_called_once()
@patch('requests.get')
def test_api_error_handling(self, mock_get):
"""Test handling of API errors"""
mock_get.return_value = Mock(
status_code=500,
text="Internal Server Error"
)
tool = WeatherTool(api_key="test")
result = tool.get_current("Boston")
assert not result.success
assert "error" in result.error.lower()
```
### Parametrized Tests
Test multiple scenarios efficiently:
```python theme={null}
# tests/unit/test_memory/test_memory_types.py
import pytest
from memory_module.memory import MemoryEntry, MemoryType
class TestMemoryTypes:
@pytest.mark.parametrize("memory_type,expected_decay,expected_capacity", [
(MemoryType.SENSORY, 0.99, 100),
(MemoryType.WORKING, 0.5, 9),
(MemoryType.SHORT_TERM, 0.1, 1000),
(MemoryType.LONG_TERM, 0.001, None),
])
def test_memory_characteristics(self, memory_type, expected_decay, expected_capacity):
"""Test memory type characteristics"""
entry = MemoryEntry(
content="Test",
type=memory_type
)
assert entry.decay_rate == pytest.approx(expected_decay, 0.01)
if expected_capacity:
assert memory_type.max_capacity == expected_capacity
```
## Performance Testing
### Benchmark Tests
```python theme={null}
# tests/performance/test_memory_performance.py
import pytest
from memory_module.memory import Memory, MemoryEntry
class TestMemoryPerformance:
@pytest.fixture
def large_memory(self):
"""Create memory with many entries"""
memory = Memory(backend="sqlite")
for i in range(1000):
memory.store(MemoryEntry(content=f"Memory {i}"))
return memory
@pytest.mark.benchmark
def test_search_performance(self, benchmark, large_memory):
"""Benchmark memory search performance"""
result = benchmark(
large_memory.search,
"Memory 500",
k=10
)
assert len(result) > 0
@pytest.mark.benchmark
def test_store_performance(self, benchmark, large_memory):
"""Benchmark memory storage performance"""
entry = MemoryEntry(content="New memory")
benchmark(large_memory.store, entry)
```
### Load Testing
```python theme={null}
# tests/performance/test_model_load.py
import pytest
import asyncio
from concurrent.futures import ThreadPoolExecutor
class TestModelLoad:
@pytest.mark.slow
def test_concurrent_load(self, model):
"""Test model under concurrent load"""
num_requests = 100
def make_request(i):
return model.generate(f"Request {i}")
with ThreadPoolExecutor(max_workers=10) as executor:
futures = [
executor.submit(make_request, i)
for i in range(num_requests)
]
results = [f.result() for f in futures]
assert len(results) == num_requests
success_rate = sum(1 for r in results if r.success) / num_requests
assert success_rate > 0.95 # 95% success rate
```
## Test Coverage (Coming soon!)
### Planned Coverage Testing
```bash theme={null}
# Future coverage commands (Coming soon!)
pytest --cov=. --cov-report=html
pytest --cov=state_module tests/unit/test_state/
pytest --cov=. --cov-report=term-missing
```
### Planned Coverage Goals
| Module | Target | Current Status |
| ------ | ------ | ---------------- |
| State | 85% | Coming soon! |
| Memory | 80% | Coming soon! |
| Model | 75% | Basic tests only |
| Tool | 80% | Coming soon! |
| Agent | 85% | Coming soon! |
## CI/CD Integration
### GitHub Actions Workflow
```yaml theme={null}
# .github/workflows/test.yml
name: Tests
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.8", "3.9", "3.10", "3.11"]
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Cache dependencies
uses: actions/cache@v3
with:
path: ~/.cache/pip
key: ${{ runner.os }}-pip-${{ hashFiles('requirements.txt') }}
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install -r requirements-test.txt
- name: Run tests
run: |
pytest tests/ -v --cov=. --cov-report=xml
- name: Upload coverage
uses: codecov/codecov-action@v3
with:
file: ./coverage.xml
fail_ci_if_error: true
```
## Testing Best Practices
### Guidelines
1. **Test Isolation**: Each test should be independent
2. **Clear Names**: Test names should describe what they test
3. **Arrange-Act-Assert**: Follow AAA pattern
4. **One Assertion**: Generally one assertion per test
5. **Fast Tests**: Keep unit tests under 100ms
6. **Mock External**: Mock all external dependencies
7. **Test Behavior**: Test behavior, not implementation
### Common Pitfalls
Avoid these common testing mistakes:
* Testing private methods directly
* Over-mocking leading to meaningless tests
* Ignoring edge cases
* Not testing error paths
* Brittle tests that break with refactoring
## Running Tests
### Current Testing Commands
```bash theme={null}
# Run current basic tests (if available)
python -m pytest tests/ -v
```
### Planned Test Commands (Coming soon!)
```bash theme={null}
# Future comprehensive test commands (Coming soon!)
pytest # Run all tests
pytest -v # Verbose output
pytest tests/unit/test_state/ # Specific test file
pytest -k "memory" # Tests matching pattern
pytest -m "unit and not slow" # Marked tests
pytest --log-level=DEBUG # With debug logging
pytest -x # Stop on first failure
pytest --lf # Run failed tests from last run
pytest -n 4 # Parallel execution
```
## Debugging Tests
### Using PDB
```python theme={null}
def test_complex_logic():
data = process_data()
import pdb; pdb.set_trace() # Debugger breakpoint
assert data.is_valid()
```
### Verbose Logging
```python theme={null}
import logging
def test_with_logging(caplog):
with caplog.at_level(logging.DEBUG):
result = complex_function()
assert "Expected log message" in caplog.text
assert result == expected_value
```
## Next Steps
Set up your development environment
Contribution guidelines
API testing documentation
Example test implementations
# ARKOS - Introduction
Source: https://arkos.mintlify.app/introduction
Automated Resource Knowledgebase Operating System - An open-source interface for local LLM agent building with long-term memory
# Welcome to ARKOS
**ARK (Automated Resource Knowledgebase) Operating System** improves resource management through intelligent automation. Using tailored processess and adaptive memory, ARKOS streamlines the collection, organization, and access to resource data, facilitating efficient decision-making and personalized AI interactions.
ARKOS is an open-source interface for local LLM agent building utilizing long-term memory for personalized requests. We're primarily focusing on targeting MIT students for initial deployment.
## What is ARKOS?
ARKOS is a comprehensive framework for building intelligent agents with the following features:
**Core Features:**
* **Multi-Provider LLM Support**: Unified interface for OpenAI-compatible endpoints, SGLANG, and local models via AsyncOpenAI client
* **Dual-tier Memory System**: PostgreSQL for short-term context + Mem0/Supabase for long-term vector memory
* **State Machine Architecture**: YAML-configurable state graphs for conversation flow management
* **MCP Tool Integration**: Model Context Protocol support for connecting external tools and services
* **OpenAI-Compatible API**: FastAPI server with `/v1/chat/completions` endpoint (streaming supported)
* **CLI Interface**: Interactive command-line interface for testing and development
## Key Features
Seamless integration with SGLANG, local models, and OpenAI-compatible endpoints
PostgreSQL for conversation context + Mem0/Supabase for semantic long-term memory
YAML-configurable state graphs with automatic transition handling
Model Context Protocol integration with per-user authentication support
Real-time streaming responses via OpenAI-compatible endpoint
Intelligent agent with tool calling, memory retrieval, and state management
## Architecture Overview
ARKOS employs a modular architecture with clear separation of concerns:
```mermaid theme={null}
graph TB
subgraph "Core System"
A[Base Module
FastAPI Server] --> B[Agent Module]
B --> C[State Module]
B --> D[Model Module]
B --> E[Memory Module]
B --> F[Tool Module]
end
subgraph "Memory Layer"
E --> G[PostgreSQL
Short-term]
E --> H[Mem0/Supabase
Long-term Vector]
end
subgraph "External Services"
D --> I[SGLANG Server]
F --> J[MCP Servers]
end
```
## Technology Stack
* **Core Language**: Python 3.8+
* **Web Framework**: FastAPI + Uvicorn
* **LLM Interface**: AsyncOpenAI client (OpenAI-compatible)
* **Data Validation**: Pydantic 2.x
* **Database**: PostgreSQL (psycopg2-binary) + Supabase
* **Memory**: Mem0 for vector storage and retrieval
* **Embeddings**: Sentence Transformers via HuggingFace
* **Configuration**: YAML + python-dotenv
* **HTTP Client**: aiohttp for async requests
* **Authentication**: google-auth-oauthlib for OAuth flows
## Project Status
ARKOS is under active development. The core architecture is stable, but APIs may change as we refine the system based on user feedback and research insights.
## Use Cases
ARKOS is designed for:
1. **Personal AI Assistants**: Build agents that remember conversations and learn from interactions
2. **Research Tools**: Create intelligent research assistants with persistent knowledge bases
3. **Educational Applications**: Develop tutoring systems that adapt to student learning patterns
4. **Task Automation**: Build agents that can manage complex, multi-step workflows
5. **Knowledge Management**: Create systems that organize and retrieve information intelligently
## Getting Started
Ready to build your first ARKOS agent?
Get up and running with ARKOS in minutes
## Contributing
ARKOS is open-source and welcomes contributions! Whether you're interested in adding new features, improving documentation, or reporting bugs, we'd love to have you involved.
View source code and contribute
Set up your development environment
## Becoming a Core Maintainer
We are actively looking for committed contributors who want to help shape the long term direction of the project.
Core maintainers are expected to:
* Make regular, high quality contributions
* Participate in design and architecture discussions
* Review pull requests and help guide technical decisions
* Uphold the project’s standards around reliability, security, and documentation
* Attend weekly maintainer meetings to coordinate work and align on priorities
If you are interested in becoming a core maintainer, email **[sipb-ark@mit.edu](mailto:sipb-ark@mit.edu)** with a brief introduction and a summary of your contributions or areas of interest.
## Team
| Name | Role | Affiliation |
| ---------------- | -------------- | ------------ |
| Nathaniel Morgan | Project Leader | MIT |
| Kshitij Duraphe | Dev Ops | Industry |
| Ilya Gulko | Backend | MIT |
| Jack Luo | Backend | Georgia Tech |
| Angela Liu | Backend | MIT |
| Ishaana Misra | Backend | MIT |
| Hudson Hilal | Backend | MIT |
| Joshua Guo | Frontend | MIT |
| Calvin Baker | Backend | MIT |
## License
ARKOS is released under the AGPL-3.0 License.
# Agent Module Overview
Source: https://arkos.mintlify.app/modules/agent/overview
Intelligent agent orchestration with state machine and tool integration
# Agent Module Overview
The Agent Module is the core orchestrator of ARKOS, coordinating the state machine, memory, LLM, and tool systems to provide intelligent conversational interactions.
## Core Concepts
The Agent class manages the complete conversation lifecycle, including context retrieval, state transitions, tool execution, and response generation.
## Architecture
```mermaid theme={null}
graph TB
subgraph "Agent Core"
AGENT[Agent]
AGENT --> FLOW[StateHandler]
AGENT --> MEM[Memory]
AGENT --> LLM[ArkModelLink]
AGENT --> TOOLS[MCPToolManager]
end
subgraph "Key Methods"
STEP[step]
STREAM[step_stream]
CALL[call_llm]
CTX[get_context]
end
subgraph "External"
API[FastAPI Server]
API --> AGENT
end
AGENT --> STEP
AGENT --> STREAM
AGENT --> CALL
AGENT --> CTX
```
## Agent Class
### Initialization
```python theme={null}
from agent_module.agent import Agent
from state_module.state_handler import StateHandler
from memory_module.memory import Memory
from model_module.ArkModelNew import ArkModelLink
from tool_module.tool_call import MCPToolManager
# Initialize components
flow = StateHandler(yaml_path="state_module/state_graph.yaml")
memory = Memory(
user_id="default_user",
session_id=None,
db_url=db_url,
use_long_term=False
)
llm = ArkModelLink(base_url="http://localhost:30000/v1")
tool_manager = MCPToolManager(mcp_config, token_store=token_store)
# Create agent
agent = Agent(
agent_id="my-agent",
flow=flow,
memory=memory,
llm=llm,
tool_manager=tool_manager
)
```
### Key Attributes
```python theme={null}
class Agent:
agent_id: str # Unique identifier
flow: StateHandler # State machine
memory: Memory # Memory system
llm: ArkModelLink # LLM interface
tool_manager: MCPToolManager # Tool manager (optional)
current_state: State # Current state in flow
system_prompt: str # System prompt with tools
available_tools: dict # Discovered MCP tools
current_user_id: str # For per-user tool auth
```
## Core Methods
### step() - Main Processing Loop
Process messages through the state machine:
```python theme={null}
async def step(self, messages, user_id: str = None):
"""
Runs the agent until reaching a terminal state.
Parameters
----------
messages : list
List of messages to process
user_id : str, optional
User ID for per-user tool authentication
Returns
-------
AIMessage
The last AI message produced
"""
self.current_user_id = user_id
self.add_context(messages)
last_ai_message = None
retry_count = 0
while not self.current_state.is_terminal:
if retry_count > MAX_ITER: # MAX_ITER = 10
break
retry_count += 1
context = self.get_context()
update = await self.current_state.run(context, self)
if update:
self.add_context([update])
if isinstance(update, AIMessage):
last_ai_message = update
if self.current_state.is_terminal:
break
# Handle state transitions
messages_list = self.memory.retrieve_short_memory(5)
if self.current_state.check_transition_ready(messages_list):
transition_dict = self.flow.get_transitions(
self.current_state, messages_list
)
transition_names = transition_dict["tt"]
if len(transition_names) == 1:
next_state_name = transition_names[0]
else:
next_state_name = await self.choose_transition(
transition_dict, messages_list
)
self.current_state = self.flow.get_state(next_state_name)
# Reset to initial state for next conversation
self.current_state = self.flow.get_state("agent_reply")
return last_ai_message
```
### step\_stream() - Streaming Response
Stream responses character by character:
```python theme={null}
async def step_stream(self, messages, user_id: str = None):
"""
Streaming version of step. Yields characters as they're produced.
Yields
------
str
Characters/chunks from each state's output
"""
self.current_user_id = user_id
self.add_context(messages)
retry_count = 0
while not self.current_state.is_terminal:
if retry_count > MAX_ITER:
yield "\n[Max iterations reached]"
break
retry_count += 1
context = self.get_context()
try:
update = await self.current_state.run(context, self)
except Exception as e:
update = AIMessage(content=f"Error: {str(e)[:200]}")
self.current_state = self.flow.get_state("agent_reply")
# Stream character by character
if update and hasattr(update, 'content') and update.content:
self.add_context([update])
for char in update.content:
yield char
if self.current_state.is_terminal:
break
# Handle transitions (same as step)
# ... transition logic ...
if not self.current_state.is_terminal:
yield "\n\n" # Separator between states
self.current_state = self.flow.get_state("agent_reply")
```
### call\_llm() - LLM Interface
Call the language model:
```python theme={null}
async def call_llm(self, context=None, json_schema=None):
"""
Agent's interface with chat model.
Parameters
----------
context : list
Messages to send to LLM
json_schema : dict, optional
JSON schema for structured output
Returns
-------
AIMessage
The LLM's response
"""
llm_response = await self.llm.generate_response(context, json_schema)
return AIMessage(content=llm_response)
```
### Context Management
```python theme={null}
def add_context(self, messages):
"""Store messages in memory."""
for message in messages:
self.memory.add_memory(message)
def get_context(self, turns=5, include_long_term=True):
"""
Retrieve context for LLM.
Returns list of messages including:
- Long-term memories (if enabled)
- Recent conversation turns
"""
short_term_mem = self.memory.retrieve_short_memory(turns)
if include_long_term:
long_term_mem = self.memory.retrieve_long_memory(context=short_term_mem)
if long_term_mem and long_term_mem.content.strip():
return [long_term_mem] + short_term_mem
return short_term_mem
```
## Transition Selection
When multiple state transitions are possible:
```python theme={null}
async def choose_transition(self, transitions_dict, messages):
"""Use LLM to select the best next state."""
transition_tuples = list(zip(
transitions_dict["tt"],
transitions_dict["td"]
))
prompt = f"""Given the conversation context and these options:
{transition_tuples}
Output the most reasonable next state.
Do not use tool result to determine the next state."""
# Create Pydantic model for structured output
NextStates = self.create_next_state_class(transition_tuples)
json_schema = {
"type": "json_schema",
"json_schema": {
"name": "class_options",
"schema": NextStates.model_json_schema(),
},
}
context_text = [SystemMessage(content=prompt)] + messages
output = await self.call_llm(context=context_text, json_schema=json_schema)
return json.loads(output.content)["next_state"]
```
## Tool Integration
### Tool Argument Classes
```python theme={null}
def fill_tool_args_class(self, tool_name: str, tool_args: Dict[str, Any]):
"""Create Pydantic model for tool call."""
ToolCall = create_model(
"ToolCall",
tool_name=(str, Field(description="Tool name to execute")),
tool_args=(Dict[str, Any], Field(default_factory=dict)),
)
return ToolCall(tool_name=tool_name, tool_args=tool_args)
async def create_tool_option_class(self):
"""Create enum of available tools for LLM selection."""
server_tool_map = await self.tool_manager.list_all_tools()
enum_members = {}
for server_name in server_tool_map:
for tool_name in server_tool_map[server_name]:
enum_members[tool_name] = tool_name
ToolEnum = Enum("ToolEnum", enum_members)
return create_model(
"ToolCall",
tool_name=(ToolEnum, Field(description="Tool to execute next"))
)
```
## Usage with FastAPI
The agent is used in the API server:
```python theme={null}
# base_module/app.py
from agent_module.agent import Agent
# Initialize agent at startup
agent = Agent(
agent_id=config.get("memory.user_id"),
flow=flow,
memory=memory,
llm=llm,
tool_manager=tool_manager,
)
@app.on_event("startup")
async def startup():
if tool_manager:
await tool_manager.initialize_servers()
agent.available_tools = await tool_manager.list_all_tools()
agent.system_prompt = base_prompt + "\n\n" + tool_prompt
@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
# ... parse request ...
if stream:
async def generate_stream():
async for chunk in agent.step_stream(context_msgs, user_id=user_id):
# Format as SSE
yield f"data: {json.dumps(data)}\n\n"
return StreamingResponse(generate_stream(), media_type="text/event-stream")
# Non-streaming
agent_response = await agent.step(context_msgs, user_id=user_id)
return JSONResponse(content=completion)
```
## Performance Timing
The agent includes built-in timing logs:
```python theme={null}
# Output during execution:
# [TIMING] add_context: 0.015s
# [TIMING] get_context: 0.102s
# [TIMING] state.run: 1.234s
# [TIMING] loop total: 1.351s
# [TIMING] step total: 2.847s
```
## Configuration
Agents are configured via `config_module/config.yaml`:
```yaml theme={null}
memory:
user_id: "default_user"
use_long_term: false
llm:
base_url: "http://localhost:30000/v1"
state:
graph_path: "state_module/state_graph.yaml"
app:
system_prompt: "You are a helpful AI assistant."
```
## Error Handling
```python theme={null}
async def step_stream(self, messages, user_id=None):
# ...
try:
update = await self.current_state.run(context, self)
except Exception as e:
print(f"[STREAM] State error: {e}")
update = AIMessage(content=f"Error: {str(e)[:200]}")
self.current_state = self.flow.get_state("agent_reply")
# ...
```
## Debug Logging
The agent logs state transitions:
```python theme={null}
print("agent.py received message")
print("agent.py CURR STATE:", self.current_state)
print("agent.py IS TERMINAL?:", self.current_state.is_terminal)
print(f"Inner loop #{retry_count + 1}")
print("REACHED TERMINAL")
print("LAST_AI_MSG", last_ai_message)
```
## Best Practices
1. **Always set user\_id**: For per-user tool authentication
2. **Monitor MAX\_ITER**: Prevent infinite loops
3. **Use streaming**: For better UX with long responses
4. **Reset state**: Agent resets to initial state after each conversation
5. **Handle errors**: Graceful degradation on state failures
## Troubleshooting
Check state machine for deadlocks:
```python theme={null}
print(f"Current state: {agent.current_state.name}")
print(f"Is terminal: {agent.current_state.is_terminal}")
```
Verify tool manager is initialized:
```python theme={null}
print(f"Available tools: {agent.available_tools.keys()}")
```
Increase the turns parameter (config.yaml):
```yaml theme={null}
short_term_turns: 50
long_term_limit: 50
```
## Next Steps
Learn about state management
Explore memory systems
Understand tool integration
See the API server
# Base Module Overview
Source: https://arkos.mintlify.app/modules/base/overview
FastAPI server with OpenAI-compatible API and CLI interface
# Base Module Overview
The Base Module provides the primary interface layer for ARKOS, including a FastAPI server with OpenAI-compatible endpoints and an interactive CLI for testing.
## Core Components
OpenAI-compatible `/v1/chat/completions` endpoint with streaming support
Interactive command-line interface for testing and development
Google OAuth integration for per-user tool authentication
Health check endpoint for system status
## Architecture
```mermaid theme={null}
graph TB
subgraph "Base Module"
APP["app.py
FastAPI Server"]
CLI["main_interface.py
CLI Interface"]
AUTH["auth.py
OAuth Routes"]
end
subgraph "Endpoints"
APP --> CHAT["/v1/chat/completions"]
APP --> HEALTH["/health"]
end
subgraph "Agent Integration"
APP --> AGENT["Agent"]
AGENT --> FLOW["StateHandler"]
AGENT --> MEM["Memory"]
AGENT --> LLM["ArkModelLink"]
AGENT --> TOOLS["MCPToolManager"]
end
```
## FastAPI Server (app.py)
### Starting the Server
```bash theme={null}
python base_module/app.py
```
This starts the server on **port 1111** by default.
### Endpoints
#### POST /v1/chat/completions
OpenAI-compatible chat completions endpoint:
```python theme={null}
@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
"""OAI-compatible endpoint wrapping the full ArkOS agent."""
payload = await request.json()
messages = payload.get("messages", [])
model = payload.get("model", "ark-agent")
stream = payload.get("stream", False)
user_id = request.headers.get("X-User-ID") or payload.get("user_id")
# Convert to internal message format
context_msgs = [SystemMessage(content=agent.system_prompt)]
for msg in messages:
if msg["role"] == "system":
context_msgs.append(SystemMessage(content=msg["content"]))
elif msg["role"] == "user":
context_msgs.append(UserMessage(content=msg["content"]))
elif msg["role"] == "assistant":
context_msgs.append(AIMessage(content=msg["content"]))
if stream:
return StreamingResponse(generate_stream(), media_type="text/event-stream")
agent_response = await agent.step(context_msgs, user_id=user_id)
return JSONResponse(content=completion)
```
**Request Format:**
```json theme={null}
{
"model": "ark-agent",
"messages": [
{"role": "system", "content": "You are helpful"},
{"role": "user", "content": "Hello!"}
],
"stream": false,
"user_id": "alice"
}
```
**Response Format:**
```json theme={null}
{
"id": "chatcmpl-abc123",
"object": "chat.completion",
"created": 1706300000,
"model": "ark-agent",
"choices": [{
"index": 0,
"message": {"role": "assistant", "content": "Hello! How can I help?"},
"finish_reason": "stop"
}]
}
```
#### GET /health
Health check endpoint:
```python theme={null}
@app.get("/health")
async def health_check():
"""Health check endpoint to verify server and dependencies."""
llm_status = "unknown"
try:
response = requests.get("http://localhost:30000/v1/models", timeout=2)
llm_status = "running" if response.status_code == 200 else "error"
except:
llm_status = "not_running"
return JSONResponse(content={
"status": "ok",
"llm_server": llm_status,
"port": 1111
})
```
### Streaming Responses
```python theme={null}
async def generate_stream():
chunk_id = f"chatcmpl-{uuid.uuid4().hex[:8]}"
async for chunk in agent.step_stream(context_msgs, user_id=user_id):
data = {
"id": chunk_id,
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": model,
"choices": [{
"index": 0,
"delta": {"content": chunk},
"finish_reason": None,
}],
}
yield f"data: {json.dumps(data)}\n\n"
# Final chunk
yield f"data: {json.dumps(final_data)}\n\n"
yield "data: [DONE]\n\n"
```
### Startup Initialization
```python theme={null}
@app.on_event("startup")
async def startup():
base_system_prompt = config.get("app.system_prompt") or ""
if tool_manager:
await tool_manager.initialize_servers()
print(f"Initialized {len(tool_manager.clients)} MCP servers")
agent.available_tools = await tool_manager.list_all_tools()
print(f"Available tools: {list(agent.available_tools.keys())}")
tool_prompt = format_tools_for_system_prompt(agent.available_tools)
agent.system_prompt = base_system_prompt + "\n\n" + tool_prompt
else:
agent.system_prompt = base_system_prompt
```
## CLI Interface (main\_interface.py)
### Starting the CLI
```bash theme={null}
python base_module/main_interface.py
```
### Features
* Interactive conversation loop
* Type messages and press Enter to send
* Type `exit` or `quit` to stop
* Rich terminal output support (main\_interface\_rich.py)
### Basic Usage
```python theme={null}
from base_module.main_interface import MainInterface
interface = MainInterface(
config_path="config_module/config.yaml"
)
interface.run_interactive()
```
## OAuth Authentication (auth.py)
Routes for per-user OAuth authentication:
```python theme={null}
from base_module.auth import router as auth_router
app.include_router(auth_router)
# Provides:
# GET /auth/google/login?user_id=alice
# GET /auth/google/callback
```
### Google Calendar OAuth Flow
1. User calls `/auth/google/login?user_id=alice`
2. Redirects to Google OAuth consent screen
3. Google redirects to `/auth/google/callback`
4. Token stored in `UserTokenStore`
5. User can now use Google Calendar tools
## Configuration
### Environment Variables
```bash theme={null}
# .env
# Database (REQUIRED)
DB_URL=postgresql://user:pass@host:port/db
# Google OAuth (for calendar integration)
GOOGLE_OAUTH_CREDENTIALS=/path/to/credentials.json
# MCP tool credentials
BRAVE_API_KEY=your-api-key
```
### YAML Configuration
```yaml theme={null}
# config_module/config.yaml
app:
host: "0.0.0.0"
port: 1111
reload: false
system_prompt: "You are a helpful AI assistant."
llm:
base_url: "http://localhost:30000/v1"
database:
url: "${DB_URL}"
memory:
user_id: "default_user"
use_long_term: false
state:
graph_path: "state_module/state_graph.yaml"
mcp_servers:
google-calendar:
transport: stdio
command: npx
args: ["-y", "@anthropic/google-calendar-mcp"]
```
## Using the API
### Python Client
```python theme={null}
from openai import OpenAI
client = OpenAI(
base_url="http://localhost:1111/v1",
api_key="not-needed"
)
# Non-streaming
response = client.chat.completions.create(
model="ark-agent",
messages=[{"role": "user", "content": "Hello!"}]
)
print(response.choices[0].message.content)
# Streaming
stream = client.chat.completions.create(
model="ark-agent",
messages=[{"role": "user", "content": "Hello!"}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="")
```
### cURL
```bash theme={null}
# Non-streaming
curl -X POST http://localhost:1111/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{
"model": "ark-agent",
"messages": [{"role": "user", "content": "Hello!"}]
}'
# Streaming
curl -X POST http://localhost:1111/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{
"model": "ark-agent",
"messages": [{"role": "user", "content": "Hello!"}],
"stream": true
}'
# With user ID for tool auth
curl -X POST http://localhost:1111/v1/chat/completions \
-H "Content-Type: application/json" \
-H "X-User-ID: alice" \
-d '{
"model": "ark-agent",
"messages": [{"role": "user", "content": "What's on my calendar?"}]
}'
```
## Tool System Prompt
Tools are formatted for the system prompt:
```python theme={null}
def format_tools_for_system_prompt(tools: dict) -> str:
lines = [
"You have access to the following tools.",
"Use them when appropriate. Only call tools that are listed below.",
""
]
for name, tool in tools.items():
lines.append(f"Tool name: {name}")
if getattr(tool, "description", None):
lines.append(f"Description: {tool.description}")
if getattr(tool, "input_schema", None):
lines.append("Input schema:")
lines.append(str(tool.input_schema))
lines.append("")
return "\n".join(lines)
```
## Error Handling
The server handles common error cases:
* **Missing LLM server**: Health check reports `llm_server: "not_running"`
* **Database errors**: Logged with traceback
* **Tool auth required**: Returns structured error for frontend handling
## Deployment
### Development
```bash theme={null}
python base_module/app.py
```
### Production with Uvicorn
```bash theme={null}
uvicorn base_module.app:app \
--host 0.0.0.0 \
--port 1111 \
--workers 4
```
### Shared Server Considerations
On shared servers like MIT SIPB:
1. Check if LLM server is already running on port 30000
2. Use unique ports if running multiple instances
3. Coordinate with team on port assignments
## Troubleshooting
Ensure the server is running:
```bash theme={null}
python base_module/app.py
```
Check if SGLang is running:
```bash theme={null}
curl http://localhost:30000/v1/models
```
Verify user has connected the service:
```python theme={null}
status = tool_manager.get_user_service_status(user_id)
```
Find and kill the process:
```bash theme={null}
lsof -i :1111
kill -9
```
## Next Steps
Learn about agent orchestration
Explore tool integration
Get started with ARKOS
Complete API documentation
# Memory Module Overview
Source: https://arkos.mintlify.app/modules/memory/overview
Dual-tier persistent memory system with PostgreSQL and vector search
# Memory Module Overview
The ARKOS Memory Module implements a dual-tier memory system combining PostgreSQL for conversation context storage with Mem0/Supabase for semantic long-term memory retrieval.
## Core Concepts
The Memory Module provides both short-term conversation context (PostgreSQL) and long-term semantic memory (Mem0 with Supabase vector store) for personalized AI interactions.
### Memory Architecture
The system uses two complementary storage backends:
1. **Short-term Memory (PostgreSQL)**: Stores recent conversation turns with full message content
2. **Long-term Memory (Mem0/Supabase)**: Vector-based semantic storage for retrieval across sessions
## Architecture Diagram
```mermaid theme={null}
graph TB
subgraph "Memory Module"
M[Memory Class]
M --> STM[Short-term Memory]
M --> LTM[Long-term Memory]
end
subgraph "Short-term Storage"
STM --> PG[PostgreSQL]
PG --> POOL[Connection Pool]
end
subgraph "Long-term Storage"
LTM --> MEM0[Mem0]
MEM0 --> SUPA[Supabase Vector Store]
MEM0 --> EMB[HuggingFace Embeddings]
end
subgraph "Message Types"
MSG[Message Base Class]
MSG --> UM[UserMessage]
MSG --> AM[AIMessage]
MSG --> SM[SystemMessage]
MSG --> TM[ToolMessage]
end
```
## Message Types
All memory operations use Pydantic message classes:
```python theme={null}
from model_module.ArkModelNew import (
Message,
UserMessage,
AIMessage,
SystemMessage,
ToolMessage
)
# User message
user_msg = UserMessage(content="Hello, how are you?")
# AI response
ai_msg = AIMessage(content="I'm doing well, thanks!")
# System instruction
system_msg = SystemMessage(content="You are a helpful assistant")
# Tool result
tool_msg = ToolMessage(content='{"result": "success"}')
```
## Memory Class
### Initialization
```python theme={null}
from memory_module.memory import Memory
memory = Memory(
user_id="alice", # Required: identifies the user
session_id=None, # Optional: auto-generates UUID if None
db_url="postgresql://...", # Required: PostgreSQL connection string
use_long_term=True # Optional: enable Mem0 (default: True)
)
```
### Configuration
The Memory class uses global configuration for Mem0:
```python theme={null}
# Configured in memory_module/memory.py
config = {
"vector_store": {
"provider": "supabase",
"config": {
"connection_string": os.environ["DB_URL"],
"collection_name": "memories",
"index_method": "hnsw",
"index_measure": "cosine_distance",
},
},
"llm": {
"provider": "vllm",
"config": {
"model": "Qwen/Qwen2.5-7B-Instruct",
"vllm_base_url": "http://localhost:30000/v1",
},
},
"embedder": {
"provider": "huggingface",
"config": {"huggingface_base_url": "http://localhost:4444/v1"},
},
}
```
## Memory Operations
### Adding Memories
Store a message to both PostgreSQL (immediate) and Mem0 (background):
```python theme={null}
from model_module.ArkModelNew import UserMessage, AIMessage
# Add user message
memory.add_memory(UserMessage(content="My favorite color is blue"))
# Add AI response
memory.add_memory(AIMessage(content="I'll remember that you like blue!"))
```
Messages are stored in PostgreSQL immediately (fast, synchronous) and sent to Mem0 in a background thread (non-blocking) for long-term storage.
### Retrieving Short-term Memory
Get recent conversation turns from PostgreSQL:
```python theme={null}
# Get last 5 conversation turns
context = memory.retrieve_short_memory(turns=5)
# Returns list of Message objects in chronological order
for msg in context:
print(f"{msg.role}: {msg.content}")
```
### Retrieving Long-term Memory
Search for semantically relevant memories using Mem0:
```python theme={null}
# Get recent context first
context = memory.retrieve_short_memory(turns=2)
# Retrieve relevant long-term memories
long_term = memory.retrieve_long_memory(
context=context, # Used to build search query
mem0_limit=10 # Max results to return
)
# Returns a SystemMessage with retrieved memories
print(long_term.content)
# Output: "retrieved memories:\nuser: My favorite color is blue\n..."
```
### Session Management
```python theme={null}
# Start a new session (generates new UUID)
new_session_id = memory.start_new_session()
# Access current session
print(memory.session_id)
```
## Database Schema
The PostgreSQL `conversation_context` table:
```sql theme={null}
CREATE TABLE conversation_context (
id SERIAL PRIMARY KEY,
user_id VARCHAR(255) NOT NULL,
session_id VARCHAR(255) NOT NULL,
role VARCHAR(50) NOT NULL,
message TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_user_id ON conversation_context(user_id);
CREATE INDEX idx_session_id ON conversation_context(session_id);
```
## Message Serialization
Messages are serialized to JSON for storage:
```python theme={null}
# Serialize message to JSON string
json_str = memory.serialize(user_msg)
# '{"content": "Hello", "role": "user"}'
# Deserialize back to Message object
msg = memory.deserialize(json_str, role="user")
# UserMessage(content="Hello", role="user")
```
## Connection Pooling
The module uses a global connection pool for efficiency:
```python theme={null}
# Pool is initialized lazily on first use
# Configured with:
# - minconn=1 (minimum connections)
# - maxconn=10 (maximum connections)
# Connections are automatically managed
conn = pool.getconn()
try:
# Use connection
cur = conn.cursor()
cur.execute(...)
conn.commit()
finally:
pool.putconn(conn) # Return to pool
```
## Background Processing
Long-term memory storage happens in a background thread to avoid blocking:
```python theme={null}
# Background executor for non-blocking mem0 operations
_executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="mem0_bg")
def _add_to_mem0_background(self, content: str, metadata: dict):
"""Background task to add to mem0 (non-blocking)."""
try:
if self._mem0:
self._mem0.add(
messages=content,
metadata=metadata,
user_id=self.user_id
)
except Exception as e:
print(f"[mem0 background] Error: {e}")
```
## Integration with Agent
The Agent module uses Memory for context management:
```python theme={null}
class Agent:
def __init__(self, ..., memory: Memory, ...):
self.memory = memory
def add_context(self, messages):
"""Store messages in memory."""
for message in messages:
self.memory.add_memory(message)
def get_context(self, turns=5, include_long_term=True):
"""Retrieve context for LLM."""
short_term = self.memory.retrieve_short_memory(turns)
if include_long_term:
long_term = self.memory.retrieve_long_memory(context=short_term)
if long_term and long_term.content.strip():
return [long_term] + short_term
return short_term
```
## Performance Considerations
### Optimizations Implemented
1. **Connection Pooling**: Reuses database connections
2. **Background Processing**: Mem0 operations don't block the main thread
3. **Lazy Initialization**: Mem0 only initialized if `use_long_term=True`
4. **Limited Query Scope**: Long-term search uses only recent context (last 2 messages)
### Disabling Long-term Memory
For faster responses, disable long-term memory:
```python theme={null}
memory = Memory(
user_id="alice",
session_id=None,
db_url=db_url,
use_long_term=False # Skip Mem0 initialization and operations
)
```
## Environment Variables
Required environment variables:
```bash theme={null}
# PostgreSQL connection (REQUIRED)
DB_URL=postgresql://user:password@host:port/database
# OpenAI API key (required by Mem0, can be placeholder)
OPENAI_API_KEY=sk-placeholder
```
## Error Handling
The module includes robust error handling:
```python theme={null}
def add_memory(self, message) -> bool:
"""Returns True on success, False on error."""
try:
# Store in PostgreSQL
# Store in Mem0 (background)
return True
except Exception as e:
traceback.print_exc()
return False
def retrieve_short_memory(self, turns):
"""Returns empty list on error."""
try:
# Query PostgreSQL
return messages
except Exception as e:
print(f"[retrieve_short_memory] Error: {e}")
return []
def retrieve_long_memory(self, context, mem0_limit=10):
"""Returns empty SystemMessage on error."""
try:
# Query Mem0
return SystemMessage(content=memory_string)
except Exception as e:
print(f"[retrieve_long_memory] Error: {e}")
return SystemMessage(content="")
```
## Testing
Basic test example:
```python theme={null}
if __name__ == "__main__":
test_instance = Memory(
user_id="alice_test",
session_id="session_test",
db_url=os.environ["DB_URL"]
)
# Test adding memory
print(test_instance.add_memory(
SystemMessage(content="My favorite color is blue")
))
# Test retrieval
context = test_instance.retrieve_short_memory(turns=2)
print(context)
print(test_instance.retrieve_long_memory(context))
```
## Next Steps
Deep dive into memory implementation
Learn about agent orchestration
Explore LLM integration
Set up your environment
# Model Module Overview
Source: https://arkos.mintlify.app/modules/model/overview
AsyncOpenAI-based LLM interface for SGLANG and OpenAI-compatible endpoints
# Model Module Overview
The ARKOS Model Module provides an asynchronous interface for communicating with LLMs using the AsyncOpenAI client. It's designed to work with SGLANG servers and any OpenAI-compatible endpoint.
## Core Components
Main class for LLM communication using AsyncOpenAI
Pydantic models for different message types
Non-blocking I/O for better performance
Real-time token streaming for responsive UX
## Architecture
```mermaid theme={null}
graph TB
subgraph "Model Module"
AML[ArkModelLink]
AML --> GEN[generate_response]
AML --> STREAM[generate_stream]
AML --> CALL[make_llm_call]
end
subgraph "Message Types"
MSG[Message Base]
MSG --> UM[UserMessage]
MSG --> AM[AIMessage]
MSG --> SM[SystemMessage]
MSG --> TM[ToolMessage]
end
subgraph "External"
CALL --> ASYNC[AsyncOpenAI]
ASYNC --> SGLANG[SGLANG Server]
end
```
## Message Classes
All messages extend the base `Message` Pydantic model:
```python theme={null}
from model_module.ArkModelNew import (
Message,
UserMessage,
AIMessage,
SystemMessage,
ToolMessage
)
# Base class
class Message(BaseModel):
content: str
role: str
# User input
user_msg = UserMessage(content="Hello!")
# role = "user"
# AI response
ai_msg = AIMessage(content="Hi there!")
# role = "assistant"
# content can be None for tool-only responses
# tool_calls: Optional[dict] = None
# System instruction
system_msg = SystemMessage(content="You are helpful")
# role = "system"
# Tool result
tool_msg = ToolMessage(content='{"result": "data"}')
# role = "tool"
# tool_calls: Optional[dict] = None
```
## ArkModelLink Class
### Initialization
```python theme={null}
from model_module.ArkModelNew import ArkModelLink
# Default configuration
llm = ArkModelLink()
# base_url = "http://0.0.0.0:30000/v1"
# model_name = "tgi"
# max_tokens = 1024
# temperature = 0.7
# Custom configuration
llm = ArkModelLink(
base_url="http://localhost:30000/v1",
model_name="qwen-2.5",
max_tokens=2048,
temperature=0.5
)
```
### Configuration Options
| Parameter | Default | Description |
| ------------- | --------------------------- | ---------------------------- |
| `model_name` | `"tgi"` | Model identifier for the API |
| `base_url` | `"http://0.0.0.0:30000/v1"` | LLM server endpoint |
| `max_tokens` | `1024` | Maximum response tokens |
| `temperature` | `0.7` | Creativity (0-2) |
## Core Methods
### generate\_response()
Main method for getting LLM responses:
```python theme={null}
async def generate_response(
self,
messages: List[Message],
json_schema: Optional[dict] = None
) -> str:
"""
Asynchronously generate a response from the model.
Parameters
----------
messages : List[Message]
List of messages in the conversation
json_schema : dict, optional
JSON schema for structured output
Returns
-------
str
The raw response content
"""
```
**Usage:**
```python theme={null}
from model_module.ArkModelNew import ArkModelLink, UserMessage, SystemMessage
llm = ArkModelLink(base_url="http://localhost:30000/v1")
messages = [
SystemMessage(content="You are a helpful assistant"),
UserMessage(content="What is Python?")
]
# Basic generation
response = await llm.generate_response(messages, json_schema=None)
print(response) # "Python is a programming language..."
```
### generate\_stream()
Stream tokens as they're generated:
```python theme={null}
async def generate_stream(
self,
messages: List[Message]
) -> AsyncIterator[str]:
"""
Stream tokens as they're generated.
Yields
------
str
Individual tokens/chunks
"""
```
**Usage:**
```python theme={null}
async for token in llm.generate_stream(messages):
print(token, end="", flush=True)
```
### make\_llm\_call()
Low-level method for API calls:
```python theme={null}
async def make_llm_call(
self,
messages: List[Message],
json_schema: Optional[dict],
stream: bool = False
) -> Union[Dict[str, Any], str]:
"""
Make an asynchronous call to the LLM endpoint.
Parameters
----------
messages : List[Message]
Conversation messages
json_schema : dict, optional
Response format schema
stream : bool
Enable streaming (not fully implemented)
Returns
-------
str
The LLM response content
"""
```
## Structured Output
Use JSON schemas for structured responses:
```python theme={null}
from pydantic import BaseModel, Field
# Define schema using Pydantic
class MovieReview(BaseModel):
title: str = Field(description="Movie title")
rating: int = Field(description="Rating 1-10")
summary: str = Field(description="Brief review")
# Create JSON schema
json_schema = {
"type": "json_schema",
"json_schema": {
"name": "movie_review",
"schema": MovieReview.model_json_schema()
}
}
# Generate structured response
response = await llm.generate_response(messages, json_schema=json_schema)
# Parse JSON response
import json
review = json.loads(response)
print(review["title"], review["rating"])
```
## Message Formatting
Messages are automatically converted to OpenAI format:
```python theme={null}
def _format_messages(self, messages: List[Message]) -> List[Dict[str, str]]:
"""Convert Message objects to OpenAI format."""
formatted = []
for msg in messages:
if isinstance(msg, (UserMessage, SystemMessage, ToolMessage)):
formatted.append({"role": msg.role, "content": msg.content or ""})
elif isinstance(msg, AIMessage):
formatted.append({"role": "assistant", "content": msg.content or ""})
return formatted
```
## Integration with Agent
The Agent module uses ArkModelLink through `call_llm()`:
```python theme={null}
class Agent:
def __init__(self, ..., llm: ArkModelLink, ...):
self.llm = llm
async def call_llm(self, context=None, json_schema=None):
"""Agent's interface with chat model."""
llm_response = await self.llm.generate_response(context, json_schema)
return AIMessage(content=llm_response)
```
## AsyncOpenAI Client
The module uses AsyncOpenAI internally:
```python theme={null}
@property
def client(self) -> AsyncOpenAI:
"""Returns the configured AsyncOpenAI client."""
return AsyncOpenAI(
base_url=self.base_url,
api_key="-", # Placeholder for local deployment
)
```
## Configuration via YAML
Configure the LLM endpoint in `config_module/config.yaml`:
```yaml theme={null}
llm:
base_url: "http://localhost:30000/v1"
```
Access in code:
```python theme={null}
from config_module.loader import config
llm = ArkModelLink(base_url=config.get("llm.base_url"))
```
## Error Handling
```python theme={null}
async def make_llm_call(self, messages, json_schema, stream=False):
try:
chat_completion = await self.client.chat.completions.create(
model=self.model_name,
messages=openai_messages_payload,
max_tokens=self.max_tokens,
temperature=self.temperature,
response_format=json_schema,
)
return chat_completion.choices[0].message.content
except Exception as e:
print(f"Error during async LLM call: {e}")
return f"Error: An error occurred during async LLM call: {e}"
```
## Streaming Implementation
```python theme={null}
async def generate_stream(self, messages: List[Message]) -> AsyncIterator[str]:
"""Stream tokens as they're generated."""
openai_messages = self._format_messages(messages)
try:
stream = await self.client.chat.completions.create(
model=self.model_name,
messages=openai_messages,
max_tokens=self.max_tokens,
temperature=self.temperature,
stream=True,
)
async for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
except Exception as e:
print(f"Error during streaming: {e}")
yield f"Error: {e}"
```
## SGLANG Server
The model module is designed to work with SGLANG:
```bash theme={null}
# Start SGLANG server (port 30000)
bash model_module/run.sh
```
This runs Qwen 2.5-7B-Instruct with an OpenAI-compatible API.
### Verify Server
```bash theme={null}
curl http://localhost:30000/v1/models
```
## Testing
Basic test example:
```python theme={null}
# model_module/tests_arkmodel.py
import asyncio
from model_module.ArkModelNew import ArkModelLink, UserMessage, SystemMessage
async def test():
llm = ArkModelLink(base_url="http://localhost:30000/v1")
messages = [
SystemMessage(content="You are helpful"),
UserMessage(content="Say hello")
]
response = await llm.generate_response(messages, None)
print(response)
asyncio.run(test())
```
## Best Practices
1. **Use async/await**: All LLM calls should be awaited
2. **Set appropriate timeouts**: Prevent hanging on slow responses
3. **Handle errors gracefully**: Catch exceptions and provide fallbacks
4. **Use streaming for UX**: Better user experience for long responses
5. **Validate schemas**: Test JSON schemas before production use
## Troubleshooting
Ensure SGLANG server is running:
```bash theme={null}
curl http://localhost:30000/v1/models
```
Check GPU utilization and model loading:
```bash theme={null}
nvidia-smi
```
Verify your schema matches expected output:
```python theme={null}
print(json_schema)
print(response)
```
## Next Steps
Learn about agent orchestration
Explore memory integration
Understand tool calling
Set up your environment
# State Module Overview
Source: https://arkos.mintlify.app/modules/state/overview
YAML-configurable state machine for conversation flow management
# State Module Overview
The State Module implements a YAML-configurable state machine that manages conversation flow and agent behavior in ARKOS.
## Core Concepts
The State Module provides a declarative way to define conversation flows using YAML configuration files, with automatic transition handling based on context.
### What is a State?
A state represents a distinct phase in the conversation or agent processing pipeline. Each state:
* Has a unique name
* Defines specific behavior when active
* Specifies possible transitions to other states
* Can be marked as terminal (ending the flow)
## Architecture
```mermaid theme={null}
graph TB
subgraph "State Module"
SH[StateHandler]
SH --> SR[State Registry]
SH --> YAML[YAML Config]
end
subgraph "State Types"
SR --> AS[AgentState Enum]
AS --> GU[GREET_USER]
AS --> FP[FETCH_PRODUCT]
AS --> SR2[SUMMARIZE_RESULT]
AS --> DONE[DONE]
end
subgraph "State Base"
S[State Class]
S --> RUN[run method]
S --> CHECK[check_transition_ready]
end
```
## State Class
The base `State` class that all states extend:
```python theme={null}
from state_module.state import State
class State:
def __init__(self, name: str, config: Dict[str, Any]):
self.name = name
self.is_terminal: bool = False
self.transition = config.get("transition", {})
def check_transition_ready(self, context: Dict[str, Any]) -> bool:
"""
Override to define when this state can transition.
"""
raise NotImplementedError
def run(self, context: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""
Override to define state behavior.
"""
raise NotImplementedError
```
## AgentState Enum
Predefined states available in the system:
```python theme={null}
from enum import Enum
class AgentState(Enum):
"""Central registry of all possible agent states."""
GREET_USER = "greet_user"
FETCH_PRODUCT = "fetch_product"
SUMMARIZE_RESULT = "summarize_result"
DONE = "done"
```
## StateHandler
Manages the state machine lifecycle and transitions:
```python theme={null}
from state_module.state_handler import StateHandler
# Initialize with YAML configuration
handler = StateHandler(yaml_path="state_module/state_graph.yaml")
# Get initial state
initial_state = handler.get_initial_state()
# Get a specific state
state = handler.get_state("agent_reply")
# Get possible transitions from current state
transitions = handler.get_transitions(current_state, messages)
# Returns: {"tt": ["state1", "state2"], "td": ["description1", "description2"]}
```
## YAML Configuration
### Basic State Graph
```yaml theme={null}
# state_module/state_graph.yaml
ask_user:
description: "Waiting for user input"
type: user
transition:
next: [agent_reply]
agent_reply:
description: "AI reasoning and response generation"
type: agent
transition:
next: [ask_user, use_tool]
use_tool:
description: "Execute a specific tool"
type: tool
transition:
next: [agent_reply, ask_user]
```
### Transition Format
Transitions use a list of names corresponding to the next state
```yaml theme={null}
transition:
next: [agent_reply, ask_user, use_tool]
```
## State Execution Flow
```mermaid theme={null}
sequenceDiagram
participant Agent
participant StateHandler
participant State
participant LLM
Agent->>StateHandler: get_initial_state()
StateHandler-->>Agent: current_state
loop Until Terminal State
Agent->>State: run(context, agent)
State-->>Agent: update (AIMessage/ToolMessage)
Agent->>State: check_transition_ready(messages)
State-->>Agent: ready (bool)
alt Single Transition
Agent->>StateHandler: get_state(next_name)
else Multiple Transitions
Agent->>LLM: choose_transition(options)
LLM-->>Agent: selected_state
Agent->>StateHandler: get_state(selected)
end
end
```
## Integration with Agent
The Agent uses StateHandler for flow control:
```python theme={null}
class Agent:
def __init__(self, ..., flow: StateHandler, ...):
self.flow = flow
self.current_state = self.flow.get_initial_state()
async def step(self, messages, user_id=None):
while not self.current_state.is_terminal:
# Run current state
context = self.get_context()
update = await self.current_state.run(context, self)
if update:
self.add_context([update])
# Check for transition
if self.current_state.check_transition_ready(messages):
transition_dict = self.flow.get_transitions(
self.current_state, messages
)
transition_names = transition_dict["tt"]
if len(transition_names) == 1:
next_state_name = transition_names[0]
else:
# LLM chooses between options
next_state_name = await self.choose_transition(
transition_dict, messages
)
self.current_state = self.flow.get_state(next_state_name)
```
## Transition Selection
When multiple transitions are possible, the LLM chooses:
```python theme={null}
async def choose_transition(self, transitions_dict, messages):
"""Use LLM to choose the best next state."""
# Create tuples of (state_name, description)
transition_tuples = list(zip(
transitions_dict["tt"],
transitions_dict["td"]
))
prompt = f"""Given the conversation context and these options:
{transition_tuples}
Choose the most appropriate next state."""
# Create Pydantic model for structured output
NextStates = self.create_next_state_class(transition_tuples)
# Get LLM decision
output = await self.call_llm(
context=[SystemMessage(content=prompt)] + messages,
json_schema=NextStates.model_json_schema()
)
return json.loads(output.content)["next_state"]
```
## Custom State Types
### Creating a Custom State
### (NOTE: all states must be registered with @register\_state and follow state\_\[name].py convention)
```python theme={null}
from state_module.state import State
from model_module.ArkModelNew import AIMessage
@register_state
class GreetingState(State):
def __init__(self, name: str, config: dict):
super().__init__(name, config)
self.greeting = config.get("greeting", "Hello!")
def check_transition_ready(self, context) -> bool:
# Always ready to transition after greeting
return True
def run(self, context, agent=None):
return AIMessage(content=self.greeting)
```
### Tool State Example
```python theme={null}
from model_module.ArkModelNew import SystemMessage
from tool_module.tool_call import AuthRequiredError
from state_module.state import State
from state_module.state_registry import register_state
@register_state
class StateTool(State):
type = "tool"
def __init__(self, name: str, config: dict):
super().__init__(name, config)
self.is_terminal = False
def check_transition_ready(self, context):
return True
async def choose_tool(self, context, agent):
"""
Chooses tool to use based on the context and server
"""
prompt = "Based on the user request, choose the best tool. Return JSON with tool_name field."
instructions = context + [SystemMessage(content=prompt)]
# Get available tools and build schema
tool_option_class = await agent.create_tool_option_class()
json_schema = {
"type": "json_schema",
"json_schema": {
"name": "tool_choice",
"schema": tool_option_class.model_json_schema(),
},
}
# Get tool choice from LLM
output = await agent.call_llm(instructions, json_schema)
parsed = json.loads(output.content)
tool_name = parsed["tool_name"]
# Get tool spec
server_name = agent.tool_manager._tool_registry[tool_name]
all_tools = await agent.tool_manager.list_all_tools()
tool_spec = all_tools[server_name][tool_name]
# Get tool arguments
args_prompt = f"Fill in arguments for tool '{tool_name}'. Return JSON matching the schema."
args_context = context + [SystemMessage(content=args_prompt)]
args_schema = {
"type": "json_schema",
"json_schema": {
"name": "tool_args",
"schema": tool_spec.get("inputSchema", {"type": "object", "properties": {}}),
},
}
args_output = await agent.call_llm(args_context, args_schema)
tool_args = json.loads(args_output.content)
return {"tool_name": tool_name, "tool_args": tool_args}
async def execute_tool(self, tool_call, agent):
"""
Parses and fills args for chosen tool for tool call execution
"""
tool_name = tool_call["tool_name"]
tool_args = tool_call["tool_args"]
tool_result = await agent.tool_manager.call_tool(
tool_name=tool_name,
arguments=tool_args,
user_id=agent.current_user_id,
)
return tool_result
async def run(self, context, agent=None):
try:
tool_arg_dict = await self.choose_tool(context=context, agent=agent)
tool_result = await self.execute_tool(tool_call=tool_arg_dict, agent=agent)
return SystemMessage(content=str(tool_result))
except AuthRequiredError as e:
# Return friendly message with connect link
return SystemMessage(
content=f"To complete this request, please connect your {e.service_info.get('name', e.service)}:\n\n"
f" {e.connect_url}\n\n"
f"After connecting, try your request again."
)
```
## State Types Reference
| Type | Purpose | Terminal |
| ---------- | --------------------- | -------- |
| `agent` | Generate AI response | No |
| `user` | Wait for user input | No |
| `tool` | Execute tool call | No |
| `decision` | Make routing decision | No |
| `terminal` | End conversation | Yes |
## Best Practices
### State Design Guidelines
1. **Single Responsibility**: Each state should have one clear purpose
2. **Clear Transitions**: Define unambiguous transition conditions
3. **Terminal States**: Ensure all paths eventually reach a terminal
4. **Descriptive Names**: Use clear, action-oriented state names
### YAML Configuration Tips
```yaml theme={null}
# Good: Clear state names and descriptions
states:
gather_requirements:
type: user
transition:
next: [validate_input, ask_clarification]
# Avoid: Ambiguous transitions
states:
process:
type: unknown
transition:
next: [next1, next2]
```
## Debugging States
### Logging State Transitions
```python theme={null}
# Agent logs state changes
print(f"agent.py CURR STATE: {self.current_state}")
print(f"agent.py IS TERMINAL?: {self.current_state.is_terminal}")
```
### Common Issues
Check that `check_transition_ready()` returns `True`:
```python theme={null}
ready = state.check_transition_ready(messages)
print(f"Transition ready: {ready}")
```
Improve transition descriptions for better LLM selection:
```yaml theme={null}
transition:
next: [help, search]
t
```
Ensure there's always a path to a terminal state and MAX\_ITER limit is respected.
## Example: Complete State Graph
```yaml theme={null}
initial: agent_reply
states:
# === Core States ===
ask_user:
description: "Waiting for user input"
type: user
transition:
next: [agent_reply]
agent_reply:
description: "AI reasoning and response generation"
type: agent
transition:
next: [agent_reply, ask_user, use_tool, autonomous_plan, learn_skill]
use_tool:
description: "Execute a specific tool"
type: tool
transition:
next: [agent_reply, ask_user]
# === Autonomous Planning Loop (OpenClaw-style) ===
autonomous_plan:
description: "Create multi-step execution plan for complex tasks"
type: planner
transition:
next: [execute_step, ask_user] # ask_user if permission needed
execute_step:
description: "Execute current step in the plan"
type: executor
transition:
next: [execute_step, verify_progress] # loop or verify
verify_progress:
description: "Check if goal achieved or continue execution"
type: verifier
transition:
next: [execute_step, autonomous_plan, agent_reply, ask_user] # continue, replan, or finish
# === Self-Improving Skills ===
learn_skill:
description: "Analyze and potentially create new capabilities"
type: skill_builder
transition:
next: [agent_reply, autonomous_plan] # use skill or plan with it
```
## Next Steps
Deep dive into state implementation
Learn about agent orchestration
Explore memory integration
Understand LLM integration
# Tool Module Overview
Source: https://arkos.mintlify.app/modules/tool/overview
MCP-compatible tool system for extensible agent capabilities
# Tool Module Overview
The ARKOS Tool Module implements the Model Context Protocol (MCP) for connecting to external tool servers, enabling agents to interact with calendars, search engines, and other services.
## What is MCP?
The Model Context Protocol (MCP) is a standard for AI systems to communicate with external tools via JSON-RPC 2.0. ARKOS supports both stdio and HTTP transports.
ARKOS tools follow the MCP specification, allowing seamless integration with any MCP-compatible server.
## Architecture
```mermaid theme={null}
graph TB
subgraph "Tool Module"
TM[MCPToolManager]
TM --> C1[MCPClient 1]
TM --> C2[MCPClient 2]
TM --> CN[MCPClient N]
end
subgraph "Transports"
C1 --> STDIO[StdioTransport]
C2 --> HTTP[HTTPTransport]
end
subgraph "MCP Servers"
STDIO --> GC[Google Calendar]
HTTP --> BS[Brave Search]
end
subgraph "Per-User Auth"
TM --> TS[UserTokenStore]
TS --> DB[(PostgreSQL)]
end
```
## Core Components
### MCPClient
Manages a single MCP server connection:
```python theme={null}
from tool_module.tool_call import MCPClient, MCPServerConfig
from tool_module.transports import StdioTransport
# Create config
config = MCPServerConfig(
name="google-calendar",
transport="stdio",
command="npx",
args=["-y", "@anthropic/google-calendar-mcp"]
)
# Create transport
transport = StdioTransport(
command=config.command,
args=config.args,
env=config.env
)
# Create and start client
client = MCPClient(config, transport)
await client.start()
# List available tools
tools = await client.list_tools()
# Call a tool
result = await client.call_tool(
name="create_event",
arguments={"title": "Meeting", "start": "2024-01-26T14:00:00Z"}
)
```
### MCPToolManager
Coordinates multiple MCP server connections:
```python theme={null}
from tool_module.tool_call import MCPToolManager
from tool_module.token_store import UserTokenStore
# Configuration from YAML
mcp_config = {
"google-calendar": {
"transport": "stdio",
"command": "npx",
"args": ["-y", "@anthropic/google-calendar-mcp"]
},
"brave-search": {
"transport": "http",
"url": "http://localhost:8080"
}
}
# Initialize with token store for per-user auth
token_store = UserTokenStore(db_url)
tool_manager = MCPToolManager(mcp_config, token_store=token_store)
# Initialize all servers
await tool_manager.initialize_servers()
# List all available tools
all_tools = await tool_manager.list_all_tools()
# Call a tool
result = await tool_manager.call_tool(
tool_name="search",
arguments={"query": "weather today"},
user_id="alice" # For per-user auth
)
```
## Transport Types
### Stdio Transport
Launches MCP servers as subprocesses:
```python theme={null}
from tool_module.transports import StdioTransport
transport = StdioTransport(
command="npx",
args=["-y", "@anthropic/google-calendar-mcp"],
env={"GOOGLE_TOKEN_PATH": "/path/to/token.json"}
)
await transport.connect()
response = await transport.send_request("tools/list", {})
await transport.close()
```
### HTTP Transport
Connects to MCP servers over HTTP:
```python theme={null}
from tool_module.transports import HTTPTransport
transport = HTTPTransport(
url="http://localhost:8080",
auth_config={"api_key": "..."}
)
await transport.connect()
response = await transport.send_request("tools/call", {
"name": "search",
"arguments": {"query": "test"}
})
```
## MCP Protocol
### Initialization Handshake
```python theme={null}
# 1. Send initialize request
init_response = await transport.send_request(
"initialize",
{
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "arkos", "version": "1.0.0"}
}
)
# 2. Send initialized notification
await transport.send_notification("notifications/initialized", {})
```
### Tool Discovery
```python theme={null}
# List available tools
response = await transport.send_request("tools/list", {})
tools = response["result"]["tools"]
# Each tool has:
# - name: Tool identifier
# - description: What the tool does
# - inputSchema: JSON Schema for parameters
```
### Tool Execution
```python theme={null}
# Call a tool
response = await transport.send_request(
"tools/call",
{
"name": "create_event",
"arguments": {
"title": "Team Meeting",
"start": "2024-01-26T14:00:00Z",
"duration": 60
}
}
)
result = response["result"]
```
## Per-User Authentication
Some tools (like Google Calendar) require per-user OAuth authentication.
### Services Requiring Auth
```python theme={null}
PER_USER_SERVICES = {
"google-calendar": {
"name": "Google Calendar",
"auth_path": "/auth/google/login",
"scopes": ["calendar"],
}
}
```
### AuthRequiredError
When a user hasn't authenticated:
```python theme={null}
from tool_module.tool_call import AuthRequiredError
try:
result = await tool_manager.call_tool(
"create_event",
{"title": "Meeting"},
user_id="alice"
)
except AuthRequiredError as e:
# Redirect user to authenticate
print(f"Please connect: {e.connect_url}")
# e.to_dict() returns structured error for API responses
```
### UserTokenStore
Manages OAuth tokens per user:
```python theme={null}
from tool_module.token_store import UserTokenStore
token_store = UserTokenStore(db_url)
# Check if user has token
if token_store.has_token(user_id, "google-calendar"):
# User is authenticated
pass
# Write token to file for MCP server
token_store.write_token_file(user_id, "google-calendar", "/path/to/token.json")
```
### User Service Status
Check which services a user has connected:
```python theme={null}
status = tool_manager.get_user_service_status("alice")
# {
# "google-calendar": {
# "connected": True,
# "name": "Google Calendar",
# "connect_url": "/auth/google/login?user_id=alice"
# }
# }
# Get missing services
missing = tool_manager.get_missing_services("alice")
# [{"service": "google-calendar", "name": "Google Calendar", ...}]
```
## Configuration
### YAML Configuration
```yaml theme={null}
# config_module/config.yaml
mcp_servers:
google-calendar:
transport: stdio
command: npx
args: ["-y", "@anthropic/google-calendar-mcp"]
env:
GOOGLE_CALENDAR_MCP_TOKEN_PATH: "${GOOGLE_CALENDAR_MCP_TOKEN_PATH}"
brave-search:
transport: http
url: "http://localhost:8080"
auth:
api_key: "${BRAVE_API_KEY}"
```
### Environment Variables
```bash theme={null}
# Google Calendar OAuth
GOOGLE_OAUTH_CREDENTIALS=/path/to/credentials.json
GOOGLE_CALENDAR_MCP_TOKEN_PATH=/path/to/token.json
# Brave Search API
BRAVE_API_KEY=your-api-key
```
## Integration with Agent
The Agent receives tools on startup:
```python theme={null}
# In base_module/app.py
@app.on_event("startup")
async def startup():
if tool_manager:
await tool_manager.initialize_servers()
# Cache tools on agent
agent.available_tools = await tool_manager.list_all_tools()
# Build system prompt with tool descriptions
tool_prompt = format_tools_for_system_prompt(agent.available_tools)
agent.system_prompt = base_prompt + "\n\n" + tool_prompt
```
### System Prompt Format
```python theme={null}
def format_tools_for_system_prompt(tools: dict) -> str:
lines = [
"You have access to the following tools.",
"Use them when appropriate. Only call tools that are listed below.",
""
]
for name, tool in tools.items():
lines.append(f"Tool name: {name}")
if tool.description:
lines.append(f"Description: {tool.description}")
if tool.input_schema:
lines.append("Input schema:")
lines.append(str(tool.input_schema))
lines.append("")
return "\n".join(lines)
```
## Error Handling
### Connection Errors
```python theme={null}
try:
await client.start()
except RuntimeError as e:
logger.error(f"Failed to start MCP server: {e}")
```
### Tool Execution Errors
```python theme={null}
try:
result = await client.call_tool(name, arguments)
except RuntimeError as e:
logger.error(f"Tool execution failed: {e}")
except ValueError as e:
logger.error(f"Unknown tool: {e}")
```
## Lifecycle Management
### Graceful Shutdown
```python theme={null}
await tool_manager.shutdown()
# This will:
# 1. Stop all shared clients
# 2. Stop all per-user clients
# 3. Clear the tool registry
```
## Adding New MCP Servers
1. **Install the MCP server package**:
```bash theme={null}
npm install -g @your-org/your-mcp-server
```
2. **Add configuration**:
```yaml theme={null}
mcp_servers:
your-server:
transport: stdio
command: npx
args: ["-y", "@your-org/your-mcp-server"]
```
3. **Restart ARKOS** - the server will be auto-discovered
## Best Practices
1. **Use stdio for local tools**: Better security and isolation
2. **Use HTTP for shared services**: Better for multi-user deployments
3. **Handle auth errors gracefully**: Show users how to connect services
4. **Set reasonable timeouts**: Prevent hanging on slow tools
5. **Log tool calls**: Helps with debugging and monitoring
## Troubleshooting
Check that the command exists and has correct permissions:
```bash theme={null}
which npx
npx -y @anthropic/google-calendar-mcp --help
```
Verify the tool is registered:
```python theme={null}
tools = await tool_manager.list_all_tools()
print(tools.keys())
```
Check user token status:
```python theme={null}
status = tool_manager.get_user_service_status(user_id)
print(status)
```
## Next Steps
Learn about agent orchestration
Explore the API server
Understand memory systems
Set up your environment
# Quick Start
Source: https://arkos.mintlify.app/quickstart
Get up and running with ARKOS in minutes
# Quick Start Guide
This guide will help you set up ARKOS and create your first intelligent agent with persistent memory.
## Prerequisites
Before you begin, ensure you have:
* Python 3.8 or higher
* Git
* PostgreSQL database (Supabase recommended)
* 8GB+ RAM recommended
* GPU optional but recommended for local models
## Installation
### 1. Clone the Repository
```bash theme={null}
git clone https://github.com/SGIARK/arkos.git
cd arkos
```
### 2. Install Dependencies
```bash theme={null}
pip install -r requirements.txt
```
### 3. Configure Your Environment
Create a `.env` file in the project root:
```bash theme={null}
cp .env.example .env
```
Edit `.env` with your configuration:
```env theme={null}
# Database Connection (REQUIRED)
# Format: postgresql://user:password@host:port/database
DB_URL=postgresql://postgres:your-password@localhost:54322/postgres
# Hugging Face Token (OPTIONAL - for gated models)
HF_TOKEN=
# MCP Server Credentials (OPTIONAL - for tool integrations)
GOOGLE_OAUTH_CREDENTIALS=
GOOGLE_CALENDAR_MCP_TOKEN_PATH=
BRAVE_API_KEY=
# OpenAI API Key (REQUIRED by mem0, but can be placeholder)
OPENAI_API_KEY=sk-placeholder
```
The `DB_URL` environment variable is **required**. ARKOS uses PostgreSQL for storing conversation context and Supabase for vector memory.
## Starting the Inference Engine
### Check if LLM Server is Already Running
Since ARKOS is often deployed on shared servers, check if the LLM server is already running:
```bash theme={null}
# Check if port 30000 is in use
lsof -i :30000
# Or verify it's responding
curl http://localhost:30000/v1/models
```
If you see output, the LLM server is already running - you can skip starting it.
### Starting the LLM Server (if not running)
The project uses SGLang to run the Qwen 2.5-7B-Instruct model:
```bash theme={null}
bash model_module/run.sh
```
This starts the SGLang server on **port 30000** using Docker and GPU. Wait for "server started" messages.
### Starting the Embedding Server (if not running)
The project uses Huggingface-TEI to run the Qwen 2 1.5B-Instruct model
```bash theme={null}
bash model_module/run_tei.sh
```
This starts the SGLang server on **port 4444** using Docker and GPU. Wait for "server started" messages.
## Running Your First Agent
### 1. Start the API Server
```bash theme={null}
python base_module/app.py
```
This starts the FastAPI server on **port 1111**, providing the `/v1/chat/completions` endpoint.
### 2. Run the Test Interface
In another terminal:
```bash theme={null}
python base_module/main_interface.py
```
This provides an interactive CLI to test the agent. Type your messages and press Enter. Type `exit` or `quit` to stop.
## Basic Usage Examples
### Using the OpenAI-Compatible API
```python theme={null}
from openai import OpenAI
client = OpenAI(
base_url="http://localhost:1111/v1",
api_key="not-needed" # Local deployment
)
response = client.chat.completions.create(
model="ark-agent",
messages=[
{"role": "user", "content": "Hello! What can you help me with?"}
]
)
print(response.choices[0].message.content)
```
### Streaming Responses
```python theme={null}
stream = client.chat.completions.create(
model="ark-agent",
messages=[{"role": "user", "content": "Tell me about ARKOS"}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="")
```
### Using Memory Directly
```python theme={null}
from memory_module.memory import Memory
from model_module.ArkModelNew import UserMessage, AIMessage, SystemMessage
# Initialize memory system
memory = Memory(
user_id="alice",
session_id=None, # Auto-generates session ID
db_url="postgresql://...",
use_long_term=True # Enable Mem0 vector memory
)
# Store a message
memory.add_memory(UserMessage(content="My favorite color is blue"))
# Retrieve recent context
context = memory.retrieve_short_memory(turns=5)
# Retrieve relevant long-term memories
long_term = memory.retrieve_long_memory(context=context)
```
## Configuration Files
### Main Configuration (`config_module/config.yaml`)
```yaml theme={null}
app:
host: "0.0.0.0"
port: 1111
reload: false
system_prompt: "You are a helpful AI assistant."
llm:
base_url: "http://localhost:30000/v1"
database:
url: "${DB_URL}"
memory:
user_id: "default_user"
use_long_term: false # Set to true to enable Mem0
state:
graph_path: "state_module/state_graph.yaml"
# MCP Server Configuration (optional)
mcp_servers:
google-calendar:
transport: stdio
command: npx
args: ["-y", "@anthropic/google-calendar-mcp"]
```
### State Graph Configuration (`state_module/state_graph.yaml`)
```yaml theme={null}
initial: agent_reply
states:
ask_user:
description: "state used for input from user"
type: user
transition:
next: [agent_reply]
agent_reply:
description: "state used for your reasoning"
type: agent
transition:
next: [ask_user, use_tool]
use_tool:
description: "state used for tool use"
type: tool
transition:
next: [agent_reply]
```
## Testing Your Setup
### Health Check
```bash theme={null}
curl http://localhost:1111/health
```
Expected response:
```json theme={null}
{"status": "ok", "llm_server": "running", "port": 1111}
```
### Test Chat Completion
```bash theme={null}
curl -X POST http://localhost:1111/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{
"model": "ark-agent",
"messages": [{"role": "user", "content": "Hello!"}]
}'
```
## Common Issues and Solutions
Ensure your `DB_URL` is correctly set in `.env`:
```bash theme={null}
# Check if PostgreSQL is accessible
psql $DB_URL -c "SELECT 1"
```
Make sure the `conversation_context` table exists in your database.
Check if the SGLang server is running on port 30000:
```bash theme={null}
curl http://localhost:30000/v1/models
```
If not running, start it with `bash model_module/run.sh`
Ensure PyTorch is installed with CUDA support:
```bash theme={null}
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118
```
Find and kill the process using the port:
```bash theme={null}
lsof -i :1111
kill -9
```
## Next Steps
Understand the system architecture
Deep dive into each module
Set up your development environment
Explore the API endpoints
## Getting Help
* **GitHub Issues**: [Report bugs or request features](https://github.com/SGIARK/arkos/issues)
* **Documentation**: You're already here!
* **Community**: Join our discussions on GitHub