This tutorial will guide you in the creation of a Graph Agent Framework powered by Google Gemini API. The goal of this tutorial is to develop intelligent multi-step agents which execute tasks by using an interconnected graph structure. Each node is responsible for a particular function. These functions can include taking input, processing logically, deciding, or producing outputs. Python is the language of choice, NetworkX provides graph modelling, and matplotlib allows for visualization. We implement and run, at the end of the project, two examples: a Research assistant and a problem solver, in order to show how efficiently the framework handles complex reasoning workflows.
Google-generativeai Matplotlib
import google.generativeai as genai
Import networkx as Nx
Matplotlib.pyplot can be imported as a plt
Import Dict List Any Callable
Import json
import asyncio
Import dataclasses from dataclasses
Import Enum
API_KEY = "use your API key here"
genai.configure(api_key=API_KEY)
Installing the libraries google-generativeai (also known as Google-GenerativeAI), networkx and matplotlib is our first step to create a graph-based framework. We import the necessary modules and then configure Gemini API with our API key. This will enable content generation within our agent framework.
Take a look at the Codes.
Class NodeType (Enum)
Input = "input"
PROCESS = "process"
DECISION "decision"
Output = "output"
@dataclass
Class AgentNode
Str
NodeType
prompt: str
function: Callable There are no other options.
dependencies: List[str] = NoneNodeType is an enumeration that classifies different types of nodes, such as input, output, decision and process. We then structure nodes using the dataclass AgentNode. Each one has an ID, type prompt, optional functions, and dependencies.
def create_research_agent():
Agent = GraphAgent()
# Node input
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 AgentThe graph is created by adding special nodes sequentially. After defining a flow, which includes literature reviews, planning and analysis, we start with the topic. After analyzing the data, the agent makes a decision on quality and generates an exhaustive research report. This captures the entire lifecycle of structured research workflow.
Take a look at the 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 AgentThe problem solving agent is built by creating a logic sequence starting at the receiving of the statement of problem. The agent analyses the problem and generates several solution approaches. It then evaluates these on the basis of feasibility and effectiveness. Finally, it produces a structured plan for implementation, which allows automated step-by-step problem resolution.
Take a look at the 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:
if node_id >= "topic_input":
You can continue reading
context = {}
Node = agent_research.nodes[node_id]
if node.dependencies:
For dep.dependencies in node:
The following is a list of other words and phrases that you can use.[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}nnTask: {prompt}"
try:
response = research_agent.model.generate_content(prompt)
Result = 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:
If node_id is equal to "problem_input":
You can continue reading
context = {}
Node = Problem_Solver.Nodes[node_id]
if node.dependencies:
For dep.dependencies in node:
The following is a list of other words and phrases that you can use.[dep] = problem_solver.results.get(dep, "")
prompt = node.prompt
If you are not sure, please check the context.
context_str = "n".join([f"{k}: {v}" for k, v in context.items()])
prompt = f"Context:n{context_str}nnTask: {prompt}"
try:
response = problem_solver.model.generate_content(prompt)
Result = text.strip()
problem_solver.results[node_id] =====>
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!")The tutorial concludes with two powerful agents: one agent for solving problems and the other for doing research. We visualize the graph, initialize input and then execute each agent using topological ordering. Gemini provides contextual responses to each of the steps. This allows us to observe how autonomously agents progress through planning, analyses, decisions, and output creation.
We have successfully created and implemented intelligent agents which break tasks down into smaller steps and then solve them using a graph driven architecture. Each node is able to process context-dependent requests, utilize Gemini’s content creation capabilities, and pass results on to the next node. Modular design increases flexibility while allowing us to see the logic flow.
Take a look at the Codes. This research is the work of researchers on this project. SUBSCRIBE NOW Our AI Newsletter
Asif Razzaq serves as the CEO at Marktechpost Media Inc. As an entrepreneur, Asif has a passion for harnessing Artificial Intelligence to benefit society. Marktechpost is his latest venture, a media platform that focuses on Artificial Intelligence. It is known for providing in-depth news coverage about machine learning, deep learning, and other topics. The content is technically accurate and easy to understand by an audience of all backgrounds. Over 2 million views per month are a testament to the platform’s popularity.
