logos-adk 0.3.1

Framework for building AI agents

Logos ADK — AI Agent Development Kit

Logos ADK is a Python framework for building AI agents, multi-agent systems, and workflow-driven applications. It combines tools, memory, guardrails, RAG, observability, and serving primitives in one codebase.

---

🌟 Features

Core Capabilities

Feature	Description
🤖 Agent System	Create agents with configurable models, roles, goals, and backstories
🔧 Tools	Async tool registration with automatic schema generation
🧠 Memory	Session, long-term, and shared memory with multiple strategies
🛡️ Guardrails	Input/output validation with multiple guard types
📚 RAG	Retrieval with loaders, splitters, and vector stores
🔗 MCP	Model Context Protocol integration for external tool servers
👥 Teams	Multi-agent orchestration with configurable topology
📊 Workflows	Graph, pipeline, crew, and swarm orchestration
🔭 Observability	Built-in tracing with multiple exporters
💾 Caching	Exact and semantic response caching
🌐 Serving	HTTP API with streaming and CORS support
🎛️ CLI	Command-line interface for chat, testing, and operations

Production Features

Feature	Description
🔒 Security	Input validation, prompt injection detection, PII redaction, audit logging
♻️ Reliability	Retry policies, circuit breakers, fallback models, timeouts
💰 Cost Control	Budget tracking, cost-aware routing, spend ledgers
📈 Learning	Model routing, tool workflow, and retrieval learning loops
🔄 State	Persistent state with SQLite and PostgreSQL backends
⚡ Scaling	Horizontal scaling with load balancing and session affinity
✅ Self-QA	Self-verification with assertions and retries
📬 Events	Event bus with subscriptions and agent-to-agent communication

---

📦 Installation

pip install logos-adk

Optional Extras

pip install logos-adk[all]
pip install logos-adk[anthropic]
pip install logos-adk[openai]
pip install logos-adk[mcp]
pip install logos-adk[serve]
pip install logos-adk[cli]
pip install logos-adk[rag-pdf]
pip install logos-adk[rag-chroma]

The examples below use current provider model names as of March 26, 2026. These names can change, so check provider docs before copying them into production config.

---

🚀 Quick Start

1. Basic Agent

import asyncio
from logos_adk import Agent

async def main():
    agent = Agent(
        "assistant",
        model="claude-sonnet-4-6",
        system="Answer clearly and keep responses short.",
    )
    response = await agent.run("What can you do?")
    print(response.content)

asyncio.run(main())

2. Agent with Tools

import asyncio
from logos_adk import Agent, tool

@tool
async def get_weather(city: str) -> str:
    """Return a stub weather report."""
    return f"Weather for {city}: clear, 22C"

@tool
async def search_docs(query: str) -> str:
    """Return a stub search result."""
    return f"Results for {query}"

async def main():
    agent = Agent("assistant", model="gpt-5.4").tools([get_weather, search_docs])
    response = await agent.run("Find the weather in Berlin")
    print(response.content)

asyncio.run(main())

3. Multi-Agent Team

import asyncio
from logos_adk import Agent, Team

async def main():
    researcher = Agent(
        "researcher",
        model="claude-sonnet-4-6",
        system="Collect the facts and keep notes concise.",
    )
    writer = Agent(
        "writer",
        model="gpt-5.4",
        system="Turn the research into a readable draft.",
    )

    team = Team(
        "content-team",
        agents=[researcher, writer],
        orchestrator="researcher",
        topology="mesh",
    )

    response = await team.run("Write a short explainer about vector databases")
    print(response.content)

asyncio.run(main())

4. Graph Workflow

from logos_adk import AgentStep, Graph

graph = Graph("review-workflow", max_visits=10)

graph.node(AgentStep("draft", writer_agent))
graph.node(AgentStep("review", reviewer_agent))
graph.node(AgentStep("approve", approval_agent))

graph.edge("draft", "review", label="draft_ready")
graph.edge("review", "approve", when=lambda ctx: ctx["approved"], label="approved")
graph.edge("review", "draft", when=lambda ctx: not ctx["approved"], label="needs_revision")

graph.entrypoint("draft")
graph.terminal("approve")

5. YAML Config

agent.yaml:

name: assistant
model: claude-sonnet-4-6
system_prompt: |
  You help users and keep answers practical.

security:
  input_validation_enabled: true
  prompt_shield_enabled: true

reliability:
  timeout_seconds: 60.0
  retry_max_attempts: 3

Load and run:

from logos_adk import Agent

agent = Agent.from_config("agent.yaml")
response = await agent.run("Hello")

---

🏗️ Core API

from logos_adk import Agent, SessionMemory
from logos_adk.guardrails import ContentFilter, MaxLengthGuard
from logos_adk.observe import ConsoleExporter
from logos_adk.reliability import RetryPolicy
from logos_adk.security import InputValidator, PromptShield

agent = (
    Agent("assistant", model="gpt-5.4")
    .system("Be precise.")
    .tools([tool1, tool2])
    .memory(SessionMemory(strategy="sliding_window", max_messages=20))
    .input_guardrail(ContentFilter(blocked=["spam"]))
    .output_guardrail(MaxLengthGuard(max_chars=500))
    .observe(ConsoleExporter())
    .security(
        input_validator=InputValidator(max_length=5000),
        prompt_shield=PromptShield(threshold=0.5),
    )
    .reliability(
        retry_policy=RetryPolicy(max_attempts=3),
        provider_timeout=60.0,
    )
)

Common Methods:

Method	Purpose
.model(model)	Set the model name or provider
.system(prompt)	Set the system prompt
.role(role, goal, backstory)	Set crew-style role metadata
.tools(tools)	Attach tools
.memory(*memories)	Attach memory strategies
.mcp(source)	Connect to an MCP server
.observe(*exporters)	Attach exporters
.input_guardrail(guard)	Validate input
.output_guardrail(guard)	Validate output
.security(...)	Configure security controls
.reliability(...)	Configure retries and timeouts
.budget(...)	Set task budget
.self_qa(...)	Enable self-checks
.state_backend(store)	Attach persistent state
.knowledge(kb, mode, limit)	Attach a knowledge base

Tools

Tools are async functions with automatic schema generation:

from logos_adk import tool, Toolkit

@tool
async def search(query: str, limit: int = 10) -> str:
    """Search for information.
    
    Args:
        query: Search query
        limit: Maximum results (default: 10)
    """
    return f"Results for: {query}"

# Create toolkit
class SearchToolkit(Toolkit):
    def get_tools(self):
        return [search, analyze, summarize]

agent.tools([search])
# or
agent.tools(SearchToolkit())

Memory

Built-in memory types:

from logos_adk import SessionMemory, LongTermMemory, SharedMemory
from logos_adk.memory import SQLiteBackend

# Session memory (short-term)
session_mem = SessionMemory(
    strategy="sliding_window",  # or "summarize", "token_limit"
    max_messages=20,
    max_tokens=16000
)

# Long-term memory (persistent)
long_term = LongTermMemory(
    backend=SQLiteBackend("memory.db"),
    retrieval="hybrid",  # or "keyword", "semantic"
    limit=5
)

# Shared memory (cross-agent)
shared = SharedMemory(backend=SQLiteBackend("shared.db"))

agent.memory(session_mem, long_term, shared)

Guardrails

Input and output guards:

from logos_adk.guardrails import (
    ContentFilter,      # Block words/phrases
    RegexGuard,         # Pattern matching
    MaxLengthGuard,     # Length limits
    KeywordGuard,       # Keyword detection
    ToxicityGuard,      # Toxic language
    ContentModerationGuard,  # Content moderation
    HallucinationGuard  # Fact checking
)

agent = (
    Agent("bot", model="gpt-5.4")
    .input_guardrail(
        ContentFilter(blocked=["spam", "abuse"]),
        MaxLengthGuard(max_chars=5000)
    )
    .output_guardrail(
        ToxicityGuard(threshold=2),
        ContentFilter(blocked=["competitor"])
    )
)

RAG (Retrieval-Augmented Generation)

Add knowledge base capabilities:

from logos_adk import KnowledgeBase, rag_tool
from logos_adk.rag import (
    TextLoader, MarkdownLoader,
    RecursiveCharacterSplitter,
    SQLiteVectorStore
)

# Load documents
loader = MarkdownLoader("docs/")
documents = loader.load()

# Create vector store
splitter = RecursiveCharacterSplitter(chunk_size=500, chunk_overlap=50)
vector_store = SQLiteVectorStore("rag.db")
vector_store.add_documents(documents, splitter)

# Create knowledge base
kb = KnowledgeBase(
    store=vector_store,
    embed_fn=your_embedding_function,
    splitter=splitter
)

# Option 1: Auto-inject context
agent.knowledge(kb, mode="memory", limit=5)

# Option 2: Add as tool
agent.tools([rag_tool(kb, limit=5)])

Workflows

Graph Workflow

Stateful graph with conditional routing:

from logos_adk import Graph, AgentStep, FunctionStep

graph = Graph("workflow", max_visits=10)

graph.node(AgentStep("analyze", analyst))
graph.node(AgentStep("write", writer))
graph.node(FunctionStep("save", save_function))

graph.edge("analyze", "write", label="ready")
graph.edge("write", "save", when=lambda ctx: ctx["approved"])
graph.edge("write", "analyze", when=lambda ctx: not ctx["approved"])

graph.entrypoint("analyze")
graph.terminal("save")

result = await graph.run("Write an article")

Pipeline Workflow

Linear workflow with step composition:

from logos_adk import Pipeline

pipeline = Pipeline("pipeline", timeout=120.0)

pipeline.step(agent1, name="step1")
pipeline.function(process_fn, name="step2")
pipeline.parallel(agent2, agent3, name="parallel")
pipeline.conditional(
    lambda ctx: ctx["score"] > 0.8,
    then=agent_yes,
    else_=agent_no
)
pipeline.loop(agent, until=lambda ctx: ctx["done"], max_iterations=5)
pipeline.approval(name="approval", message="Approve?")

result = await pipeline.run("prompt")

Crew Process

CrewAI-style task orchestration:

from logos_adk import CrewProcess, CrewTask, Team

tasks = [
    CrewTask(
        name="research",
        description="Research the topic",
        role="researcher",
        depends_on=[],
        async_execution=False
    ),
    CrewTask(
        name="write",
        description="Write article",
        role="writer",
        depends_on=["research"]
    ),
    CrewTask(
        name="approve",
        description="Approve content",
        role="manager",
        task_type="approval",
        depends_on=["write"]
    )
]

crew = CrewProcess(
    name="content_crew",
    team=team,
    tasks=tasks,
    process="sequential"  # or "hierarchical"
)

result = await crew.run("kickoff prompt")

Swarm (Event-Driven)

Async event-driven orchestration with EventBus:

from logos_adk.swarm import Swarm
from logos_adk.event_bus import InMemoryEventBus

bus = InMemoryEventBus()
swarm = Swarm("seo-swarm", event_bus=bus)

# Subscribe agents to topics
swarm.subscribe("need_seo", agent1, publish_to="need_copy")
swarm.subscribe("need_copy", agent2, publish_to="done")

# Dispatch and process
response = await swarm.dispatch(
    prompt="Optimize listing",
    request_topic="need_seo",
    response_topic="done"
)

Swarm patterns:

• Event-driven async processing
• Topic-based pub/sub
• Chain of handlers
• Fan-out/fan-in

See docs/SWARM.md for complete documentation.

Security

Security primitives:

from logos_adk.security import (
    InputValidator,
    PromptShield,
    PIIDetector,
    AuditLogger
)

agent.security(
    input_validator=InputValidator(
        min_length=1,
        max_length=20000,
        reject_null_bytes=True
    ),
    prompt_shield=PromptShield(
        patterns=["inject", "ignore"],
        threshold=0.5
    ),
    pii_detector=PIIDetector(
        blocked_types=["ssn", "credit_card", "email"]
    ),
    audit_logger=AuditLogger(
        redact_pii=True,
        retention_days=30
    )
)

Reliability

Reliability controls:

from logos_adk.reliability import RetryPolicy, CircuitBreaker

agent.reliability(
    retry_policy=RetryPolicy(
        max_attempts=3,
        base_delay=0.25,
        backoff="exponential"
    ),
    provider_timeout=60.0,
    fallback_models=["gpt-5.4", "claude-sonnet-4-6"]
)

# Circuit breaker auto-configured with:
# - failure_threshold: 5
# - recovery_timeout: 30s

Observability

Tracing exporters:

from logos_adk.observe import (
    ConsoleExporter,
    OpenTelemetryExporter,
    LangfuseExporter
)

agent.observe(
    ConsoleExporter(),  # Debug
    OpenTelemetryExporter(
        endpoint="http://localhost:4318/v1/traces",
        compression="gzip"
    ),
    LangfuseExporter(
        base_url="https://cloud.langfuse.com",
        public_key="pk-...",
        secret_key="sk-..."
    )
)

---

YAML Configuration

Agent Configuration

name: assistant
model: claude-sonnet-4-6
system_prompt: |
  You are a helpful assistant.

provider:
  type: openai-compatible
  base_url: http://localhost:11434/v1
  model: local-model

tools:
  - name: search
    description: Search for information
    parameters:
      type: object
      properties:
        query:
          type: string
      required: [query]

memories:
  session:
    strategy: sliding_window
    max_messages: 20
  long_term:
    backend:
      type: sqlite
      path: .logos_adk/memory.db
    retrieval: hybrid
    limit: 5

security:
  input_validation_enabled: true
  prompt_shield_enabled: true
  pii_detection_enabled: true
  output_max_length: 20000

reliability:
  timeout_seconds: 60.0
  retry_max_attempts: 3
  circuit_breaker_failure_threshold: 5

guardrails:
  input:
    - type: max_length
      max_chars: 5000
  output:
    - type: content_filter
      blocked: ["spam"]

knowledge:
  backend: sqlite
  path: .logos_adk/rag.db
  mode: tool
  limit: 5
  splitter:
    chunk_size: 500
    chunk_overlap: 50

observe:
  exporter: console

Graph Configuration

name: workflow
max_visits: 10
entrypoint: "step1"
terminal_nodes: ["step3"]

nodes:
  - name: "step1"
    agent: "agent1"
    metadata:
      ui: {x: 100, y: 100}
      description: "First step"
  
  - name: "step2"
    approval:
      message: "Approve?"
  
  - name: "step3"
    function: "my_function"

edges:
  - from: "step1"
    to: "step2"
    label: "ready"
  
  - from: "step2"
    to: "step3"
    when: "approval.status == 'approved'"
    label: "approved"
  
  - from: "step2"
    to: "step1"
    when: "approval.status == 'rejected'"
    label: "rejected"

Team Configuration

name: content_team
agents:
  - name: researcher
    role: researcher
    goal: Gather facts
    model: claude-sonnet-4-6
  - name: writer
    role: writer
    goal: Create content
    model: gpt-5.4

orchestrator: researcher
topology: mesh
max_rounds: 10

process: sequential
manager_role: manager

budget:
  team_budget_usd: 10.0
  task_budget_usd: 0.50

Swarm Configuration

name: seo_swarm
kind: swarm

event_bus:
  backend: sqlite
  path: .logos_adk/events.db

subscriptions:
  - agent_name: seo_expert
    topic: need_seo
    publish_to: need_copy
  
  - agent_name: copywriter
    topic: need_copy
    publish_to: need_quality
  
  - agent_name: quality_analyst
    topic: need_quality
    publish_to: publish_ready

# Usage in Python:
# swarm = Swarm.from_config("swarm.yaml", agents=agents, event_bus=bus)
# response = await swarm.dispatch(
#     prompt="Optimize listing",
#     request_topic="need_seo",
#     response_topic="publish_ready"
# )

---

CLI Usage

# Interactive chat
logos-adk chat agent.yaml

# Single prompt
logos-adk run agent.yaml "What is AI?"

# Streaming response
logos-adk run agent.yaml "Tell a story" --stream

# Validate config
logos-adk validate agent.yaml

# Test agent
logos-adk test agent.yaml

# Start HTTP server
logos-adk serve --agents agent.yaml --port 8000

# Inspect project and release helpers
logos-adk project --help
logos-adk upgrade --help

# State management
logos-adk storage upgrade-check --kind state --backend sqlite --path .logos_adk/state.db
logos-adk storage migrate --kind state --backend sqlite --path .logos_adk/state.db

Learning And Operator Storage

Learning/operator CLI commands now support both SQLite and PostgreSQL.

Use SQLite paths when you want local files:

logos-adk evaluation scorecard --db-path .logos_adk/learning.db --workspace-id acme
logos-adk retrieval summary --db-path .logos_adk/retrieval_learning.db --workspace-id acme

Use DSNs when the same data lives in PostgreSQL:

logos-adk evaluation scorecard --dsn postgresql://user:pass@localhost/learning --workspace-id acme
logos-adk model-routing leaderboard --dsn postgresql://user:pass@localhost/routing --workspace-id acme
logos-adk retrieval summary --dsn postgresql://user:pass@localhost/retrieval --workspace-id acme
logos-adk tool-workflow tool-scores --dsn postgresql://user:pass@localhost/tool_workflow --workspace-id acme

For multi-store commands, pass component-specific DSNs:

logos-adk improvement policy \
  --dsn postgresql://user:pass@localhost/learning \
  --model-routing-dsn postgresql://user:pass@localhost/routing \
  --retrieval-dsn postgresql://user:pass@localhost/retrieval \
  --tool-workflow-dsn postgresql://user:pass@localhost/tool_workflow \
  --workspace-id acme

logos-adk transfer artifacts \
  --dsn postgresql://user:pass@localhost/learning \
  --reflection-dsn postgresql://user:pass@localhost/reflection \
  --prompt-registry-dsn postgresql://user:pass@localhost/prompts \
  --retrieval-dsn postgresql://user:pass@localhost/retrieval \
  --tool-workflow-dsn postgresql://user:pass@localhost/tool_workflow \
  --model-routing-dsn postgresql://user:pass@localhost/routing \
  --source-agent-name support-bot

---

Examples

The public repository ships its runnable examples as documentation-first examples in docs/examples.md. The quickest production-oriented walkthroughs are:

• docs/examples.md for copy-paste examples
• docs/production.md for deployment and storage guidance
• docs/http-api.md for the HTTP surface

---

Project Structure

logos_adk/
├── agent/                # Agent core and execution helpers
├── team/                 # Multi-agent orchestration
├── cli/                  # Command-line interface
├── serve_routes/         # HTTP route handlers and schemas
├── learning/             # Evaluation and self-improvement modules
├── workflow/
│   ├── graph.py          # Graph workflow
│   ├── pipeline.py       # Pipeline workflow
│   ├── config.py         # YAML parsing
│   └── context.py        # Execution context
├── tools/                # Tool system
├── memory/               # Memory backends
├── guardrails/           # Input/output validation
├── rag/                  # RAG components
├── mcp/                  # MCP integration
├── observe/              # Observability exporters
├── cache/                # Response caching
├── providers/            # LLM providers
├── security.py           # Security primitives
├── reliability.py        # Retry, circuit breaker
└── serve.py              # FastAPI application factory

---

Providers

Supported provider integrations:

Provider	Notes	Installation
Anthropic	Current examples: claude-sonnet-4-6, claude-opus-4-6	pip install logos-adk[anthropic]
OpenAI	Current examples: gpt-5.4, gpt-5-mini, gpt-5-nano	pip install logos-adk[openai]
Google Gemini	Current examples: gemini-2.5-pro, gemini-2.5-flash, gemini-2.5-flash-lite	Built-in
OpenAI-compatible APIs	Ollama, LM Studio, vLLM, and similar endpoints	Built-in

Provider auto-registration is based on the model name pattern. If you want full control, pass a provider instance directly instead of a string.

Publishing

PyPI publishing is automated with GitHub Actions via .github/workflows/publish-pypi.yml.

Recommended setup:

1. Create the public GitHub repository.
2. Create the logos-adk project on PyPI.
3. In PyPI, add a Trusted Publisher for:
   • repository owner: your GitHub user or org
   • repository name: logos-adk
   • workflow file: .github/workflows/publish-pypi.yml
   • environment: pypi
4. Create a GitHub release. The workflow will build the package and publish it to PyPI.

Documentation

• Production Guide — MVP baseline for production
• Examples — Working code examples
• HTTP API — Public server routes and schemas
• Quality Testing — Public quality baseline and operator workflows

---

Contributing

Contributions are welcome. Before opening a PR:

1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Run tests: pytest
5. Run linters: ruff check . && mypy .
6. Submit a PR

---

License

MIT License. See LICENSE.