Siguiendo con la serie de artículos sobre la orquestación de multiagentes en Strands, le toca el turno a Graph Multi-Agent Pattern, ejemplificado con la construcción de un knowledge graph en base a un input de texto (abstract), donde los nodos de un grafo determinista tienen las responsabilidades divididas para lograrlo. Se probó de dos maneras: con Gemma4 + Ollama y con la API de Gemini en su capa gratuita.
Introducción
La información del presente artículo la podés seguir teniendo alguna noción sobre agentes y grafos. Pero te propongo hacer un curso de fundamentos del framework Strands Agents, para que en futuros proyectos puedas dar mayor complejidad a tus agentes. La idea principal es experimentar con el patrón sin gastar en suscripciones pagas de tokens, dejando la implementación en AWS para otra instancia.
✳ El caso de uso que elegí para representar el patrón Graph es la Extracción de entidades/relaciones para formar un Knowledge Graph con validación. Se me ocurrió hacer algo útil para la comunidad académica, un mecanismo de extracción de conocimiento que sea capaz de tomar parte de un artículo científico e ingestar lo más importante en un documento json, que pueda ser legible por humanos y sistemas. La solución es un pipeline de construcción de knowledge graph a partir de un texto (ej. un abstract), usando la topología Parallel Processing→ with Aggregation→ Feedback Loop.
El diseño es una versión reducida de arquitecturas multi-agente real para enriquecimiento de knowledge graph descritos en:
Conceptos Claves
➡ Graph es un patrón de orquestación de agentes determinista
El desarrollador define explícitamente la topología de ejecución: qué nodos existen, en qué orden corren, y bajo qué condiciones se transita de uno a otro. Cada nodo es un agente y a diferencia de otros patrones como Swarm, donde el flujo surge de la negociación entre agentes, en Graph el flujo está fijado de antemano como un grafo dirigido (DAG o con ciclos controlados).
Graph es apropiado cuando:
- La estructura de dependencias es conocida de antemano: sabés qué pasos necesitan el resultado de otros antes de poder ejecutarse.
- Hay paralelismo real que después converge: varios agentes pueden trabajar al mismo tiempo sobre partes independientes del problema, y sus resultados se consolidan en un punto de agregación explícito.
-
Se necesita revisión con ciclos controlados: un resultado puede requerir corrección y volver a un paso anterior (feedback loop), con límites de seguridad (
set_max_node_executions) que garantizan que el proceso termine. - La trazabilidad y el control son prioritarios: como cada nodo y cada transición son explícitos, es más fácil auditar, debuggear y predecir el comportamiento, es bueno en contextos con validación humana o compliance.
➡ Conceptos genéricos de Grafos Agenticos
El framework propone una interfaz sencilla para la construcción de grafos, con 3 componentes base: GraphNode, GraphEdge, GraphBuilder
1.Los nodos representan agentes, nodos personalizados o sistemas multiagente. Ej:
from strands import Agent
from strands.multiagent import GraphBuilder
# Build the graph
builder = GraphBuilder()
# Add nodes
builder.add_node(agent_1, "[node_id_1]")
builder.add_node(agent_2, "[node_id_2]")
2.Las aristas definen las dependencias y el flujo de información entre los nodos. Además se puede agregar lógica condicional (no lo utilicé en el caso de uso analizado):
# Add edges (dependencies)
builder.add_edge("[node_id_1]", "[node_id_2]")
3.La ejecución sigue la estructura del grafo, respeta dependencias, la salida de un nodo se convierte en la entrada de su dependiente. Y en el caso de que varios nodos tienen aristas hacia un nodo de destino, el comportamiento predeterminado para la ejecución del destino puede variar según el SDK (Python utiliza semánticas OR: el nodo destino se dispara en cuanto cualquiera de sus dependencias completa).
4.Compatibilidad con patrones anidados, un grafo como nodo dentro de otro grafo.
5.Remote Agents vía A2A: Graph soporta agentes remotos como nodos a través de A2AAgent, para arquitecturas distribuidas donde la orquestación corre localmente pero tareas especializadas se ejecutan en servicios remotos.
6.Cyclic graph permite que un nodo sea revisitado dentro del mismo flujo, con límites de ejecución y gestión de estado.
7.Custom Node Types: Graph permite extender MultiAgentBase para crear nodos que ejecutan funciones Python deterministas en lugar de llamar a un LLM, útil para lógica de negocio, validaciones y pipelines de procesamiento de datos dentro del mismo grafo.
8.Graphs as a Tool: mediante el paquete strands_tools, un agente puede recibir el graph como tool y construir/ejecutar su propia topología de nodos en tiempo real, dejando que el LLM decida dinámicamente la orquestación en lugar de definirla de antemano con GraphBuilder.
➡ Topologías comunes
Strands documenta 4 topologías base que cubren la mayoría de casos de uso, y se pueden combinar según la complejidad del problema. Sequential Pipeline sirve cuando cada paso depende estrictamente del anterior. Parallel + Aggregation es ideal cuando hay trabajo independiente que puede correr simultáneamente y luego converger en un solo resultado. Branching Logic permite que el flujo tome caminos distintos según el contenido generado en un paso previo. Y Feedback Loop habilita ciclos de revisión con vuelta atrás, para casos donde un resultado necesita validación antes de darse por definitivo.
➡ Evaluaciones (Evals) Strands cuenta con una marco oficial para realizar evaluaciones de resultados simples hasta el análisis de interacciones de multiagentes complejos.
En el caso de uso para la creación de Knowledge Graph se realizaron 6 evaluaciones (3 sin utilizar el framework).
| Evaluación / Métrica | Tipo / Origen | ¿Qué evalúa? | Escala |
|---|---|---|---|
| OutputEvaluator | LLM-as-a-Judge (strands-agents-evals) |
Verifica que la salida del nodo Publisher sea un JSON válido (entities y relations) y que las relaciones estén fundamentadas fácticamente en el texto original. |
0 / 0.5 / 1 |
| SemanticCorrectnessEvaluator | LLM-as-a-Judge (strands-agents-evals) |
Compara semánticamente el knowledge graph generado contra las relaciones esperadas (ground truth), dando crédito parcial a triples con distinto wording pero igual significado. Implementado como un segundo OutputEvaluator con rúbrica de referencia, ya que el CorrectnessEvaluator oficial requiere un objeto Session de Strands (incompatible con grafos multi-agente). |
0 / 0.5 / 1 |
| Exact-Match (P / R / F1) | Métrica Determinista (metrics.py) |
Mide la coincidencia exacta (normalizada a minúsculas/espacios) entre las triples extraídas y las triples de referencia del caso de prueba. | 0.0 – 1.0 |
| Loop Iterations Count | Métrica de Topología (metrics.py) |
Cuenta cuántas iteraciones de revisión ocurrieron en el feedback loop entre el Validator y el Aggregator antes de la aprobación final. | entero ≥ 0 |
| Bias-Flag Rate | Métrica Determinista (metrics.py) |
Detecta si la respuesta del nodo Validator levantó alguna objeción o alerta de sesgo/desbalance en las relaciones o entidades analizadas. |
true / false
|
| Compound Entity Check | Métrica Determinista (metrics.py) |
Chequeo de regresión: detecta entidades compuestas (ej. "entities and relations") que deberían haberse dividido en dos nodos separados, según la regla en extractor_agent.py/aggregator_agent.py. |
lista de entidades detectadas |
Para clarificar: una triple es la unidad básica de información que compone un Knowledge Graph. Representa una afirmación o hecho estructurado combinando tres elementos:
$$\text{(Sujeto)} \xrightarrow{\quad\text{Relación / Predicado}\quad} \text{(Objeto)}$$
- Sujeto (source): La entidad de origen (ej.
"Laura Bolaños"). - Relación (relation): El tipo de vínculo o conector (ej.
"born_in"). - Objeto (target): La entidad de destino (ej.
"Argentina").
Descripción de la solución
📦 Repositorio GitHub: github.com/reinalau/strands-graph
⭐ El caso de uso combina dos topologías: paralelismo con agregación, seguido de un ciclo de feedback antes de publicar el resultado final. Notar en la representación que Extractor es un agente que se instancia 3 veces, una única definición de agente, instanciada 3 veces (una por chunk_number) dentro del loop en builder.py.
Para la prueba de concepto utilicé dos opciones locales de ejecución que me permitió analizar sus resultados, pero podes utilizar solo una:
a. Ollama + el pequeño modelo gemma4:e2b-it-qat (que pesa poco mas de 4gb) ejecutando en Docker:
# Start the Ollama server with a persistent volume
docker run -d --name ollama -p 11434:11434 -v ollama_data:/root/.ollama ollama/ollama
# Download the model
docker exec -it ollama ollama pull gemma4:e2b-it-qat
# Test that the model responds
# If the container has already been created and
# is currently stopped, simply use: docker start ollama
docker exec -it ollama ollama run gemma4:e2b-it-qat
# Verify that the model is running
docker exec -it ollama ollama ps
b. Api de gemini. La api key se puede generar de manera gratuita desde aquí y utilizar al menos estos modelos (experimentar con los que te permita):
gemini-2.5-flash-lite
gemini-2.5-flash
gemini-3.5-flash
gemini-3.5-flash-lite
El código del multiagente Graph está en Python y la estructura del proyecto se diseñó de manera que sea más explicativo su lógica:
strands-graph/
├── README.md # Project overview, setup, and usage instructions
├── requirements.txt # Python dependencies (strands-agents[gemini,ollama], strands-agents-evals, pydantic, etc.)
├── .env.example # Template for API keys (GEMINI_API_KEY) and Ollama config
├── .gitignore
│
├── src/
│ ├── __init__.py
│ ├── config.py # Model configs: role -> provider mapping (Gemini/Ollama), get_model()
│ ├── main.py # Entry point: builds graph, runs on input, prints/exports result, saves run log
│ │
│ ├── agents/
│ │ ├── __init__.py # Re-exports all create_*_agent() factories
│ │ ├── coordinator_agent.py # Entry point: splits input text into chunks for parallel extraction
│ │ ├── extractor_agent.py # Shared extractor logic, instantiated 3x (one per chunk); no compound entities
│ │ ├── aggregator_agent.py # Conflict resolution: merges, deduplicates, splits any compound entities
│ │ ├── validator_agent.py # structured_output_model: checks format, consistency, completeness, bias
│ │ └── publisher_agent.py # Formats and exports the final validated KG
│ │
│ ├── graph/
│ │ ├── __init__.py
│ │ ├── builder.py # GraphBuilder setup: 7 nodes, edges (incl. aggregator->publisher), entry point
│ │ └── conditions.py # Edge conditions: needs_revision, is_approved, all_dependencies_complete
│ │
│ ├── models/
│ │ └── schemas.py # Pydantic schemas: Relation, KnowledgeGraph, ValidatorOutput
│ │
│ └── utils/
│ ├── logging_config.py # Debug logging + tee_console_to_file (saves full run output to logs/)
│ └── kg_export.py # Parses and exports final KG to JSON
│
├── tests/
│ ├── __init__.py
│ ├── test_conditions.py # Unit tests for all_dependencies_complete, needs_revision, is_approved
│ ├── test_kg_export.py # Unit tests for parse_publisher_output (valid/invalid JSON, schema mismatch)
│ ├── test_metrics.py # Unit tests for evals/metrics.py, incl. find_compound_entities regression check
│ ├── test_schemas.py # Unit tests for Relation/KnowledgeGraph/ValidatorOutput validation
│ └── test_settings.py # Unit tests for get_model role/provider resolution and missing API key error
│
├── examples/
│ └── sample_input.txt # Short input text used as the default demo run
│
├── evals/
│ ├── __init__.py
│ ├── test_cases/
│ │ └── kg_extraction_cases.json # Input texts + expected relations for evaluation
│ ├── evaluators.py # Custom Strands Evals SDK evaluators (precision/recall, feedback loop)
│ ├── evaluate_graph.py # Strands Evals SDK runner: Case/Experiment against test cases
│ └── metrics.py # Domain metrics: precision/recall, loop iterations, bias-flag rate, compound entities
│
├── logs/
│ └── .gitkeep # Full console output per run (logs/run_<timestamp>.log), gitignored otherwise
│
└── outputs/
└── .gitkeep # Generated KG JSON + eval reports land here, gitignored otherwise
Ejecución Local
Una vez que tenemos el repositorio del código fuente clonado y el docker de ollama con el modelo gemma 4 o la api key de gemini generada, pasamos armar el ambiente.
Los requerimientos que vamos a encontrar en requirements.txt:
strands-agents[gemini,ollama]>=1.0.0
strands-agents-tools>=0.1.0
strands-agents-evals
python-dotenv>=1.0.0
pydantic>=2.0.0
pytest
pip install -r requirements.txt
En .env se necesitan estos valores. Notar que depende el modelo que elegimos podemos intercambiar:
#Gemini (required if any role uses provider="gemini") ---
GEMINI_API_KEY=your_gemini_api_key_here
GEMINI_MODEL_ID=gemini-2.5-flash
#Ollama (required if any role uses provider="ollama") ---
OLLAMA_HOST=http://localhost:11434
OLLAMA_MODEL_ID=gemma4:e2b-it-qat
#Per-role provider overrides (optional) ---
#Values: "gemini" or "ollama". Defaults are set in src/config.py.
MODEL_PROVIDER_COORDINATOR=ollama
MODEL_PROVIDER_EXTRACTOR=ollama
MODEL_PROVIDER_AGGREGATOR=ollama
MODEL_PROVIDER_VALIDATOR=ollama
MODEL_PROVIDER_PUBLISHER=ollama
cp .env.example .env
En src/config.py está la configuración de despliegue, en este caso de ejecución local se encuentran parámetros como la temperature, top_p, max_output_tokens (muy importante porque limita los tokens que pueden gastar los agentes).
Las constantes estructurales del grafo a nivel módulo, se encuentran en src/graph/builder.py :
-
NUM_EXTRACTORS→ cantidad de nodos extractores paralelos (y de chunks que pide el Coordinator). Cambiarlo requiere que coincida con elnum_chunkspasado acreate_coordinator_agent(). -
MAX_NODE_EXECUTIONS→ límite global de ejecuciones de nodos en todo el grafo, la salvaguarda contra el feedback loop infinito Validator↔Aggregator. -
EXECUTION_TIMEOUT_SECONDS→ timeout global del grafo completo.
NUM_EXTRACTORS = 3
MAX_NODE_EXECUTIONS = 15
EXECUTION_TIMEOUT_SECONDS = 1200 # (For local execution, it's advisable to have a high value)
Las pruebas están organizadas en tests/ en un solo nivel: ejecución rápida y determinista. Validan la lógica de las funciones de condición de Graph (needs_revision, is_approved, all_dependencies_complete), el parsing/validación de schemas (Relation, KnowledgeGraph, ValidatorOutput).
Se testea, además de los posibles errores del runtime, la resolución de provider/modelo por rol de src/config.py:
-
_resolve_provider()— que cada rol (coordinator,extractor,aggregator,validator,publisher) resuelva al provider correcto: el default (ROLE_PROVIDER_DEFAULTS) cuando no hay override, y la variable de entorno (MODEL_PROVIDER_VALIDATOR=ollama, etc.) cuando sí la hay. -
get_model()— que devuelva una instancia deOllamaModelcuando el provider es"ollama", yGeminiModelcuando es"gemini".
python -m pytest
Procedemos a ejecutar el Graph principal. En examples/sample_input.txtestá el input de ejemplo.
Se puede ejecutar contra el modelo ollama local o vía api de gemini comentado más arriba.
python -m src.main
El log de las llamadas entre el pipeline de agentes se visualiza en la carpeta logs/. Contiene la data cruda que permite comparar comportamiento del modelo (cuántos turnos internos hace cada uno, cuánto tarda, si usa tool calls, etc.).
El resultado final
Es un Knowledge graph y se guarda en un .json enoutputs/.
Se visualiza la representación estructurada del texto de entrada, describiendo: qué conceptos existen (entities) y cómo se relacionan entre sí (relations), en formato sujeto-predicado-objeto.
Por último, hacemos una evaluación de pruebas sobre el grafo multiagente real con el concepto de "Evals". Esto es importante para todos los agentes futuros que construyas, tomar la documentación de Strands y descubrir cómo y para qué tenemos que evaluar nuestros agentes. El resultado se guarda en outputs/eval_report.json (el detalle de las evaluaciones están en "Conceptos Claves"):
python -m evals.evaluate_graph
📝 Nota: La explicación del código fuente y paso a paso de la ejecución se encuentra detallado en el README.md en el repositorio de github .
Conclusiones y lecciones aprendidas
Construir este pipeline con Graph me dejó varias lecciones que van más allá de la forma de crearlo según su documentación:
Las constantes estructurales del grafo no son de uso de común para todos los modelos.
NUM_EXTRACTORS,MAX_NODE_EXECUTIONSyEXECUTION_TIMEOUT_SECONDSenbuilder.pyparecen valores de infraestructura, pero en la práctica están acoplados al comportamiento del modelo elegido. Con Gemma4 local, el cuello de botella fue el tiempo (EXECUTION_TIMEOUT_SECONDStuvo que subir a 1800s). Con Gemini, el cuello de botella fue la convergencia del feedback loop: contemperature=0.3ytop_p=0.9el Aggregator regeneraba las triples con variación de wording en cada ronda, y el Validator nunca encontraba dos corridas iguales para aprobar, agotandoMAX_NODE_EXECUTIONSsin converger. Solución: bajar atemperature=0ytop_p=0.5.Análisis del input propagation de Graph Un nodo sólo recibe el resultado de sus dependencias directas + el input original con el que comienza el grafo. El Publisher, dependiendo solo del Validator, se quedó sin las triples reales porque el Validator fue instruido a responder con una confirmación corta, no a repetir los datos. La solución fue agregar
aggregator → publishercomo dependencia adicional. Hay que diseñar edges, pensando qué datos necesita cada nodo, no solo qué controla el flujo.Las semánticas OR de Python son una fuente de bugs silenciosos. Sin
all_dependencies_complete(), el Aggregator se hubiera ejecutado 3 veces (una por extractor) en lugar de una vez con los 3 resultados completos. Cualquier nodo con múltiples edges de entrada en Python necesita esta verificación explícita si queremos esperar a todas las dependencias.La granularidad de extracción es responsabilidad del prompt, no del framework. Ver entidades compuestas como "entities and relations" en lugar de dos nodos separados es variabilidad normal de un LLM con temperatura > 0. Hay que reforzar reglas en el prompt del Extractor y del Aggregator.
structured_output_modelrequiere que el modelo soporte tool calling (este parámetro se pasa en la creación del Agent. Hay que verificarlo (ollama show <model>→Capabilities: tools) antes de asumir que cualquier modelo local lo soporta.Exact-match y LLM-as-judge miden cosas distintas. En una corrida, el mismo caso obtuvo F1 exact-match = 0.0 pero SemanticCorrectnessEvaluator = 1.0, el pipeline había extraído el 100% del conocimiento correcto, solo con sinónimos distintos ("Aggregator node" vs "Aggregator"). Confiar solo en exact-match hubiera hecho pensar que el pipeline falló por completo.
Top comments (0)