Skip to main content

Forseti 461 Architecture

Technical documentation for the Forseti 461 agent implementation.

Agents Directory Structure

Architecture

┌─────────────────────────────────────────────────────────────────────────────┐
│ FORSETI 461 AGENT │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ app/agents/forseti/ │
│ ├── __init__.py Module exports │
│ ├── agent.py ForsetiAgent class │
│ ├── models.py Pydantic models (ValidationResult, etc.) │
│ ├── prompts.py Re-exports from app/prompts/ │
│ └── features/ │
│ ├── __init__.py Feature exports │
│ ├── base.py FeatureBase abstract class │
│ ├── charter_validation.py │
│ ├── category_classification.py │
│ ├── wording_correction.py │
│ ├── anonymization.py LLM-based PII anonymization │
│ └── translation.py FR→EN translation (available, not integrated) │
│ │
│ app/prompts/ │
│ ├── local/forseti.py Python prompts (fallback) │
│ └── local/forseti_charter.json Opik-synced prompts (chat format) │
│ │
└─────────────────────────────────────────────────────────────────────────────┘

Core Concepts

Prompts (System+User) matching concept prompts

Feature executions

Features are registered with agents and executed independently: features details

Data Flow

                                  ┌─────────────────┐
│ REST API │
│ /api/v1/validate│
└────────┬────────┘


┌─────────────────────────────────────────────────────────────┐
│ ForsetiAgent │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ Persona │ │ Features │ │ Provider │ │
│ │ Prompt │ │ │ │ (Gemini/Claude/...)│ │
│ └─────────────┘ │ ┌─────────┐ │ └─────────────────────┘ │
│ │ │Charter │ │ │
│ │ │Validation│───────────┐ │
│ │ └─────────┘ │ │ │
│ │ ┌─────────┐ │ ▼ │
│ │ │Category │ │ ┌──────────┐ │
│ │ │Classify │─────▶│ LLM │ │
│ │ └─────────┘ │ │ Provider │ │
│ │ ┌─────────┐ │ └──────────┘ │
│ │ │Wording │ │ │ │
│ │ │Correct │────────────┘ │
│ │ └─────────┘ │ │
│ │ ┌─────────┐ │ │
│ │ │Anonymize│ │ (standalone, not auto- │
│ │ │ (PII) │ │ registered) │
│ │ └─────────┘ │ │
│ └─────────────┘ │
└─────────────────────────────────────────────────────────────┘


┌─────────────────┐
│ AgentTracer │
│ (Opik) │
└─────────────────┘

Pydantic Models

Core Models

class ValidationResult(BaseModel):
is_valid: bool
violations: list[str]
encouraged_aspects: list[str]
reasoning: str
confidence: float # 0.0 - 1.0

class ClassificationResult(BaseModel):
category: str # One of CATEGORIES
reasoning: str
confidence: float

class FullValidationResult(BaseModel):
is_valid: bool
category: str
original_category: str | None
violations: list[str]
encouraged_aspects: list[str]
reasoning: str
confidence: float

Anonymization Models

class EntityType(str, Enum):
PERSON = "PERSONNE"
EMAIL = "EMAIL"
PHONE = "TELEPHONE"
ADDRESS = "ADRESSE"

class DetectedEntity(BaseModel):
original: str
placeholder: str
entity_type: EntityType

class AnonymizationResult(BaseModel):
anonymized_text: str
entities: list[DetectedEntity]
entity_mapping: dict[str, str]
keywords_extracted: list[str]
reasoning: str

Batch Models

class BatchItem(BaseModel):
id: str
title: str
body: str
category: str | None

class BatchResult(BaseModel):
id: str
is_valid: bool
violations: list[str]
encouraged_aspects: list[str]
category: str
reasoning: str
confidence: float

Configuration

Environment Variables

# Provider selection
DEFAULT_PROVIDER=gemini # gemini | claude | mistral | ollama

# Gemini
GOOGLE_API_KEY=...
GEMINI_MODEL=gemini-1.5-flash
GEMINI_RATE_LIMIT=12.0 # seconds between calls

# Claude
ANTHROPIC_API_KEY=...
CLAUDE_MODEL=claude-3-haiku-20240307

# Mistral
MISTRAL_API_KEY=...
MISTRAL_MODEL=mistral-small-latest

# Ollama
OLLAMA_HOST=http://localhost:11434
OLLAMA_MODEL=mistral:latest

# Tracing
OPIK_API_KEY=...
OPIK_WORKSPACE=...
OPIK_PROJECT=forseti

ProviderConfig

Configuration is managed via pydantic-settings:

from app.providers.config import get_config

config = get_config()
print(config.default_provider) # "gemini"
print(config.gemini_model) # "gemini-1.5-flash"

Tracing

Automatic Feature Tracing

Use the @trace_feature decorator:

from app.agents.tracing import trace_feature

class MyFeature(FeatureBase):
@trace_feature("my_feature")
async def execute(self, provider, system_prompt, **kwargs):
# Automatically traced
return result

Manual Tracing

from app.agents.tracing import get_tracer

tracer = get_tracer()
tracer.trace(
name="custom_operation",
input={"title": "..."},
output={"is_valid": True},
metadata={"confidence": 0.95},
tags=["forseti", "custom"],
)

Error Handling

Features fail gracefully with safe defaults:

# Charter validation: fail open (allow content through)
except Exception as e:
return ValidationResult(
is_valid=True, # Fail open
violations=[],
reasoning=f"Validation error: {e}",
confidence=0.5,
)

# Category classification: use default category
except Exception as e:
return ClassificationResult(
category=current_category or CATEGORIES[0], # Default to first
reasoning=f"Classification error: {e}",
confidence=0.5,
)

Testing

# Run all tests
poetry run pytest tests/test_providers/ tests/test_agents/

# Test specific provider
poetry run pytest tests/test_providers/test_gemini.py

# Test Forseti agent
poetry run pytest tests/test_agents/test_forseti.py