[](https://www.mangaba.ia.br)
[](https://www.mangaba.ia.br)
# Mangaba AI
[](https://pypi.org/project/mangaba/)
[](https://www.python.org/)
[](https://opensource.org/licenses/MIT)
[](https://github.com/Mangaba-ai/mangaba_ai/actions/workflows/ci.yml)
Framework profissional de orquestração multi-agente com ReAct reasoning, function calling nativo, RAG, memória persistente, protocolos A2A/MCP, vector stores e suporte resiliente a múltiplos provedores LLM.
Execute o exemplo abaixo em 1 comando (veja `examples/quickstart.py`):
```bash
git clone https://github.com/Mangaba-ai/mangaba_ai.git && cd mangaba_ai
pip install -e ".[google]"
export GOOGLE_API_KEY="sua-chave"
python examples/quickstart.py
```
Alternativa leve e completa a CrewAI + LangChain em um único pacote, com interoperabilidade real entre provedores, arquitetura resiliente e comunicação entre agentes via protocolos padrão.
## Destaques da v4.0.0
- **Flows** — Orquestração orientada a eventos com `@start`, `@listen`, `@router`, combinadores `and_`/`or_`, estado tipado em Pydantic, persistência em SQLite (`@persist`, com resume e fork) e gráfico de execução em HTML via `flow.plot()`
- **Knowledge** — Base de conhecimento separada da memória: PDF, DOCX, Excel, JSON, CSV, URL e diretórios inteiros, com `score_threshold` e escopo por agente ou por crew
- **Memória unificada** — Classe `Memory` com recall semântico composto (similaridade + recência + importância), extração automática de fatos, consolidação de duplicatas e gravação assíncrona
- **CLI** — `mangaba create/run/test/train/chat/config/reset-memories/replay/install`, com projetos declarados em `agents.yaml` / `tasks.yaml`
- **Treino e avaliação** — `mangaba train` (loop de feedback humano) e `mangaba test` (nota de qualidade 1–10 por tarefa e por agente)
- **Modelos locais** — Ollama e qualquer servidor OpenAI-compatible (vLLM, LM Studio, llama.cpp, LocalAI), com tool calling nativo e fallback automático
- **MCP (Model Context Protocol)** — Consuma ferramentas de servidores MCP externos como `BaseTool` nativas, via stdio ou HTTP
- **Human-in-the-loop** — `human_input=True` pausa e devolve o trabalho ao agente com as notas do revisor; revisores plugáveis para rodar fora do terminal
- **Reasoning e Planning** — `Agent(reasoning=True)` planeja e critica o próprio plano antes de agir; `Crew(planning=True)` planeja todas as tarefas antes de começar
- **Manager dedicado** — `manager_agent` / `manager_llm` delega dinamicamente ao especialista certo e devolve o trabalho quando reprova
- **Guardrails com juiz LLM** — Critérios em linguagem natural; a rejeição volta como feedback para o agente corrigir, em vez de repetir o mesmo prompt
- **Multimodal** — `Agent(multimodal=True)` aceita imagens junto do texto
- **Observabilidade** — OpenTelemetry, Langfuse, MLflow e Arize Phoenix; um `trace_id` único costura a execução inteira, inclusive entre threads
### Compatibilidade com CrewAI
O Mangaba implementa as principais primitivas do CrewAI OSS, com a mesma nomenclatura quando aplicável:
- **Crew** — `akickoff()`, `kickoff_for_each()`, `replay(task_id)`, `train()`/`test()`, `before/after_kickoff_callbacks`, `output_log_file`, `usage_metrics`/`token_usage`, `CrewOutput(raw, pydantic, json_dict)`, `function_calling_llm`
- **Task** — `output_pydantic`/`output_json`, `guardrail` (singular), `max_retries`, `depends_on`, `condition`
- **Agent** — `max_iter`, `max_retry_limit`, `kickoff()` direto, `step_callback` por etapa
- **Tools** — `arun()` assíncrono, `cache_function`, `max_retries`, `max_usage_count`
- **Memória multiusuário** — `Memory(user_id=..., namespace=...)` com isolamento por usuário, `remember/recall/forget/remember_many`, recall profundo (`depth="deep"`) e provedor externo `Mem0Memory`
- **Enterprise leve** — Redação de PII (`redact_pii`), auditoria em JSONL com RBAC (`AuditLogger`, `Role`) e estimativa de custo/métricas Prometheus (`estimate_cost`, `CrewOutput.cost()`)
### Funcionalidades consolidadas (desde v3.0)
- **OpenRouter Native Support** — Roteamento dinâmico com fallback automático entre modelos
- **Multi-Provider Interoperability** — Misture agentes de diferentes provedores (ex: Gemini + Llama) na mesma Crew
- **ReAct Reasoning** — Loop Thought, Action, Observation com function calling nativo
- **7 Provedores LLM** — Google Gemini, OpenAI GPT, Anthropic Claude, HuggingFace, OpenRouter, Ollama e servidores OpenAI-compatible
- **4 Processos de Crew** — Sequential, Hierarchical, Parallel (asyncio), Consensual
- **Tool System** — Decorator `@tool`, schemas Pydantic, JSON schema automático para LLM
- **RAG Pipeline** — Document loaders, text splitters, embeddings, vector store, retriever
- **Memória** — Curto prazo (sliding window), longo prazo (SQLite), entidades
- **Guardrails** — Validação de tamanho, filtro de conteúdo, schema validation
- **Observabilidade** — EventBus com mais de 20 tipos de evento, callbacks console/arquivo
- **Workflow Engine** — Pipelines com stages sequenciais, paralelos e condicionais
- **Cache e Retry** — Cache LRU + disco (SQLite), retry com backoff exponencial + fallback automático
## Instalação
```bash
# Núcleo: OpenAI, OpenRouter, Ollama e servidores OpenAI-compatible
pip install mangaba
# SDK do provedor que você for usar
pip install "mangaba[google]" # Gemini
pip install "mangaba[anthropic]" # Claude
pip install "mangaba[huggingface]" # HuggingFace Inference
pip install "mangaba[providers]" # todos os provedores + tiktoken
# Tudo (provedores, RAG, vector stores, documentos)
pip install "mangaba[all]"
# Desenvolvimento
git clone https://github.com/Mangaba-ai/mangaba_ai.git && cd mangaba_ai
pip install -e ".[dev]"
```
Requer Python 3.10 ou mais novo.
## Quick Start
### Agente simples com ferramenta
```python
from mangaba.core import Agent, Task, Crew, Process, tool
from mangaba.core.types import LLMConfig
@tool
def search(query: str) -> str:
"""Search the web for information."""
return f"Results for: {query}"
researcher = Agent(
role="Research Analyst",
goal="Find accurate information",
backstory="Expert researcher with 10 years of experience",
tools=[search],
llm_config=LLMConfig(provider="google", model="gemini-2.5-flash", api_key="sua-chave"),
)
task = Task(
description="Research the latest AI trends in 2026",
expected_output="A list of the top 5 trends with explanations",
agent=researcher,
)
crew = Crew(
agents=[researcher],
tasks=[task],
process=Process.SEQUENTIAL,
)
result = crew.kickoff()
print(result.final_output)
```
### Multi-Provider Crew com fallback
```python
from mangaba.core import Agent, Task, Crew
from mangaba.core.crew import Process
from mangaba.core.types import OpenRouterConfig, LLMConfig
pesquisador = Agent(
role="Pesquisador",
goal="Analisar vulnerabilidades",
llm_config=OpenRouterConfig(
provider="openrouter",
model=[
"google/gemini-2.5-flash",
"anthropic/claude-3.5-sonnet"
],
api_key="SUA_KEY"
)
)
revisor = Agent(
role="Revisor",
goal="Revisar análise técnica",
llm_config=LLMConfig(
provider="hf",
model="meta-llama/Meta-Llama-3-8B-Instruct",
api_key="SUA_KEY"
)
)
task = Task(
description="Explique buffer overflow",
expected_output="Análise técnica detalhada",
agent=pesquisador,
)
review = Task(
description="Revise a análise",
expected_output="Pontos fortes e fracos",
agent=revisor,
)
crew = Crew(
agents=[pesquisador, revisor],
tasks=[task, review],
process=Process.SEQUENTIAL,
)
result = crew.kickoff()
print(result.final_output)
```
### Pipeline com stages
```python
from mangaba import Pipeline, Stage, ParallelStage
pipeline = Pipeline(stages=[
Stage("research", [research_task]),
ParallelStage("analysis", [task_a, task_b]),
Stage("report", [write_task]),
])
result = pipeline.run({"topic": "AI"})
```
### RAG (Retrieval-Augmented Generation)
```python
from mangaba.rag import TextLoader, RecursiveTextSplitter, RAGChain, Retriever
from mangaba.embeddings import OpenAIEmbedding
from mangaba.vectorstores import InMemoryVectorStore
docs = TextLoader("data.txt").load()
chunks = RecursiveTextSplitter(chunk_size=500).split_documents(docs)
embedding = OpenAIEmbedding(api_key="YOUR_KEY")
store = InMemoryVectorStore(embedding)
store.add(chunks)
retriever = Retriever(embedding=embedding, vector_store=store)
from mangaba.core.llm import create_llm_client
llm = create_llm_client(provider="google", api_key="YOUR_KEY")
chain = RAGChain(llm=llm, retriever=retriever)
answer = chain.query("What are the main topics?")
```
### Memória persistente
```python
from mangaba.memory import ShortTermMemory, LongTermMemory
short = ShortTermMemory(max_items=50)
short.add("User asked about Python")
long_mem = LongTermMemory(storage_path="memory.db")
long_mem.add("User prefers concise answers")
results = long_mem.search("preferences")
```
### Guardrails e Output Parsers
```python
from mangaba import Agent, Task
from mangaba.core.guardrails import LengthGuardrail, GuardrailChain
from mangaba.core.output_parsers import JSONOutputParser
task = Task(
description="List the top 3 programming languages",
expected_output="JSON with name and reason",
agent=agent,
guardrails=[LengthGuardrail(max_length=2000)],
output_parser=JSONOutputParser(),
)
```
### Comunicação entre agentes (protocolo A2A interno)
```python
from protocols.a2a import A2AProtocol, A2AMessage
protocolo = A2AProtocol()
# Enviar mensagem de um agente para outro
msg = A2AMessage(
sender="pesquisador",
recipient="revisor",
content="Análise de vulnerabilidade concluída",
msg_type="request"
)
protocolo.send(msg)
# Broadcast para todos os agentes
protocolo.broadcast(A2AMessage(
sender="manager",
recipient="*",
content="Iniciando nova tarefa",
msg_type="broadcast"
))
```
### Vector store com ChromaDB
```python
from mangaba.vectorstores import ChromaVectorStore, create_vectorstore
from mangaba.embeddings import OpenAIEmbedding
embedding = OpenAIEmbedding(api_key="KEY")
# Via factory
store = create_vectorstore("chroma", embedding=embedding, persist_directory="./chroma_db")
# Adicionar e buscar
store.add(chunks)
results = store.similarity_search("machine learning", k=5)
```
## Flows — orquestração orientada a eventos
Crews coordenam agentes; Flows coordenam tudo o mais. Cada método reage ao
resultado de outro, o estado é tipado, e o progresso pode ser persistido para
retomar de onde parou.
```python
from pydantic import BaseModel
from mangaba import Flow, start, listen, router, persist, and_
class Estado(BaseModel):
texto: str = ""
risco: str = ""
@persist # checkpoint a cada passo
class Triagem(Flow):
state_model = Estado
@start()
def coletar(self):
self.state.texto = "contrato para revisar"
return self.state.texto
@listen(coletar)
def analisar(self, texto):
return "alto" if "contrato" in texto else "baixo"
@router(analisar)
def rotear(self, risco):
self.state.risco = risco
return "juridico" if risco == "alto" else "arquivo"
@listen("juridico")
def escalar(self):
return "enviado ao jurídico"
@listen("arquivo")
def arquivar(self):
return "arquivado"
flow = Triagem()
print(flow.kickoff()) # "enviado ao jurídico"
print(flow.usage_metrics) # tokens somados de tudo que rodou dentro
flow.plot("triagem.html") # gráfico de execução, HTML autocontido
```
`and_(a, b)` dispara só quando ambos terminam, `or_(a, b)` quando qualquer um
terminar. Métodos podem ser `async def`. Para retomar após uma queda:
```python
resultado = Triagem().resume("6f1c...") # pula os passos já concluídos
ramo = Triagem().fork("6f1c...") # ramifica a partir do checkpoint
```
## Knowledge — fundamentar respostas em documentos
Diferente da memória (o que o agente viveu), a Knowledge é o que ele pode
consultar. Aceita PDF, DOCX, Excel, JSON, CSV, URL e diretórios inteiros.
```python
from mangaba import Agent, Knowledge, PDFKnowledgeSource, DirectoryKnowledgeSource
from mangaba.embeddings import OpenAIEmbedding
conhecimento = Knowledge(
embedding=OpenAIEmbedding(api_key="..."),
sources=[
PDFKnowledgeSource(file_path="normas/nr12.pdf"),
DirectoryKnowledgeSource(path="politicas/", recursive=True),
],
results_limit=3,
score_threshold=0.35,
)
agente = Agent(role="Auditor", goal="Verificar conformidade",
backstory="Especialista em segurança do trabalho",
knowledge=conhecimento)
```
Passe `knowledge=` para a `Crew` e todos os agentes que não têm base própria
recebem a mesma.
## Memória com recall semântico
A classe `Memory` pontua cada lembrança por similaridade, recência e
importância — não só por parecido.
```python
from mangaba import Memory, MemoryScope
memoria = Memory(embedding=embedding, db_path=".mangaba/memoria.db")
memoria.add("O cliente prefere respostas curtas", metadata={"importance": 0.9})
memoria.add_interaction("Qual meu plano?", "Você está no plano Pro anual.")
memoria.consolidate() # funde duplicatas
memoria.search("preferências", scope=MemoryScope.AGENT)
agente = Agent(role="Suporte", goal="Atender bem", backstory="...", memory=memoria)
```
As classes antigas (`ShortTermMemory`, `LongTermMemory`, `EntityMemory`)
continuam funcionando sem mudanças.
Para produção multiusuário, isole por usuário e use recall profundo:
```python
from mangaba.memory import Memory
from mangaba.memory.external import Mem0Memory
memoria = Memory(embedding=embedding, user_id="cliente-123", namespace="suporte")
memoria.remember("O cliente prefere respostas curtas")
memoria.recall("preferências do cliente", depth="deep")
# Memória externa (requer: pip install mem0ai)
externa = Mem0Memory(config={"user_id": "cliente-123"})
crew = Crew(agents=[...], tasks=[...], memory=externa)
```
## Human-in-the-loop
`human_input=True` pausa de verdade: o revisor aprova, reescreve, ou devolve
com notas — e as notas voltam para o agente corrigir.
```python
from mangaba import Task, CallbackHumanInput, HumanFeedback
def revisar_no_slack(descricao, saida, papel):
resposta = slack.perguntar(f"{papel} respondeu:\n{saida}")
return HumanFeedback(approved=resposta == "ok", feedback=resposta)
task = Task(
description="Redigir a resposta ao cliente",
expected_output="Um e-mail pronto para enviar",
agent=redator,
human_input=True,
human_input_handler=CallbackHumanInput(revisar_no_slack),
max_human_iterations=3,
)
```
Sem handler, usa o terminal — e aprova sozinho quando não há terminal, para
não travar execuções automatizadas.
## Reasoning, Planning e manager dedicado
```python
# O agente planeja e critica o próprio plano antes de agir
agente = Agent(role="Analista", goal="...", backstory="...",
reasoning=True, max_reasoning_attempts=3)
# A crew planeja todas as tarefas antes de começar, e um manager
# dedicado escolhe quem faz o quê — e reprova o que não serve
crew = Crew(
agents=[pesquisador, analista, redator],
tasks=[t1, t2, t3],
process=Process.HIERARCHICAL,
planning=True,
manager_llm=create_llm_client(provider="openai", api_key="..."),
)
```
## Guardrails com juiz LLM
```python
from mangaba import LLMGuardrail, FunctionGuardrail
task = Task(
description="Resumir o relatório",
expected_output="Um resumo com fontes",
agent=agente,
guardrails=[LLMGuardrail("Deve citar ao menos duas fontes com URL", llm=llm)],
guardrail_max_retries=3, # a rejeição volta como feedback, não como repetição
)
```
## CLI
```bash
mangaba create crew pesquisa # esqueleto com agents.yaml e tasks.yaml
cd pesquisa && cp .env.example .env
mangaba run --input topic="energia solar em Alagoas"
mangaba test -n 3 # nota de qualidade por tarefa e por agente
mangaba train -n 5 # loop de feedback humano, salvo em pickle
mangaba chat # REPL contra um agente
mangaba config # mostra provedor/modelo (nunca a chave)
mangaba reset-memories --all
mangaba replay -t