On this tutorial, we information you thru the event of a complicated Graph Agent framework, powered by the Google Gemini API. Our objective is to construct clever, multi-step brokers that execute duties via a well-defined graph construction of interconnected nodes. Every node represents a selected perform, starting from taking enter, performing logical processing, making choices, and producing outputs. We use Python, NetworkX for graph modeling, and matplotlib for visualization. By the tip, we implement and run two full examples, a Analysis Assistant and a Drawback Solver, to exhibit how the framework can effectively deal with advanced reasoning workflows.
!pip set up -q google-generativeai networkx matplotlib
import google.generativeai as genai
import networkx as nx
import matplotlib.pyplot as plt
from typing import Dict, Record, Any, Callable
import json
import asyncio
from dataclasses import dataclass
from enum import Enum
API_KEY = "use your API key right here"
genai.configure(api_key=API_KEY)
We start by putting in the required libraries, google-generativeai, networkx, and matplotlib, to help our graph-based agent framework. After importing important modules, we configure the Gemini API utilizing our API key to allow highly effective content material era capabilities inside our agent system.
Take a look at the Codes.
class NodeType(Enum):
INPUT = "enter"
PROCESS = "course of"
DECISION = "determination"
OUTPUT = "output"
@dataclass
class AgentNode:
id: str
sort: NodeType
immediate: str
perform: Callable = None
dependencies: Record[str] = NoneWe outline a NodeType enumeration to categorise completely different sorts of agent nodes: enter, course of, determination, and output. Then, utilizing a dataclass AgentNode, we construction every node with an ID, sort, immediate, elective perform, and a listing of dependencies, permitting us to construct a modular and versatile agent graph.
def create_research_agent():
agent = GraphAgent()
# Enter node
agent.add_node(AgentNode(
id="topic_input",
sort=NodeType.INPUT,
immediate="Analysis matter enter"
))
agent.add_node(AgentNode(
id="research_plan",
sort=NodeType.PROCESS,
immediate="Create a complete analysis plan for the subject. Embrace 3-5 key analysis questions and methodology.",
dependencies=["topic_input"]
))
agent.add_node(AgentNode(
id="literature_review",
sort=NodeType.PROCESS,
immediate="Conduct an intensive literature assessment. Determine key papers, theories, and present gaps in data.",
dependencies=["research_plan"]
))
agent.add_node(AgentNode(
id="evaluation",
sort=NodeType.PROCESS,
immediate="Analyze the analysis findings. Determine patterns, contradictions, and novel insights.",
dependencies=["literature_review"]
))
agent.add_node(AgentNode(
id="quality_check",
sort=NodeType.DECISION,
immediate="Consider analysis high quality. Is the evaluation complete? Are there lacking views? Return 'APPROVED' or 'NEEDS_REVISION' with causes.",
dependencies=["analysis"]
))
agent.add_node(AgentNode(
id="final_report",
sort=NodeType.OUTPUT,
immediate="Generate a complete analysis report with govt abstract, key findings, and suggestions.",
dependencies=["quality_check"]
))
return agentWe create a analysis agent by sequentially including specialised nodes to the graph. Beginning with a subject enter, we outline a course of movement that features planning, literature assessment, and evaluation. The agent then makes a top quality determination based mostly on the examine and at last generates a complete analysis report, capturing the total lifecycle of a structured analysis workflow.
Take a look at the Codes.
def create_problem_solver():
agent = GraphAgent()
agent.add_node(AgentNode(
id="problem_input",
sort=NodeType.INPUT,
immediate="Drawback assertion"
))
agent.add_node(AgentNode(
id="problem_analysis",
sort=NodeType.PROCESS,
immediate="Break down the issue into elements. Determine constraints and necessities.",
dependencies=["problem_input"]
))
agent.add_node(AgentNode(
id="solution_generation",
sort=NodeType.PROCESS,
immediate="Generate 3 completely different resolution approaches. For every, clarify the methodology and anticipated outcomes.",
dependencies=["problem_analysis"]
))
agent.add_node(AgentNode(
id="solution_evaluation",
sort=NodeType.DECISION,
immediate="Consider every resolution for feasibility, value, and effectiveness. Rank them and choose the very best strategy.",
dependencies=["solution_generation"]
))
agent.add_node(AgentNode(
id="implementation_plan",
sort=NodeType.OUTPUT,
immediate="Create an in depth implementation plan with timeline, assets, and success metrics.",
dependencies=["solution_evaluation"]
))
return agentWe construct a problem-solving agent by defining a logical sequence of nodes, ranging from the reception of the issue assertion. The agent analyzes the issue, generates a number of resolution approaches, evaluates them based mostly on feasibility and effectiveness, and concludes by producing a structured implementation plan, enabling automated, step-by-step decision of the issue.
Take a look at the Codes.
def run_research_demo():
"""Run the analysis agent demo"""
print("🚀 Superior Graph Agent Framework Demo")
print("=" * 50)
research_agent = create_research_agent()
print("n📊 Analysis Agent Graph Construction:")
research_agent.visualize()
print("n🔍 Executing Analysis Activity...")
research_agent.outcomes["topic_input"] = "Synthetic Intelligence in Healthcare"
execution_order = listing(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, "")
immediate = node.immediate
if context:
context_str = "n".be part of([f"{k}: {v}" for k, v in context.items()])
immediate = f"Context:n{context_str}nnTask: {immediate}"
attempt:
response = research_agent.mannequin.generate_content(immediate)
outcome = response.textual content.strip()
research_agent.outcomes[node_id] = outcome
print(f"✓ {node_id}: {outcome[:100]}...")
besides Exception as e:
research_agent.outcomes[node_id] = f"Error: {str(e)}"
print(f"✗ {node_id}: Error - {str(e)}")
print("n📋 Analysis Outcomes:")
for node_id, end in research_agent.outcomes.gadgets():
print(f"n{node_id.higher()}:")
print("-" * 30)
print(outcome)
return research_agent.outcomes
def run_problem_solver_demo():
"""Run the issue solver demo"""
print("n" + "=" * 50)
problem_solver = create_problem_solver()
print("n🛠️ Drawback Solver Graph Construction:")
problem_solver.visualize()
print("n⚙️ Executing Drawback Fixing...")
problem_solver.outcomes["problem_input"] = "Find out how to cut back carbon emissions in city transportation"
execution_order = listing(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, "")
immediate = node.immediate
if context:
context_str = "n".be part of([f"{k}: {v}" for k, v in context.items()])
immediate = f"Context:n{context_str}nnTask: {immediate}"
attempt:
response = problem_solver.mannequin.generate_content(immediate)
outcome = response.textual content.strip()
problem_solver.outcomes[node_id] = outcome
print(f"✓ {node_id}: {outcome[:100]}...")
besides Exception as e:
problem_solver.outcomes[node_id] = f"Error: {str(e)}"
print(f"✗ {node_id}: Error - {str(e)}")
print("n📋 Drawback Fixing Outcomes:")
for node_id, end in problem_solver.outcomes.gadgets():
print(f"n{node_id.higher()}:")
print("-" * 30)
print(outcome)
return problem_solver.outcomes
print("🎯 Operating Analysis Agent Demo:")
research_results = run_research_demo()
print("n🎯 Operating Drawback Solver Demo:")
problem_results = run_problem_solver_demo()
print("n✅ All demos accomplished efficiently!")We conclude the tutorial by operating two highly effective demo brokers, one for analysis and one other for problem-solving. In every case, we visualize the graph construction, initialize the enter, and execute the agent node-by-node utilizing a topological order. With Gemini producing contextual responses at each step, we observe how every agent autonomously progresses via planning, evaluation, decision-making, and output era, in the end showcasing the total potential of our graph-based framework.
In conclusion, we efficiently developed and executed clever brokers that break down and remedy duties step-by-step, using a graph-driven structure. We see how every node processes context-dependent prompts, leverages Gemini’s capabilities for content material era, and passes outcomes to subsequent nodes. This modular design enhances flexibility and in addition permits us to visualise the logic movement clearly.
Take a look at the Codes. All credit score for this analysis goes to the researchers of this mission. SUBSCRIBE NOW to our AI E-newsletter
Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is dedicated to harnessing the potential of Synthetic Intelligence for social good. His most up-to-date endeavor is the launch of an Synthetic Intelligence Media Platform, Marktechpost, which stands out for its in-depth protection of machine studying and deep studying information that’s each technically sound and simply comprehensible by a large viewers. The platform boasts of over 2 million month-to-month views, illustrating its reputation amongst audiences.
