En este tutorial, lo guiamos a través del desarrollo de un marco de agente gráfico avanzado, impulsado por la API de Google Gemini. Nuestro objetivo es construir agentes inteligentes de varios pasos que ejecuten tareas a través de una estructura gráfica bien definida de nodos interconectados. Cada nodo representa una función específica, que va desde tomar la entrada, realizar el procesamiento lógico, tomar decisiones y producir salidas. Utilizamos Python, NetworkX para el modelado de gráficos y Matplotlib para la visualización. Al final, implementamos y ejecutamos dos ejemplos completos, un asistente de investigación y un solucionador de problemas, para demostrar cómo el marco puede manejar eficientemente los flujos de trabajo de razonamiento complejos.
!pip install -q google-generativeai networkx matplotlib
import google.generativeai as genai
import networkx as nx
import matplotlib.pyplot as plt
from typing import Dict, List, Any, Callable
import json
import asyncio
from dataclasses import dataclass
from enum import Enum
API_KEY = "use your API key here"
genai.configure(api_key=API_KEY)
Comenzamos instalando las bibliotecas necesarias, Google-Generativeai, NetworkX y Matplotlib, para admitir nuestro marco de agentes basado en gráficos. Después de importar módulos esenciales, configuramos la API de Gemini utilizando nuestra clave API para habilitar potentes capacidades de generación de contenido dentro de nuestro sistema de agentes.
Mira el Codos.
class NodeType(Enum):
INPUT = "input"
PROCESS = "process"
DECISION = "decision"
OUTPUT = "output"
@dataclass
class AgentNode:
id: str
type: NodeType
prompt: str
function: Callable = None
dependencies: List[str] = NoneDefinimos una enumeración de tipo noDety para clasificar diferentes tipos de nodos de agentes: entrada, proceso, decisión y salida. Luego, utilizando un datacLass AgentNode, estructuramos cada nodo con un ID, tipo, inmediato, función opcional y una lista de dependencias, lo que nos permite construir un gráfico de agente modular y flexible.
def create_research_agent():
agent = GraphAgent()
# Input node
agent.add_node(AgentNode(
id="topic_input",
type=NodeType.INPUT,
prompt="Research topic input"
))
agent.add_node(AgentNode(
id="research_plan",
type=NodeType.PROCESS,
prompt="Create a comprehensive research plan for the topic. Include 3-5 key research questions and methodology.",
dependencies=["topic_input"]
))
agent.add_node(AgentNode(
id="literature_review",
type=NodeType.PROCESS,
prompt="Conduct a thorough literature review. Identify key papers, theories, and current gaps in knowledge.",
dependencies=["research_plan"]
))
agent.add_node(AgentNode(
id="analysis",
type=NodeType.PROCESS,
prompt="Analyze the research findings. Identify patterns, contradictions, and novel insights.",
dependencies=["literature_review"]
))
agent.add_node(AgentNode(
id="quality_check",
type=NodeType.DECISION,
prompt="Evaluate research quality. Is the analysis comprehensive? Are there missing perspectives? Return 'APPROVED' or 'NEEDS_REVISION' with reasons.",
dependencies=["analysis"]
))
agent.add_node(AgentNode(
id="final_report",
type=NodeType.OUTPUT,
prompt="Generate a comprehensive research report with executive summary, key findings, and recommendations.",
dependencies=["quality_check"]
))
return agentCreamos un agente de investigación agregando secuencialmente nodos especializados al gráfico. Comenzando con una entrada de tema, definimos un flujo de proceso que incluye planificación, revisión de la literatura y análisis. Luego, el agente toma una decisión de calidad basada en el estudio y finalmente genera un informe de investigación integral, capturando el ciclo de vida completo de un flujo de trabajo de investigación estructurado.
Mira el Codos.
def create_problem_solver():
agent = GraphAgent()
agent.add_node(AgentNode(
id="problem_input",
type=NodeType.INPUT,
prompt="Problem statement"
))
agent.add_node(AgentNode(
id="problem_analysis",
type=NodeType.PROCESS,
prompt="Break down the problem into components. Identify constraints and requirements.",
dependencies=["problem_input"]
))
agent.add_node(AgentNode(
id="solution_generation",
type=NodeType.PROCESS,
prompt="Generate 3 different solution approaches. For each, explain the methodology and expected outcomes.",
dependencies=["problem_analysis"]
))
agent.add_node(AgentNode(
id="solution_evaluation",
type=NodeType.DECISION,
prompt="Evaluate each solution for feasibility, cost, and effectiveness. Rank them and select the best approach.",
dependencies=["solution_generation"]
))
agent.add_node(AgentNode(
id="implementation_plan",
type=NodeType.OUTPUT,
prompt="Create a detailed implementation plan with timeline, resources, and success metrics.",
dependencies=["solution_evaluation"]
))
return agentCreamos un agente de resolución de problemas definiendo una secuencia lógica de nodos, comenzando desde la recepción de la declaración del problema. El agente analiza el problema, genera múltiples enfoques de solución, los evalúa en función de la viabilidad y la efectividad, y concluye produciendo un plan de implementación estructurado, que permite una resolución automatizada y paso a paso del problema.
Mira el Codos.
def run_research_demo():
"""Run the research agent demo"""
print("🚀 Advanced Graph Agent Framework Demo")
print("=" * 50)
research_agent = create_research_agent()
print("\n📊 Research Agent Graph Structure:")
research_agent.visualize()
print("\n🔍 Executing Research Task...")
research_agent.results["topic_input"] = "Artificial Intelligence in Healthcare"
execution_order = list(nx.topological_sort(research_agent.graph))
for node_id in execution_order:
if node_id == "topic_input":
continue
context = {}
node = research_agent.nodes[node_id]
if node.dependencies:
for dep in node.dependencies:
context[dep] = research_agent.results.get(dep, "")
prompt = node.prompt
if context:
context_str = "\n".join([f"{k}: {v}" for k, v in context.items()])
prompt = f"Context:\n{context_str}\n\nTask: {prompt}"
try:
response = research_agent.model.generate_content(prompt)
result = response.text.strip()
research_agent.results[node_id] = result
print(f"✓ {node_id}: {result[:100]}...")
except Exception as e:
research_agent.results[node_id] = f"Error: {str(e)}"
print(f"✗ {node_id}: Error - {str(e)}")
print("\n📋 Research Results:")
for node_id, result in research_agent.results.items():
print(f"\n{node_id.upper()}:")
print("-" * 30)
print(result)
return research_agent.results
def run_problem_solver_demo():
"""Run the problem solver demo"""
print("\n" + "=" * 50)
problem_solver = create_problem_solver()
print("\n🛠️ Problem Solver Graph Structure:")
problem_solver.visualize()
print("\n⚙️ Executing Problem Solving...")
problem_solver.results["problem_input"] = "How to reduce carbon emissions in urban transportation"
execution_order = list(nx.topological_sort(problem_solver.graph))
for node_id in execution_order:
if node_id == "problem_input":
continue
context = {}
node = problem_solver.nodes[node_id]
if node.dependencies:
for dep in node.dependencies:
context[dep] = problem_solver.results.get(dep, "")
prompt = node.prompt
if context:
context_str = "\n".join([f"{k}: {v}" for k, v in context.items()])
prompt = f"Context:\n{context_str}\n\nTask: {prompt}"
try:
response = problem_solver.model.generate_content(prompt)
result = response.text.strip()
problem_solver.results[node_id] = result
print(f"✓ {node_id}: {result[:100]}...")
except Exception as e:
problem_solver.results[node_id] = f"Error: {str(e)}"
print(f"✗ {node_id}: Error - {str(e)}")
print("\n📋 Problem Solving Results:")
for node_id, result in problem_solver.results.items():
print(f"\n{node_id.upper()}:")
print("-" * 30)
print(result)
return problem_solver.results
print("🎯 Running Research Agent Demo:")
research_results = run_research_demo()
print("\n🎯 Running Problem Solver Demo:")
problem_results = run_problem_solver_demo()
print("\n✅ All demos completed successfully!")Concluimos el tutorial ejecutando dos poderosos agentes de demostración, uno para la investigación y otro para la resolución de problemas. En cada caso, visualizamos la estructura del gráfico, inicializamos la entrada y ejecutamos el nodo del agente por nodo utilizando un orden topológico. Con Gemini que generan respuestas contextuales en cada paso, observamos cómo cada agente progresa de forma autónoma a través de la planificación, el análisis, la toma de decisiones y la generación de salida, que finalmente muestra todo el potencial de nuestro marco basado en gráficos.
En conclusión, desarrollamos y ejecutamos con éxito agentes inteligentes que descomponen y resuelven tareas paso a paso, utilizando una arquitectura basada en gráficos. Vemos cómo cada nodo procesa las indicaciones dependientes del contexto, aprovecha las capacidades de Gemini para la generación de contenido y pasa los resultados a los nodos posteriores. Este diseño modular mejora la flexibilidad y también nos permite visualizar claramente el flujo lógico.
Mira el Codos. Todo el crédito por esta investigación va a los investigadores de este proyecto. Suscríbete ahora a nuestro boletín de IA
Asif Razzaq es el CEO de MarktechPost Media Inc .. Como empresario e ingeniero visionario, ASIF se compromete a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, MarktechPost, que se destaca por su cobertura profunda de noticias de aprendizaje automático y de aprendizaje profundo que es técnicamente sólido y fácilmente comprensible por una audiencia amplia. La plataforma cuenta con más de 2 millones de vistas mensuales, ilustrando su popularidad entre el público.
