logos-adk 0.3.0

Framework for building AI agents

Logos ADK — AI Agent Development Kit

Logos ADK is a modern, production-ready framework for building AI-powered agents and multi-agent systems. It provides a clean, Pythonic API with enterprise-grade features including tools, memory, guardrails, RAG, observability, and advanced workflow orchestration.

---

🌟 Features

Core Capabilities

Feature	Description
🤖 Agent System	Create AI 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 7+ guardrail types
📚 RAG	Retrieval-Augmented Generation with loaders, splitters, vector stores
🔗 MCP	Model Context Protocol for external tool servers
👥 Teams	Multi-agent orchestration with configurable topology
📊 Workflows	Graph, Pipeline, and Swarm orchestration
🎯 CrewAI-style	Sequential and hierarchical task processes
🐝 Swarm	Event-driven async orchestration with EventBus
🔭 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, management

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 learning, tool workflow learning, retrieval learning
🔄 State	Persistent state with SQLite/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 A2A communication

---

📦 Installation

Basic Installation

pip install logos-adk

Optional Dependencies

# All features
pip install logos-adk[all]

# Providers
pip install logos-adk[anthropic]  # Anthropic Claude
pip install logos-adk[openai]     # OpenAI GPT

# Features
pip install logos-adk[mcp]        # MCP support
pip install logos-adk[serve]      # HTTP server
pip install logos-adk[cli]        # CLI tools

# RAG
pip install logos-adk[rag-pdf]    # PDF loading
pip install logos-adk[rag-chroma] # ChromaDB vector store

# Gemini and built-in observability exporters work with the core package.

---

🚀 Quick Start

1. Basic Agent

import asyncio
from logos_adk import Agent

async def main():
    agent = Agent(
        "assistant",
        model="claude-sonnet-4-20250514",
        system="You are a helpful assistant. Keep responses concise."
    )

    response = await agent.run("What is Logos ADK?")
    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:
    """Get current weather for a city."""
    return f"It's +22°C and sunny in {city}"

@tool
async def search_listings(query: str) -> str:
    """Search for marketplace listings."""
    return f"Found 150 results for: {query}"

async def main():
    agent = (
        Agent("assistant", model="gpt-4o")
        .tools([get_weather, search_listings])
    )

    response = await agent.run("What's the weather in Moscow?")
    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-20250514",
        system="You are a researcher. Gather facts and data."
    )

    writer = Agent(
        "writer",
        model="gpt-4o",
        system="You are a writer. Create engaging content from facts."
    )

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

    response = await team.run("Write an article about quantum computing")
    print(response.content)

asyncio.run(main())

4. Graph Workflow

from logos_adk import Graph, AgentStep

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

# Add nodes
graph.node(AgentStep("draft", writer_agent))
graph.node(AgentStep("review", reviewer_agent))
graph.node(AgentStep("approve", approval_step))

# Add edges with conditions
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")

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

5. YAML Configuration

Create agent.yaml:

name: assistant
model: claude-sonnet-4-20250514
system_prompt: |
  You are a helpful assistant.
  Keep responses concise and actionable.

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 Concepts

Agent

The Agent class is the fundamental building block with fluent API:

from logos_adk import Agent, SessionMemory, ContentFilter

agent = (
    Agent("name", model="gpt-4o")
    .system("System prompt")
    .role("Assistant", goal="Help users", backstory="You are helpful")
    .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
    )
)

All Agent Methods:

Method	Purpose
.model(model)	Set LLM model
.system(prompt)	Set system prompt
.role(role, goal, backstory)	Set crew-style role
.tools(tools)	Attach tools
.memory(*memories)	Attach memory strategies
.mcp(source)	Connect to MCP server
.observe(*exporters)	Attach observability
.input_guardrail(guard)	Input validation
.output_guardrail(guard)	Output validation
.security(...)	Configure security
.reliability(...)	Configure reliability
.budget(...)	Set task budget
.self_qa(...)	Enable self-verification
.state_backend(store)	Attach persistent state
.knowledge(kb, mode, limit)	Attach RAG 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

Three memory types with multiple strategies:

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

7+ guardrail types for input/output validation:

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-4o")
    .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

Enterprise-grade 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

Production-ready reliability features:

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-4o", "claude-sonnet-4"]
)

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

Observability

Multiple exporters for tracing:

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-20250514
system_prompt: |
  You are a helpful assistant.

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

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
  - name: writer
    role: writer
    goal: Create content
    model: gpt-4o

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

---

🔌 Supported Providers

Provider	Models	Installation
Anthropic	Claude family	pip install logos-adk[anthropic]
OpenAI	GPT-4, GPT-4o, o1, o3	pip install logos-adk[openai]
Google	Gemini family	Built-in
Ollama	Local models	Built-in
OpenAI-Compatible	vLLM, LM Studio, Polza AI	Built-in

Provider Auto-Registration:

• claude-* → AnthropicProvider
• gpt-*, o1-*, o3-* → OpenAIProvider
• gemini-* → GeminiProvider

---

📚 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 welcome! Please:

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 for details.

---

🙏 Acknowledgments

Logos ADK builds upon best practices from:

• CrewAI — Agent orchestration patterns
• LangChain — Tool and memory abstractions
• AutoGen — Multi-agent communication
• Model Context Protocol — External tool integration