✨ Features

🚀 Quick Start

Install

pip install "agentomatic[all]"

Create an Agent

agentomatic init my_agent --template basic

Build & Run

# main.py
from agentomatic import AgentPlatform

platform = AgentPlatform.from_folder("agents/")
app = platform.build()

uvicorn main:app --reload

Test

# CLI
agentomatic test my_agent

# curl
curl -X POST http://localhost:8000/api/v1/my_agent/invoke \
  -H "Content-Type: application/json" \
  -d '{"query": "Hello!"}'

🏗️ Architecture

┌─────────────────────────────────────────────────────────────┐
│                    AgentPlatform                            │
│                                                             │
│  ┌──────────┐  ┌──────────────┐  ┌───────────────────────┐ │
│  │ Registry │  │ Middleware   │  │ Storage               │ │
│  │          │  │ ├─ Auth      │  │ ├─ MemoryStore        │ │
│  │ agent_a  │  │ ├─ RateLimit │  │ ├─ SQLAlchemyStore   │ │
│  │ agent_b  │  │ ├─ Metrics   │  │ └─ YourStore(ABC)    │ │
│  │ agent_c  │  │ └─ Logging   │  │                       │ │
│  └──────────┘  └──────────────┘  └───────────────────────┘ │
│                                                             │
│  Per Agent: POST /invoke, /stream, /chat, /a2a/tasks ...   │
└─────────────────────────────────────────────────────────────┘

📂 Agent Structure

Only agent.py is required. Everything else is optional overrides:

agents/my_agent/
├── __init__.py      ← Optional: Python package init
├── agent.py         ← REQUIRED: Contains your BaseGraphAgent subclass
├── config.py        ← Optional: Pydantic config
├── schemas.py       ← Optional: custom request/response models
├── tools.py         ← Optional: LangChain tools
├── api.py           ← Optional: custom router (REPLACES auto-gen)
├── prompts.json     ← Optional: versioned prompt templates
├── langgraph.json   ← Optional: LangGraph Studio config
├── .env.example     ← Optional: environment variables
└── README.md        ← Optional: agent documentation

📦 Templates

agentomatic init my_agent --template <template>

🖥️ CLI

⚡ Agentomatic — Drop agents, not code

  init <name>      Scaffold an agent package from a template
  new <name>       Scaffold a complete project (alias for init --project)
  add              Add a connection or ingestion component to a project
  run              Start the platform server
  run --studio     Start with Agentomatic Studio visual debugger 🎨
  run --with-ui    Start with Chainlit chat interface 💬
  demo             Launch demo platform with Studio (no setup needed)
  list             List discovered agents (Rich table)
  test <name>      Interactive terminal testing
  inspect <name>   Show agent structure + config
  doctor           Environment health check
  optimize <name>  Run prompt optimization
  ui               Launch Chainlit debug UI standalone
  pipeline         Pipeline management commands
  stack            Multi-environment stack management
  deploy           Generate Docker deployment files
  agents-guide     Emit an Agentomatic primer for coding agents

🧬 Class-Based Agents (NEW)

Define agents as Python classes with built-in graph wiring and ML lifecycle:

from dataclasses import dataclass, field
from agentomatic import BaseGraphAgent, EarlyStopping

@dataclass
class MyState:
    query: str = ""
    output: dict = field(default_factory=dict)

class MyAgent(BaseGraphAgent[MyState]):
    agent_name = "my_agent"

    def build_graph(self):
        g = self.new_graph()
        g.add_node("process", self.process)
        g.add_node("format", self.format_out)
        g.set_entry_point("process")
        g.add_edge("process", "format")
        g.set_finish_point("format")
        return g.compile()

    def process(self, state):
        state.output = {"response": f"Hello! You asked: {state.query}"}
        return state

    def format_out(self, state):
        return state

    def input_to_state(self, data):
        return MyState(query=data.get("query", ""))

    def state_to_output(self, state):
        return state.output

# Keras-style ML workflow
agent = MyAgent()
result = agent.transform({"query": "Hello!"})

agent.compile(dataset=dataset, metrics=[accuracy], loss=my_loss)
history = agent.fit(
    dataset,
    epochs=5,
    verbose=1,                       # Keras-like per-epoch log lines
    validation_data=valset,          # adds val_* metrics
    callbacks=[EarlyStopping(monitor="val_loss", patience=2)],
)
print(history.best("val_loss", mode="min"))
report = agent.evaluate(dataset.test, metrics)
agent.save("compiled/v1")

fit() runs the prompt optimizer under the hood (via PromptFitterBridge), records per-epoch metrics/loss into a History object (also on agent.history), fires Callback hooks, and supports EarlyStopping — treating a GenAI agent like a classical trainable model. See the class-based agents guide.

🎨 Agentomatic Studio

Agentomatic ships with a built-in React-based visual studio designed for time-travel debugging, real-time node streaming, and state inspection. Works with class-based agents, LangGraph, LangChain, and any custom framework via the adapter system.

To use the studio, install the optional package dependencies and run with the --studio flag:

pip install "agentomatic[studio]"
agentomatic run --studio

The unified server will bind to http://localhost:8000 and mount the studio at http://localhost:8000/studio/ui/.

Key Studio Features:

• Live Node Streaming: Watch Server-Sent Events (SSE) transition node activity dynamically.
• Conditional Breakpoints: Right-click graph nodes to intercept flow before execution triggers.
• Time-Travel History: Rewind to any state checkpoint and replay from historical forks.
• Live State Editing: Mutate graph state payloads on the fly during a breakpoint pause.

🧠 ML Model Plugins (NEW)

Agentomatic isn't just for LLMs. Wrap classical ML models (Scikit-Learn, PyTorch, PyMC) securely with auto-generated REST endpoints:

from agentomatic.plugins import BaseMLPlugin
from pydantic import BaseModel

class IrisInput(BaseModel):
    sepal_length: float
    sepal_width: float
    petal_length: float
    petal_width: float

class IrisPlugin(BaseMLPlugin[IrisInput, dict]):
    async def load_model(self):
        # Load sklearn model from disk
        import joblib
        self.model = joblib.load("iris_model.pkl")

    async def predict(self, inputs: IrisInput) -> dict:
        prediction = self.model.predict([[
            inputs.sepal_length, inputs.sepal_width,
            inputs.petal_length, inputs.petal_width
        ]])
        return {"species": prediction[0]}

Place it in plugins/ and Agentomatic auto-discovers it alongside your AI agents!

🧵 Execution Modes & Task Board

Every resource — agents, plugins, pipelines, endpoints, and ingestors — is automatically callable in every execution mode. No extra code:

Async work is tracked by a unified task board — ideal for long-running jobs like document ingestion where the frontend polls for progress:

# Submit an async task (returns immediately with a task id)
curl -X POST http://localhost:8000/api/v1/my_agent/invoke/async \
  -H "Content-Type: application/json" -d '{"query": "long job"}'
# → {"id": "task_ab12...", "status": "queued", ...}

# Poll status + progress
curl http://localhost:8000/api/v1/tasks/task_ab12...
# → {"status": "running", "progress": {"percent": 42, "message": "chunking"}}

# Stream live progress events (SSE)
curl -N http://localhost:8000/api/v1/tasks/task_ab12.../events

# Cancel
curl -X POST http://localhost:8000/api/v1/tasks/task_ab12.../cancel

Tasks report progress (percent, message, stage), support completion webhooks (callback_url), and persist through a pluggable TaskStore (in-memory by default, or the durable SQLAlchemyTaskStore). See the Tasks guide.

🩺 Unified Status Dashboard

A single control-plane view of the whole platform's health:

open http://localhost:8000/status          # HTML dashboard
curl http://localhost:8000/api/v1/status   # JSON API

Aggregates the health of every agent, plugin, pipeline, endpoint, ingestor, the storage backend, and the task engine (queue depth, running/terminal counts) into one page — with per-resource drill-down and an overall healthy / degraded / unhealthy roll-up.

📥 Ingestion & RAG

Agentomatic is about ops, not implementation: bring your favourite libraries (PDF→markdown, loaders, splitters, embedders, vector stores) and Agentomatic packages them as a discoverable ingestor that is callable sync, async, or as a tracked task — and usable as a pipeline step.

from agentomatic.ingestion import BaseIngestor, IngestionRequest, IngestionResult

class DocsIngestor(BaseIngestor):
    ingestor_name = "docs"

    async def ingest(self, request: IngestionRequest, ctx) -> IngestionResult:
        # Reuse ANY library you like:
        text = my_pdf_lib.to_markdown(request.source)     # extract
        chunks = my_splitter.split(text)                  # chunk
        vectors = my_embedder.embed(chunks)               # embed
        await my_store.upsert(vectors)                    # persist
        await ctx.report(percent=100, message="done")     # progress → task board
        return IngestionResult(documents=1, chunks=len(chunks), upserted=len(vectors))

Drop it in ingestion/, and it's auto-discovered with its own endpoints, task support, and pipeline step. See the Ingestion guide.

🧱 Pipelines

Compose agents, plugins, endpoints, ingestors, transforms, loops, and sub-pipelines into a single graph with full control over data-passing:

# pipelines/rag_ingest.yaml
name: rag_ingest
strict_schema: true
on_error: rollback
steps:
  - ingestion: docs            # reuse your ingestor
    input: { source: "{{ input.path }}" }
    output: ingested
  - agent: summarizer
    input: { text: "{{ ingested.summary }}" }
    output: summary
    retry: { max_attempts: 3 }
    rollback: "await store.delete(ingested.id)"   # compensation
  - plugin: classifier         # call an ML plugin mid-pipeline
    input: { features: "{{ summary }}" }

Supports input/output mapping, shared context, conditionals, retries, timeouts, on_error policies (including rollback/compensation), optional input/output schema enforcement, and per-step async execution — all runnable via the same sync/async/streaming/task modes. See the Pipelines guide.

⚙️ Configuration

Agentomatic has two configuration layers that work together:

1. Pydantic BaseSettings (.env-driven)

The platform auto-loads settings via PlatformSettings (Pydantic BaseSettings):

# .env
LLM__PROVIDER=ollama
LLM__MODEL=mistral:7b
FEATURES__ENABLE_METRICS=true
AGENTOMATIC_ARTIFACT_ROOT=.local/artifacts

from agentomatic.config.settings import get_settings
settings = get_settings()
print(settings.llm.model)  # "mistral:7b"

Sections: llm, embedding, db, features, auth, rate_limit, plus app_name, log_level, artifact_root, runs_root, audit_log, chunk_size_tokens, min_quality_score.

2. Optimisation Settings (OPTIMIZER_ prefix)

OptimizerPydanticSettings for prompt-optimisation knobs:

# .env
OPTIMIZER_TRAINER_MODEL=ollama/mistral:7b
OPTIMIZER_MAX_ITERATIONS=10
OPTIMIZER_STRATEGY=combined

from agentomatic.optimize.settings import OptimizerPydanticSettings
opts = OptimizerPydanticSettings()
opts.display()  # pretty table

3. Programmatic (Python API)

from agentomatic import AgentPlatform
from agentomatic.storage import MemoryStore

platform = AgentPlatform.from_folder(
    "agents/",
    store=MemoryStore(),
    enable_auth=True, auth_api_key="secret",
    enable_metrics=True,
)
app = platform.build()

🗄️ Storage Backends

# Development
from agentomatic.storage import MemoryStore
store = MemoryStore()

# Production (PostgreSQL)
from agentomatic.storage import SQLAlchemyStore
store = SQLAlchemyStore("postgresql+asyncpg://user:pass@localhost/db")

# Custom
from agentomatic.storage import BaseStore
class RedisStore(BaseStore):
    async def create_thread(self, ...): ...
    async def get_thread(self, ...): ...

Durable task storage (for background tasks that must survive restarts or be shared across workers) uses a separate, pluggable TaskStore:

from agentomatic.tasks import SQLAlchemyTaskStore

platform = AgentPlatform.from_folder(
    "agents/",
    task_store=SQLAlchemyTaskStore("postgresql+asyncpg://user:pass@localhost/db"),
)

Defaults to an in-memory store; install with agentomatic[db] (SQLite) or agentomatic[db-postgres] (PostgreSQL). See Tasks → Persistence.

🎨 Debug UI

Built-in ChatGPT-like interface powered by Chainlit:

pip install "agentomatic[ui]"
agentomatic run --with-ui
# → http://localhost:8000/chat

Features: agent selector, streaming, tool call visualization, chain-of-thought, feedback collection.

🎨 Agentomatic Studio

Visual debugging environment for any agent framework — graph visualization, real-time execution tracing, state inspection, and time-travel debugging.

# Quick demo (no setup required)
agentomatic demo

# With your agents
agentomatic run --studio
# → Studio at http://localhost:8000/studio/ui/

Universal Framework Support:

Studio API Endpoints (mounted at /studio/):

Studio Decorators — incrementally upgrade any agent's Studio experience:

from agentomatic.studio import studio_graph, studio_state

@studio_graph
def my_topology():
    return {"nodes": [...], "edges": [...]}

@studio_state
async def get_state(thread_id: str) -> dict:
    return await my_db.get_state(thread_id)

📊 Auto-Generated Endpoints

Every agent gets 12+ endpoints automatically:

🔧 Prompt Fitting (ML-like API)

Agentomatic Optimize treats your deployed agent configuration as a parameter surface to fit against real evaluation data. The output is never a compiled program — it's a better deployment configuration: an improved prompt, tuned model parameters, optimized RAG settings, and a rollout recommendation you can ship with confidence.

Recommended entrypoints for class agents (scaffolded as train.py / eval.py):

from agentomatic.optimize import (
    TrainConfig, train_and_report, print_train_result,
    EvalConfig, evaluate_and_report,
)

result = train_and_report(agent, config=TrainConfig(
    agent_name="assistant", agent_dir=HERE, stacks_dir=ROOT / "stacks",
    epochs=2, max_trials=12, optimizer="rewrite",
    augment=True, n_examples=40, persist=True,
))
print_train_result(result)  # HolySheet HTML at result.report_path

ev = evaluate_and_report(agent, config=EvalConfig(
    agent_name="assistant", agent_dir=HERE, stacks_dir=ROOT / "stacks",
    split="test", prefer_augmented=True,
))

See the Prompt Optimization guide.

Philosophy: Your agent is already deployed. Optimization produces a better version of that deployment, not a new artifact. Every result includes a DeploymentRecommendation with canary weights and confidence scores so you can roll out safely.

EvalContract — Structural Quality Gate

Define what a valid agent response looks like before you optimize:

from agentomatic.optimize import EvalContract

contract = EvalContract(
    name="scoping_response",
    input_fields=["query", "context"],
    output_format="json",
    required_output_fields=["answer", "confidence", "risks", "next_questions"],
    constraints=["confidence must be between 0.0 and 1.0"],
)

score = contract.validate(response_text)       # 0.0 – 1.0
metric = contract.as_metric(weight=0.10)       # use inside CompositeMetric
criteria = contract.as_judge_criteria()         # feed to LLM judge

CompositeMetric — Multi-Dimensional Scoring

Combine quality judges with negative-weight cost/latency penalties so the optimizer balances accuracy against operational cost:

from agentomatic.optimize import (
    CompositeMetric, WeightedMetric,
    LocalJudgeMetric, LatencyMetric, CostMetric,
)

metric = CompositeMetric(metrics=[
    WeightedMetric("completeness",   LocalJudgeMetric("completeness"),      weight=0.30),
    WeightedMetric("relevance",      LocalJudgeMetric("business_relevance"),weight=0.25),
    WeightedMetric("risk_detection", LocalJudgeMetric("risk_detection"),    weight=0.20),
    WeightedMetric("format",         contract.as_metric(),                  weight=0.10),
    WeightedMetric("latency",        LatencyMetric(),                       weight=-0.10),
    WeightedMetric("cost",           CostMetric(),                          weight=-0.05),
])

Negative weights penalize candidates that are slower or more expensive, steering the fitter toward cost-effective configurations.

PromptSearchSpace — Full Configuration Surface

Tell the fitter what it's allowed to change:

from agentomatic.optimize import PromptSearchSpace

space = PromptSearchSpace(
    optimize_system_prompt=True,
    optimize_few_shot=True,
    optimize_model_params=True,
    optimize_model_choice=True,
    model_choices=["ollama/qwen2.5:7b", "openai/gpt-4.1"],
    fallback_models=["openai/gpt-4.1-mini"],
    model_param_space={
        "temperature": [0.0, 0.1, 0.2, 0.4, 0.7],
        "top_p": [0.7, 0.9, 1.0],
    },
    rag_param_space={"top_k": [3, 5, 8, 12], "rerank": [True, False]},
    optimize_rag_params=True,
)

PromptFitter — The scikit-learn-like API

from agentomatic.optimize import PromptFitter

fitter = PromptFitter(
    agent="scope_agent",
    task_model="ollama/qwen2.5:7b",
    rewrite_model="openai/gpt-4.1",
    optimizer="gepa_like",
    search_space=space,
    max_trials=30,
    min_absolute_improvement=0.05,
    concurrency=5,
)
result = await fitter.fit(trainset, valset, metric, testset=testset)

Access the full result surface:

result.best_prompt              # optimized system prompt
result.best_params              # {"temperature": 0.2, "top_p": 0.9}
result.best_few_shot_examples   # selected few-shot examples
result.metric_deltas            # per-dimension improvement
result.suggestions              # actionable recommendations
result.deployment_recommendation # canary rollout config
result.summary()                # human-readable summary
result.apply(version="v2_optimized")

Five optimisation strategies:

DeploymentRecommendation — Ship With Confidence

Every PromptFitResult includes a deployment recommendation based on the observed improvement magnitude and variance:

rec = result.deployment_recommendation
print(rec.confidence)              # "high" / "medium" / "low"
print(rec.rollout.strategy)        # "canary"
print(rec.rollout.initial_weight)  # 0.40
print(rec.summary())               # human-readable deployment plan

Failure Clusters — Targeted Diagnostics

The fitter groups validation failures into actionable clusters, each with the parameters most likely to resolve the issue and the expected metric gain:

Failure cluster 1:
  Agent answered without using retrieval context.
  → Suggested fix: force context-first behavior.
  → Affected params: rag.top_k, tool_policy.force_retrieval
  → Expected metric gain: faithfulness +0.18

Failure cluster 2:
  Agent produced unstructured answers.
  → Suggested fix: stronger output format block.
  → Affected params: prompt.output_contract
  → Expected metric gain: format_compliance +0.12

Optimization workflow

# 1. Run your agents
agentomatic run

# 2. Prepare a labelled JSONL evaluation dataset (query + expected_answer).
# See docs/optimization/ for the format, metrics and anti-overfitting guidance.

# 3. Optimize a running agent with the current fitter CLI.
agentomatic optimize scope_agent \
  --dataset scope_eval.jsonl \
  --mode rewrite \
  --metrics contains \
  --max-trials 20

# 4. Review the report, then verify the candidate in staging before release.

Vocabulary

🧠 Callbacks — No Adapters Needed

from agentomatic.callbacks import OptimizeEarlyStopping, ModelCheckpoint
from agentomatic.optimize import PromptFitter

fitter = PromptFitter(..., callbacks=[
    OptimizeEarlyStopping(patience=3),
    ModelCheckpoint(save_dir="checkpoints/"),
])

🔍 Agent Type Auto-Detection

from agentomatic.optimize.agent_detect import detect_agent_type, Evaluator
type = detect_agent_type(my_agent)  # STATELESS, RAG, TOOL_USING, ...
evaluator = Evaluator.for_agent(my_agent)  # auto-selects metrics

💾 Experiment Tracking

from agentomatic.optimize.experiment_tracker import ExperimentTracker
tracker = ExperimentTracker()
eid = await tracker.start_experiment("my_agent")
await tracker.log_iteration(eid, 1, 0.72)
tracker.display_experiments()

🔌 LangChain / LangGraph Adapter

from agentomatic.langchain_adapter import AgentAdapter
adapted = AgentAdapter(my_langgraph_agent)
result = await adapted.atransform({"current_query": "Hello"})

⚙️ Presets & Settings

from agentomatic.optimize.presets import Presets
Presets.for_local()    # free Ollama, 5 rounds
Presets.for_quality()  # GPT-4o, 10 rounds

🎯 agent.fit() One-Liner

from agentomatic.optimize.optimizer_mixin import OptimizerMixin
class MyAgent(OptimizerMixin, BaseGraphAgent): ...  # mixin first for fit(test_cases)
result = agent.fit(test_cases)  # sync → FitResult
# → "FitResult: 0.52 → 0.84 after 5 rounds"
# Keras path still works: agent.compile(...).fit(ds, epochs=2) → History

🧪 Per-Agent Evaluation Discovery

# agents/my_agent/evals.py — auto-discovered!
from agentomatic.optimize.evals_discovery import discover_agent_evals
discovered = discover_agent_evals()

🛠️ Development

# Install
git clone https://github.com/UnicoLab/agentomatic.git
cd agentomatic
make dev  # Installs all deps + pre-commit hooks

# Quality
make lint          # Ruff linter
make audit         # Known-vulnerability audit of locked dependencies
make format        # Auto-format
make typecheck     # Mypy
make test          # All tests
make test-cov      # With coverage
make check-all     # lint + dependency audit + typecheck + test

# Docs
make docs-serve    # Local docs server
make docs-build    # Build static site

# Build
make build         # Package
make publish       # PyPI

📜 License

MIT — see LICENSE.

👥 Authors

UnicoLab — Building the future of AI agent platforms.