Building an Advanced Graph Agent Framework with the Google Gemini API

In today’s tech landscape, harnessing the capabilities of AI in structured and intelligent ways is more important than ever. This tutorial aims to guide you through the development of an advanced Graph Agent framework powered by the Google Gemini API. Our end goal? To create intelligent, multi-step agents that can execute tasks seamlessly through a graph structure of interconnected nodes. Each node serves a unique purpose—be it taking input, processing information, making informed decisions, or generating outputs. Let’s dive in!

Setting Up the Environment

Before we get started, we’ll need to install several necessary libraries:

bash
!pip install -q google-generativeai networkx matplotlib

Once we have the libraries set up, we’ll import the essential modules and configure the Gemini API using our API key. This setup will enable us to leverage powerful content generation capabilities as part of our agent system.

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

Defining Node Types and Agent Structure

Next, we’ll introduce a straightforward yet effective structure to our agents. We’ll create a NodeType enumeration to categorize various types of nodes such as input, process, decision, and output:

python
class NodeType(Enum):
INPUT = "input"
PROCESS = "process"
DECISION = "decision"
OUTPUT = "output"

Then, using a dataclass called AgentNode, we outline the properties of each node, consisting of an ID, type, prompt, a function (if required), and any dependencies:

python
@dataclass
class AgentNode:
id: str
type: NodeType
prompt: str
function: Callable = None
dependencies: List[str] = None

This modular approach allows us to create versatile agents that can adapt to various tasks.

Creating a Research Agent

Our first example involves creating a research agent, designed to perform a comprehensive investigation on a given topic. Here’s how we build it step-by-step by adding specialized nodes to construct the agent’s workflow:

python
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

This structured approach not only offers a clear path from input to output but also showcases how a research project can be navigated through various stages—from defining questions to producing a final report.

Building a Problem-Solving Agent

Next, we’ll develop a Problem Solver agent, which evaluates and generates solutions to defined problems. We adopt a similar structure, creating a logical sequence of nodes that guide the problem-solving process:

python
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

Here, the agent takes a problem statement as input, analyzes it, suggests multiple solutions, evaluates their effectiveness, and ultimately crafts a detailed implementation plan. This approach emphasizes not just generating solutions but also assessing their viability before moving forward.

Running the Agent Demos

We can now run demos of both agents we’ve built. We execute the agents in a manner that illustrates each step and contextualizes how the processes interact:

python
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

A similar approach is taken for the Problem Solver, structured to efficiently run through its nodes and gather results based on the problem presented.

Conclusion

Using a graph-driven architecture, we successfully developed intelligent agents capable of breaking down and tackling tasks step-by-step. By visualizing each agent’s workflow, we gain insights into how context-dependent prompts can be processed, enabling the agents to leverage the Gemini API’s content generation capabilities effectively. This setup not only enhances flexibility but also provides a clear visual representation of the logic flow, empowering users to understand complex reasoning workflows in an accessible manner.