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, as we provide ready-made blocks that handle their limitations, such as HeuristicLLMAgentBlock, which heuristically extracts tool calls in JSON format from plain text and executes them transparently. See more in docs/heuristicagent.md.
• 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 & Heuristics

• 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).
• HeuristicLLMAgentBlock: A specialized agent for models with weak native tool support (like smaller local models). It heuristically parses hallucinated JSON tool calls out of plain text responses and executes them transparently.

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. For reliable local tool usage, use llama3.1 or mistral-nemo. If using granite4, prefer the HeuristicLLMAgentBlock to capture hallucinated JSON tool calls. 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, pois provemos blocos prontos que lidam com suas limitações, como HeuristicLLMAgentBlock, que extrai heuristicamente chamadas de ferramenta em formato JSON do texto plano e as executa de forma transparente. Veja mais em docs/heuristicagent.md.
• 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 & Heurísticas

• PromptBuilderBlock: Mescla saídas de múltiplos predecessores em um único prompt AgentInput formatado. Ideal para padrões de grafo em "diamante".
• HeuristicLLMAgentBlock: Agente especializado para modelos com suporte fraco a chamadas de ferramentas (como modelos locais pequenos). Ele extrai heuristicamente chamadas de ferramenta em formato JSON do texto plano e as executa de forma transparente.
• 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. Para uso local confiável de ferramentas, prefira llama3.1 ou mistral-nemo. Caso use granite4, utilize o HeuristicLLMAgentBlock. Modelos comerciais grandes geram excelentes resultados, mas exigem configuração de API KEY.