agenticblocks 🧱

A composable building block library for AI agent workflows. / Uma biblioteca componível para construir fluxos de agentes de IA.

🇺🇸 English | 🇧🇷 Português

🇺🇸 English

Philosophy

A library to build agent workflows like Lego blocks. Each step in your agentic pipeline is a self-contained block, with strictly typed inputs and outputs via Pydantic and natively concurrent execution using AsyncIO and NetworkX graphs.

• Strong typing: Pydantic validates connections and prevents unmatched dependencies between LLM tool calls.
• Standardized connections: Blocks only know their own inputs and outputs. Thus, entire workflows can act as single blocks later.
• Smart Parallelism (Waves): The asyncio engine fires simultaneous tasks (waves) whenever dependencies are resolved, maximizing API speed.
• Native Cycles: Declare bounded feedback loops directly in the graph (add_cycle()). The executor handles iteration, feedback propagation, and exit conditions automatically.
• Functions as Tools: Any plain Python function (sync or async) becomes a block with @as_tool — no class boilerplate required.
• Focus on local open-source models: small models run well with this library. Only native provider tool_calls execute actions — text is never reinterpreted as a tool call — and built-in loop detection blocks a call a model keeps repeating verbatim instead of correcting it.
• Comprehensive Agentic Components: Built-in support for various components of agentic systems, including integration with MemGPT for advanced memory management and persistent context.

Getting Started

Install the latest release:

pip install agenticblocks.io

Or for local development:

pip install -e .

1. Define Input and Output Models

from pydantic import BaseModel

class HelloInput(BaseModel):
    name: str

class HelloOutput(BaseModel):
    greeting: str

2. Create the Logic Block

from agenticblocks.core.block import Block

class HelloWorldBlock(Block[HelloInput, HelloOutput]):
    name: str = "say_hello"
    
    async def run(self, input: HelloInput) -> HelloOutput:
        msg = f"Hello, {input.name}! Welcome to agenticblocks."
        return HelloOutput(greeting=msg)

3. Connect and Execute

import asyncio
from agenticblocks.core.graph import WorkflowGraph
from agenticblocks.runtime.executor import WorkflowExecutor

async def main():
    graph = WorkflowGraph()
    graph.add_block(HelloWorldBlock(name="say_hello"))

    executor = WorkflowExecutor(graph)
    ctx = await executor.run(initial_input={"name": "Alice"})
    
    print(ctx.get_output("say_hello").greeting)

asyncio.run(main())

4. Functions as Tools

Any Python function can be registered as a block with @as_tool. Both sync and async functions are supported — sync functions run in a thread pool automatically.

from agenticblocks import as_tool
from agenticblocks.blocks.llm.agent import LLMAgentBlock

@as_tool
async def fetch_weather(city: str) -> str:
    """Returns the current weather for a city."""
    return f"Sunny in {city}."

agent = LLMAgentBlock(
    name="assistant",
    model="gpt-4o-mini",
    tools=[fetch_weather],   # same interface as any Block
)

5. LLM Agent Autonomy & A2A

LLMAgentBlock is a ready-to-use orchestrator that dynamically translates your other Blocks into Tools (Agent-to-Agent) seamlessly.

• Bounded tool loop: max_tool_calls prevents runaway loops.
• A2A bridging: sub-agents are called as tools transparently — the parent LLM receives only the text response, not raw JSON metadata.
• Connection Pooling: Pass any model_kargs (HTTP clients, timeouts, etc.) to optimize API performance.
• Iteration Hooks: Use the on_iteration parameter to inject callbacks that execute at the beginning of each loop iteration. These callbacks receive the current iteration number and the list of messages, allowing for real-time monitoring, logging, or dynamic prompt adjustments during the agent's execution.

6. Advanced Flow Control

• PromptBuilderBlock: Merges outputs from multiple predecessors into a single formatted AgentInput prompt using Python format-strings. Useful for "diamond" graph patterns (e.g., feeding both the original topic and a search report to a final summarizer).
• ImageGenerationBlock: Generates images through LiteLLM behind a single contract — agenticblocks.ImageGenerationBlock returns ImageArtifacts carrying real bytes with a sniffed MIME type, whichever route the provider exposes (images API or multimodal chat). Register your own route with register_image_adapter; anything that cannot generate raises a classified ImageGenerationError instead of returning an empty file.
• Loop detection: LLMAgentBlock and MemGPTAgentBlock track the (name, arguments) signature of every tool call. A call repeated more than loop_detection_limit times (default 3) is blocked before execution and answered with a corrective tool result, so a weak model changes approach instead of burning its whole budget on the same failing call.

7. Native Feedback Cycles

Declare validator loops directly in the graph without any wrapper block:

from agenticblocks import as_tool
from agenticblocks.core.graph import WorkflowGraph
from agenticblocks.blocks.llm.agent import LLMAgentBlock

writer = LLMAgentBlock(
    name="writer",
    model="gpt-4o-mini",
    system_prompt="Write a detailed essay about AI."
)

@as_tool
def validate_output(content: str) -> dict:
    ok = len(content.split()) >= 100
    return {"is_valid": ok, "feedback": "Too short." if not ok else ""}

graph = WorkflowGraph()
graph.add_block(writer)
graph.add_block(validate_output)

graph.add_cycle(
    name="refine",
    edges=[("writer", "validate_output")],
    condition_block="validate_output",
    max_iterations=3,
)
# Downstream nodes connect to the cycle output as a normal node
graph.connect("refine", "publisher")

The executor runs the cycle, propagates feedback to the producer on each rejection, and stores the result in ctx under the cycle name.

8. Code-Generation Planning

AgenticBlocks supports dynamic code execution out-of-the-box. Instead of relying on predefined tools, you can use the CodePlanExecutorBlock to orchestrate an LLM that generates a standalone Python script to solve a problem. The generated code is automatically executed either locally or securely in an ephemeral Docker container via the PythonCodeExecutorBlock. See more in docs/codeplanner.md.

Examples & Model Recommendations

It is recommended to install Ollama with the model granite4:1b (ollama run granite4:1b) to test the examples locally. Alternatively, you can modify the examples to use a commercial API, such as Gemini (gemini/gemini-2.0-flash) or OpenAI.

Note: Quantized or small models like granite may produce lower-quality reasoning and struggle with native tool calling. Since only native tool_calls execute actions, a model that writes its calls as plain-text JSON simply will not act — for reliable local tool usage, use llama3.1 or mistral-nemo. Large commercial models (OpenAI, Gemini, Anthropic) yield excellent results but require an API key.

🇧🇷 Português

Filosofia

Uma biblioteca para construir fluxos de agentes no estilo Lego. Cada passo do seu pipeline agêntico é um bloco auto-contido, com entradas e saídas rigorosamente tipadas via Pydantic e execução simultânea usando AsyncIO e grafos do NetworkX.

• Forte tipagem: Pydantic valida os encaixes e previne dependências não satisfeitas.
• Encaixes padronizados: Blocos só conhecem as próprias entradas e saídas. Workflows inteiros funcionam como blocos únicos.
• Paralelismo Inteligente (Ondas): O motor dispara tarefas simultâneas sempre que as dependências de um bloco são resolvidas.
• Ciclos Nativos: Declare loops de feedback diretamente no grafo com add_cycle(). O executor gerencia iteração, propagação de feedback e condição de saída automaticamente.
• Funções como Ferramentas: Qualquer função Python (síncrona ou async) vira um bloco com @as_tool — sem boilerplate de classe.
• Foco em modelos locais open-source: modelos pequenos rodam bem com esta biblioteca. Somente tool_calls nativas do provedor executam ações — texto nunca é reinterpretado como chamada de ferramenta — e a detecção de laços bloqueia a chamada que o modelo repete literalmente em vez de corrigir.
• Componentes Agênticos Abrangentes: Suporte nativo para vários componentes de sistemas agênticos, incluindo integração com MemGPT para gerenciamento avançado de memória e contexto persistente.

Primeiros Passos

Instale o último lançamento:

pip install agenticblocks.io

Ou para desenvolvimento local:

pip install -e .

1–3. Blocos, Grafo e Execução

A estrutura básica é idêntica ao tutorial acima (seção em inglês): defina modelos Pydantic, crie um Block, adicione ao WorkflowGraph e execute com WorkflowExecutor.

4. Funções como Ferramentas

Qualquer função pode ser registrada como bloco com @as_tool. Funções síncronas rodam em thread pool automaticamente.

from agenticblocks import as_tool

@as_tool
def buscar_clima(cidade: str) -> str:
    """Retorna o clima atual de uma cidade."""
    return f"Ensolarado em {cidade}."

5. Autonomia com Agentes LLM & A2A

O LLMAgentBlock abstrai e converte sub-blocos em ferramentas nativas (A2A). Destaques:

• max_tool_calls: Limita o loop de ferramentas para evitar execuções infinitas.
• A2A transparente: Agentes subordinados são chamados como ferramentas; o agente pai recebe apenas o texto da resposta, sem metadados JSON brutos.
• Connection Pooling: Aceite sessões HTTP e parâmetros estendidos via model_kargs.

6. Controle de Fluxo Avançado

• PromptBuilderBlock: Mescla saídas de múltiplos predecessores em um único prompt AgentInput formatado. Ideal para padrões de grafo em "diamante".
• ImageGenerationBlock: Gera imagens via LiteLLM sob um contrato único — agenticblocks.ImageGenerationBlock devolve ImageArtifacts com bytes reais e MIME detectado, seja qual for a rota do provedor (API images ou chat multimodal). Registre rotas próprias com register_image_adapter; o que não puder ser gerado levanta um ImageGenerationError classificado em vez de devolver um arquivo vazio.
• Detecção de laços: LLMAgentBlock e MemGPTAgentBlock acompanham a assinatura (nome, argumentos) de cada chamada. Uma chamada repetida mais de loop_detection_limit vezes (padrão 3) é bloqueada antes da execução e respondida com um resultado corretivo, para o modelo mudar de abordagem em vez de gastar todo o orçamento na mesma chamada que falha.
• Callbacks de Iteração (on_iteration): Use o parâmetro on_iteration para injetar funções de callback que serão executadas no início de cada iteração do loop do agente. Isso permite monitoramento em tempo real, logging ou ajustes dinâmicos no prompt durante a execução do agente.

7. Ciclos de Feedback Nativos

Declare um loop validador diretamente no grafo — sem bloco orquestrador especial:

from agenticblocks import as_tool
from agenticblocks.core.graph import WorkflowGraph
from agenticblocks.blocks.llm.agent import LLMAgentBlock

escritor = LLMAgentBlock(
    name="escritor",
    model="gpt-4o-mini",
    system_prompt="Escreva uma redação detalhada sobre IA."
)

@as_tool
def validar(content: str) -> dict:
    ok = len(content.split()) >= 100
    return {"is_valid": ok, "feedback": "Muito curto." if not ok else ""}

graph = WorkflowGraph()
graph.add_block(escritor)
graph.add_block(validar)

graph.add_cycle(
    name="refinar",
    edges=[("escritor", "validar")],
    condition_block="validar",
    max_iterations=3,
)
graph.connect("refinar", "publicador")

O executor itera automaticamente, injeta o feedback no prompt do produtor a cada rejeição e disponibiliza o resultado final em ctx.get_output("refinar").

8. Planejamento Baseado em Geração de Código

O AgenticBlocks oferece suporte a execução dinâmica de código nativamente. Em vez de depender de ferramentas pré-definidas, você pode usar o CodePlanExecutorBlock para orquestrar um LLM na geração de um script Python completo capaz de resolver um problema. O código gerado é então testado e executado automaticamente no modo local ou de forma segura isolado em um contêiner Docker via PythonCodeExecutorBlock. Veja mais em docs/codeplanner.md.

Exemplos & Modelos

Recomenda-se instalar o Ollama com o modelo granite4:1b para testar localmente. Alternativamente, use uma API comercial como Gemini ou OpenAI.

Atenção: Modelos quantizados ou menores podem produzir resultados abaixo do esperado e ter dificuldade com chamadas nativas de ferramentas. Como apenas tool_calls nativas executam ações, um modelo que escreve as chamadas como JSON em texto simplesmente não age — para uso local confiável, prefira llama3.1 ou mistral-nemo. Modelos comerciais grandes geram excelentes resultados, mas exigem configuração de API KEY.