Building a Multi-Node Graph-Based AI Agent Framework for Complex Task Automation

Spread the love

In this tutorial, we guide you through the development of an advanced graph agent framework, which is operated by Google Gemini API. Our goal is to create intelligent, multi-step agents that execute tasks through a well-defined graph structure of interconnected nodes. Each node represents a specific function, in which taking input, logical processing, making decision making and output. We use python for visualization, networkx for graph modeling and matplletlib. By the end, we apply and run two full examples, to resolve a research assistant and a problem, to display how the framework can handle the complex argument workflow.

!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)

We begin by installing the libraries required to support our graph-based agent framework, Google-Generaventivai, Networkx and Matplotlib. After importing the required module, we configure Gemini API using our API key to enable powerful material generation capabilities within our agent system.

Check it Codes,

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] = None

We define a nodatype calculation to classify input, procedure, decision and output. Then, using a Dataklass agentnode, we structure each node with an ID, type, quick, alternative function and a list of dependence, allowing us to create a modular and flexible agent graph.

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 agent

We make a research agent by adding special nodes to the graph. Starting with a theme input, we define a process flow that includes planning, literature reviews and analysis. The agent then makes a quality decision based on the study and at the end creates a comprehensive research report, which captures the full life cycle of a structured research workflow.

Check it Codes,

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 agent

We create an agent solving a problem by defining the logical sequence of nodes starting from the reception of the statement of the problem. The agent analyzes the problem, produces several solution approaches, evaluates them based on viability and effectiveness, and ends by creating a structured implementation plan, which enables automatic, step-by-step resolution of the problem.

Check it Codes,

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!")

We eliminate tutorials by running two powerful demo agents, one for research and another for problems. In each case, we imagine the graph structure, insyllas the input, and execute the agent node-by-node using a topological order. With Gemini relevant reactions at each step, we see how each agent automatically moves through planning, analysis, decision making and output generations, eventually performs the full potential of our graph-based structure.

Finally, we successfully develop and execute intelligent agents that using a graph-powered architecture, break the tasks and solve step-by-step. We see how each node processes reference-dependent signs, avails Gemini’s capabilities for production, and passes the results of the latter nodes. This modular design enhances flexibility and allows us to clearly visualize logic flow.

Check it CodesAll credit for this research goes to the researchers of this project. Subscribe now To our AI newsletter

Asif razzaq is CEO of Marktechpost Media Inc .. As a visionary entrepreneur and engineer, ASIF is committed to using the ability of artificial intelligence for social good. His most recent effort is the launch of an Artificial Intelligence Media Platform, Marktekpost, which stands for his intensive coverage of machine learning and deep learning news, technically sound and easily understand by a comprehensive audience. The stage claims more than 2 million monthly ideas, reflecting its popularity among the audience.

Source link