On this tutorial, we information you thru the event of a sophisticated Graph Agent framework, powered by the Google Gemini API. Our purpose is to construct clever, multi-step brokers that execute duties via a well-defined graph construction of interconnected nodes. Every node represents a particular 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 top, we implement and run two full examples, a Analysis Assistant and a Downside Solver, to show how the framework can effectively deal with complicated 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, Listing, 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 technology capabilities inside our agent system.
Try the Codes.
class NodeType(Enum):
INPUT = "enter"
PROCESS = "course of"
DECISION = "determination"
OUTPUT = "output"
@dataclass
class AgentNode:
id: str
kind: NodeType
immediate: str
perform: Callable = None
dependencies: Listing[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, kind, immediate, optionally available perform, and an inventory 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",
kind=NodeType.INPUT,
immediate="Analysis matter enter"
))
agent.add_node(AgentNode(
id="research_plan",
kind=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",
kind=NodeType.PROCESS,
immediate="Conduct an intensive literature overview. Determine key papers, theories, and present gaps in information.",
dependencies=["research_plan"]
))
agent.add_node(AgentNode(
id="evaluation",
kind=NodeType.PROCESS,
immediate="Analyze the analysis findings. Determine patterns, contradictions, and novel insights.",
dependencies=["literature_review"]
))
agent.add_node(AgentNode(
id="quality_check",
kind=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",
kind=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 overview, and evaluation. The agent then makes a top quality determination primarily based on the research and eventually generates a complete analysis report, capturing the complete lifecycle of a structured analysis workflow.
Try the Codes.
def create_problem_solver():
agent = GraphAgent()
agent.add_node(AgentNode(
id="problem_input",
kind=NodeType.INPUT,
immediate="Downside assertion"
))
agent.add_node(AgentNode(
id="problem_analysis",
kind=NodeType.PROCESS,
immediate="Break down the issue into parts. Determine constraints and necessities.",
dependencies=["problem_input"]
))
agent.add_node(AgentNode(
id="solution_generation",
kind=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",
kind=NodeType.DECISION,
immediate="Consider every resolution for feasibility, price, and effectiveness. Rank them and choose the perfect method.",
dependencies=["solution_generation"]
))
agent.add_node(AgentNode(
id="implementation_plan",
kind=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 primarily based on feasibility and effectiveness, and concludes by producing a structured implementation plan, enabling automated, step-by-step decision of the issue.
Try 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 Job...")
research_agent.outcomes["topic_input"] = "Synthetic Intelligence in Healthcare"
execution_order = checklist(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 a part of([f"{k}: {v}" for k, v in context.items()])
immediate = f"Context:n{context_str}nnTask: {immediate}"
strive:
response = research_agent.mannequin.generate_content(immediate)
consequence = response.textual content.strip()
research_agent.outcomes[node_id] = consequence
print(f"✓ {node_id}: {consequence[: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, lead to research_agent.outcomes.objects():
print(f"n{node_id.higher()}:")
print("-" * 30)
print(consequence)
return research_agent.outcomes
def run_problem_solver_demo():
"""Run the issue solver demo"""
print("n" + "=" * 50)
problem_solver = create_problem_solver()
print("n🛠️ Downside Solver Graph Construction:")
problem_solver.visualize()
print("n⚙️ Executing Downside Fixing...")
problem_solver.outcomes["problem_input"] = "Tips on how to cut back carbon emissions in city transportation"
execution_order = checklist(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 a part of([f"{k}: {v}" for k, v in context.items()])
immediate = f"Context:n{context_str}nnTask: {immediate}"
strive:
response = problem_solver.mannequin.generate_content(immediate)
consequence = response.textual content.strip()
problem_solver.outcomes[node_id] = consequence
print(f"✓ {node_id}: {consequence[:100]}...")
besides Exception as e:
problem_solver.outcomes[node_id] = f"Error: {str(e)}"
print(f"✗ {node_id}: Error - {str(e)}")
print("n📋 Downside Fixing Outcomes:")
for node_id, lead to problem_solver.outcomes.objects():
print(f"n{node_id.higher()}:")
print("-" * 30)
print(consequence)
return problem_solver.outcomes
print("🎯 Working Analysis Agent Demo:")
research_results = run_research_demo()
print("n🎯 Working Downside 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 technology, in the end showcasing the complete potential of our graph-based framework.
In conclusion, we efficiently developed and executed clever brokers that break down and resolve 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 technology, and passes outcomes to subsequent nodes. This modular design enhances flexibility and in addition permits us to visualise the logic movement clearly.
Try the Codes. All credit score for this analysis goes to the researchers of this venture. 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 recognition amongst audiences.
