Setting up A Multi-Node Graph-Based mostly Principally AI Agent Framework For Superior Exercise Automation

!pip arrange -q google-generativeai networkx matplotlib


import google.generativeai as genai
import networkx as nx
import matplotlib.pyplot as plt
from typing import Dict, Itemizing, Any, Callable
import json
import asyncio
from dataclasses import dataclass
from enum import Enum


API_KEY = "use your API key proper right here"
genai.configure(api_key=API_KEY)

class NodeType(Enum):
    INPUT = "enter"
    PROCESS = "course of"
    DECISION = "dedication"
    OUTPUT = "output"


@dataclass
class AgentNode:
    id: str
    form: NodeType
    fast: str
    function: Callable = None
    dependencies: Itemizing[str] = None

def create_research_agent():
    agent = GraphAgent()
   
    # Enter node
    agent.add_node(AgentNode(
        id="topic_input",
        form=NodeType.INPUT,
        fast="Evaluation matter enter"
    ))
   
    agent.add_node(AgentNode(
        id="research_plan",
        form=NodeType.PROCESS,
        fast="Create an entire evaluation plan for the topic. Embrace 3-5 key evaluation questions and methodology.",
        dependencies=["topic_input"]
    ))
   
    agent.add_node(AgentNode(
        id="literature_review",
        form=NodeType.PROCESS,
        fast="Conduct a radical literature analysis. Set up key papers, theories, and current gaps in information.",
        dependencies=["research_plan"]
    ))
   
    agent.add_node(AgentNode(
        id="analysis",
        form=NodeType.PROCESS,
        fast="Analyze the evaluation findings. Set up patterns, contradictions, and novel insights.",
        dependencies=["literature_review"]
    ))
   
    agent.add_node(AgentNode(
        id="quality_check",
        form=NodeType.DECISION,
        fast="Contemplate evaluation top quality. Is the analysis full? Are there missing views? Return 'APPROVED' or 'NEEDS_REVISION' with causes.",
        dependencies=["analysis"]
    ))
   
    agent.add_node(AgentNode(
        id="final_report",
        form=NodeType.OUTPUT,
        fast="Generate an entire evaluation report with authorities summary, key findings, and recommendations.",
        dependencies=["quality_check"]
    ))
   
    return agent

def create_problem_solver():
    agent = GraphAgent()
   
    agent.add_node(AgentNode(
        id="problem_input",
        form=NodeType.INPUT,
        fast="Disadvantage assertion"
    ))
   
    agent.add_node(AgentNode(
        id="problem_analysis",
        form=NodeType.PROCESS,
        fast="Break down the difficulty into elements. Set up constraints and requirements.",
        dependencies=["problem_input"]
    ))
   
    agent.add_node(AgentNode(
        id="solution_generation",
        form=NodeType.PROCESS,
        fast="Generate 3 utterly totally different reply approaches. For each, make clear the methodology and anticipated outcomes.",
        dependencies=["problem_analysis"]
    ))
   
    agent.add_node(AgentNode(
        id="solution_evaluation",
        form=NodeType.DECISION,
        fast="Contemplate each reply for feasibility, worth, and effectiveness. Rank them and select the right technique.",
        dependencies=["solution_generation"]
    ))
   
    agent.add_node(AgentNode(
        id="implementation_plan",
        form=NodeType.OUTPUT,
        fast="Create an in depth implementation plan with timeline, sources, and success metrics.",
        dependencies=["solution_evaluation"]
    ))
   
    return agent

def run_research_demo():
    """Run the evaluation agent demo"""
    print("🚀 Superior Graph Agent Framework Demo")
    print("=" * 50)
   
    research_agent = create_research_agent()
    print("n📊 Evaluation Agent Graph Development:")
    research_agent.visualize()
   
    print("n🔍 Executing Evaluation Exercise...")
   
    research_agent.outcomes["topic_input"] = "Artificial Intelligence in Healthcare"
   
    execution_order = itemizing(nx.topological_sort(research_agent.graph))
   
    for node_id in execution_order:
        if node_id == "topic_input":
            proceed
           
        context = {}
        node = research_agent.nodes[node_id]
       
        if node.dependencies:
            for dep in node.dependencies:
                context[dep] = research_agent.outcomes.get(dep, "")
       
        fast = node.fast
        if context:
            context_str = "n".be part of([f"{k}: {v}" for k, v in context.items()])
            fast = f"Context:n{context_str}nnTask: {fast}"
       
        try:
            response = research_agent.model.generate_content(fast)
            finish end result = response.textual content material.strip()
            research_agent.outcomes[node_id] = finish end result
            print(f"✓ {node_id}: {finish end result[:100]}...")
        in addition to Exception as e:
            research_agent.outcomes[node_id] = f"Error: {str(e)}"
            print(f"✗ {node_id}: Error - {str(e)}")
   
    print("n📋 Evaluation Outcomes:")
    for node_id, finish in research_agent.outcomes.objects():
        print(f"n{node_id.increased()}:")
        print("-" * 30)
        print(finish end result)
   
    return research_agent.outcomes


def run_problem_solver_demo():
    """Run the difficulty solver demo"""
    print("n" + "=" * 50)
    problem_solver = create_problem_solver()
    print("n🛠️ Disadvantage Solver Graph Development:")
    problem_solver.visualize()
   
    print("n⚙️ Executing Disadvantage Fixing...")
   
    problem_solver.outcomes["problem_input"] = "The correct method to cut back carbon emissions in metropolis transportation"
   
    execution_order = itemizing(nx.topological_sort(problem_solver.graph))
   
    for node_id in execution_order:
        if node_id == "problem_input":
            proceed
           
        context = {}
        node = problem_solver.nodes[node_id]
       
        if node.dependencies:
            for dep in node.dependencies:
                context[dep] = problem_solver.outcomes.get(dep, "")
       
        fast = node.fast
        if context:
            context_str = "n".be part of([f"{k}: {v}" for k, v in context.items()])
            fast = f"Context:n{context_str}nnTask: {fast}"
       
        try:
            response = problem_solver.model.generate_content(fast)
            finish end result = response.textual content material.strip()
            problem_solver.outcomes[node_id] = finish end result
            print(f"✓ {node_id}: {finish end result[:100]}...")
        in addition to Exception as e:
            problem_solver.outcomes[node_id] = f"Error: {str(e)}"
            print(f"✗ {node_id}: Error - {str(e)}")
   
    print("n📋 Disadvantage Fixing Outcomes:")
    for node_id, finish in problem_solver.outcomes.objects():
        print(f"n{node_id.increased()}:")
        print("-" * 30)
        print(finish end result)
   
    return problem_solver.outcomes


print("🎯 Working Evaluation Agent Demo:")
research_results = run_research_demo()


print("n🎯 Working Disadvantage Solver Demo:")
problem_results = run_problem_solver_demo()


print("n✅ All demos completed effectively!")