Aller au contenu principal

Application Fundamentals

Shared architectural patterns for production Python applications across Locki ecosystem

This documentation applies to: OCapistaine, Vaettir, and other FastAPI/async applications

Quick Start

  1. Building a service? Follow Application Lifecycle
  2. Writing scripts? Use Scripts Standardization
  3. Building agents? Study Agents Framework
  4. Deploying API? Read Uvicorn & FastAPI
  5. Need observability? Check Logging Integration
  6. Orchestrating services? See Process Orchestration

Overview

This section documents the common patterns and architectural decisions that should be consistent across all applications in the Locki ecosystem. These patterns ensure:

  • Consistency: Same patterns used everywhere
  • Scalability: Production-ready from day one
  • Maintainability: Clear conventions reduce cognitive load
  • Cross-repo reuse: Scripts, agents, logging align
  • Developer experience: Easy onboarding to any repo

Core Sections

1. Uvicorn & FastAPI Deployment

When: You're building a REST API or webhook service

  • Why uvicorn over Flask for production
  • Async lifespan context manager (startup/shutdown)
  • Configuration: development vs. production
  • Worker scaling strategies, graceful shutdown

Applies to: OCapistaine (app/main.py), Vaettir adapters, any new Python service


2. Agents Framework

When: You're building reusable AI agent logic

  • BaseAgent abstraction pattern
  • AgentFeature composition protocol
  • Provider injection (LLMProvider with failover)
  • Feature lifecycle and testing with mock providers
BaseAgent (persona + features + provider)
├── Features (composable functionality units)
├── LLMProvider (with failover chain)
└── Opik tracing (observability)

Applies to: OCapistaine (Forseti, RAG), Vaettir workflow agents, custom agents


3. Scripts Standardization

When: You're managing a multi-service project

scripts/
├── start.sh # Service entry point
├── start_with_deps.sh # Full stack startup
├── stop.sh # Graceful shutdown
├── healthcheck.sh # Service verification
└── utils.sh # Shared bash functions

Applies to: OCapistaine (scripts/start.sh), Vaettir (scripts/docker-compose-up.sh), any repo


4. Logging Integration

When: You need observability across async services

logs/
├── agents.log # AI agent activity
├── services.log # Business logic
├── scheduler.log # Task orchestration
├── providers.log # LLM API calls
└── [domain].log # Any domain

Applies to: OCapistaine domain loggers, Vaettir N8N workflow logging, any service


5. Application Lifecycle

When: You're designing how a service starts and stops

1. Load environment (.env, secrets)
2. Initialize connections (Redis, DB, APIs)
3. Validate providers (test LLM connectivity)
4. Start scheduler (APScheduler if async)
5. Listen for requests
...
6. Shutdown signal → stop accepting → wait for tasks → cleanup → exit

Applies to: OCapistaine (Uvicorn + APScheduler), Vaettir (n8n coordination), any service


6. Process Orchestration

When: You need to coordinate multi-step workflows or multiple services

  • Workflow design patterns (sequential, event-driven, orchestrator)
  • State management, fault tolerance, retry strategies
  • Multi-service startup coordination, Docker Compose templates

Applies to: OCapistaine pipelines, Vaettir n8n orchestration, multi-stage data processing


Quick Navigation by Use Case

Use CaseStart Here
Starting a new serviceLifecycleUvicornScriptsLogging
Building an AI agentAgents FrameworkLogging
Integrating servicesOrchestrationScriptsLifecycle
TroubleshootingOrchestrationLogging

Cross-Repository Applicability

PatternOCapistaineVaettirNew Services
Uvicorn/FastAPIMain APIAdapters/microservicesAlways
Agents FrameworkForseti, RAGMCP agentsCustom logic
Scriptsscripts/start.shscripts/docker-compose-up.shTemplate in examples/
LoggingDomain loggersTask loggingAlways
LifecycleScheduler coordinationn8n startupAlways
OrchestrationUvicorn + Streamlitn8n + servicesAlways

Design Principles

1. Async-First

# Good
async def startup():
await initialize_async_resources()

# Avoid — blocks event loop
def startup():
initialize_blocking_resources()

2. Composition Over Inheritance

agent = BaseAgent(provider=my_provider)
agent.register_feature(ValidationFeature())
agent.register_feature(ClassificationFeature())

3. Dependency Injection

# Good — testable, swappable
agent = Forseti(provider=LLMProvider("claude"), logger=get_logger("agents"))

# Avoid — hard to test
agent = Forseti() # Creates provider internally

4. Graceful Degradation

provider = ProviderWithFailover(
primary="ollama",
fallback_chain=["openai", "gemini"]
)

5. Observable by Default

@OpikTracer.track
async def process_contribution(text: str):
logger.info("Processing", extra={"contribution_id": contribution_id})
result = await agent.execute(text)
logger.info("Done", extra={"status": result.status})

Implementation Checklist

When implementing a new service or pattern:

  • Startup: Follows Application Lifecycle sequence
  • Shutdown: Graceful with timeout and resource cleanup
  • Logging: Uses domain-based logging with structured context
  • Scripts: Follows Scripts Standardization format
  • Health checks: Implemented at each stage
  • Async: Using async/await throughout (if applicable)
  • Errors: Graceful handling with user-friendly messages
  • Observability: Traced with Opik or equivalent
  • Testing: Mock providers, testable features
  • Documentation: README + inline docstrings

Contributing

When adding new patterns to this section:

  1. Validate across repos: Pattern should apply to OCapistaine AND Vaettir
  2. Include examples: Concrete code from real implementations
  3. Link implementations: Reference actual files in repositories
  4. Explain trade-offs: Why this pattern vs. alternatives
  5. Update this README: Add to relevant sections above


Last Updated: 2026-03-18