On this tutorial, we implement a sophisticated graph-based AI agent utilizing the GraphAgent framework and the Gemini 1.5 Flash mannequin. We outline a directed graph of nodes, every answerable for a particular perform: a planner to interrupt down the duty, a router to manage circulate, analysis and math nodes to offer exterior proof and computation, a author to synthesize the reply, and a critic to validate and refine the output. We combine Gemini by means of a wrapper that handles structured JSON prompts, whereas native Python features act as instruments for secure math analysis and doc search. By executing this pipeline end-to-end, we exhibit how reasoning, retrieval, and validation are modularized inside a single cohesive system. Take a look at the FULL CODES right here.
import os, json, time, ast, math, getpass
from dataclasses import dataclass, discipline
from typing import Dict, Listing, Callable, Any
import google.generativeai as genai
strive:
import networkx as nx
besides ImportError:
nx = NoneWe start by importing core Python libraries for information dealing with, timing, and secure analysis, together with dataclasses and typing helpers to construction our state. We additionally load the google.generativeai consumer to entry Gemini and, optionally, NetworkX for graph visualization. Take a look at the FULL CODES right here.
def make_model(api_key: str, model_name: str = "gemini-1.5-flash"):
genai.configure(api_key=api_key)
return genai.GenerativeModel(model_name, system_instruction=(
"You might be GraphAgent, a principled planner-executor. "
"Choose structured, concise outputs; use offered instruments when requested."
))
def call_llm(mannequin, immediate: str, temperature=0.2) -> str:
r = mannequin.generate_content(immediate, generation_config={"temperature": temperature})
return (r.textual content or "").strip()We outline a helper to configure and return a Gemini mannequin with a customized system instruction, and one other perform that calls the LLM with a immediate whereas controlling temperature. We use this setup to make sure our agent receives structured, concise outputs constantly. Take a look at the FULL CODES right here.
def safe_eval_math(expr: str) -> str:
node = ast.parse(expr, mode="eval")
allowed = (ast.Expression, ast.BinOp, ast.UnaryOp, ast.Num, ast.Fixed,
ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow, ast.Mod,
ast.USub, ast.UAdd, ast.FloorDiv, ast.AST)
def examine(n):
if not isinstance(n, allowed): elevate ValueError("Unsafe expression")
for c in ast.iter_child_nodes(n): examine(c)
examine(node)
return str(eval(compile(node, "We implement two key instruments for the agent: a secure math evaluator that parses and checks arithmetic expressions with ast earlier than execution, and a easy doc search that retrieves probably the most related snippets from a small in-memory corpus. We use these to offer the agent dependable computation and retrieval capabilities with out exterior dependencies. Take a look at the FULL CODES right here.
@dataclass
class State:
job: str
plan: str = ""
scratch: Listing[str] = discipline(default_factory=checklist)
proof: Listing[str] = discipline(default_factory=checklist)
outcome: str = ""
step: int = 0
finished: bool = False
def node_plan(state: State, mannequin) -> str:
immediate = f"""Plan step-by-step to resolve the consumer job.
Activity: {state.job}
Return JSON: {{"subtasks": ["..."], "instruments": {{"search": true/false, "math": true/false}}, "success_criteria": ["..."]}}"""
js = call_llm(mannequin, immediate)
strive:
plan = json.masses(js[js.find("{"): js.rfind("}")+1])
besides Exception:
plan = {"subtasks": ["Research", "Synthesize"], "instruments": {"search": True, "math": False}, "success_criteria": ["clear answer"]}
state.plan = json.dumps(plan, indent=2)
state.scratch.append("PLAN:n"+state.plan)
return "route"
def node_route(state: State, mannequin) -> str:
immediate = f"""You're a router. Resolve subsequent node.
Context scratch:n{chr(10).be a part of(state.scratch[-5:])}
If math wanted -> 'math', if analysis wanted -> 'analysis', if prepared -> 'write'.
Return one token from [research, math, write]. Activity: {state.job}"""
alternative = call_llm(mannequin, immediate).decrease()
if "math" in alternative and any(ch.isdigit() for ch in state.job):
return "math"
if "analysis" in alternative or not state.proof:
return "analysis"
return "write"
def node_research(state: State, mannequin) -> str:
immediate = f"""Generate 3 targeted search queries for:
Activity: {state.job}
Return as JSON checklist of strings."""
qjson = call_llm(mannequin, immediate)
strive:
queries = json.masses(qjson[qjson.find("["): qjson.rfind("]")+1])[:3]
besides Exception:
queries = [state.task, "background "+state.task, "pros cons "+state.task]
hits = []
for q in queries:
hits.lengthen(search_docs(q, okay=2))
state.proof.lengthen(checklist(dict.fromkeys(hits)))
state.scratch.append("EVIDENCE:n- " + "n- ".be a part of(hits))
return "route"
def node_math(state: State, mannequin) -> str:
immediate = "Extract a single arithmetic expression from this job:n"+state.job
expr = call_llm(mannequin, immediate)
expr = "".be a part of(ch for ch in expr if ch in "0123456789+-*/().%^ ")
strive:
val = safe_eval_math(expr)
state.scratch.append(f"MATH: {expr} = {val}")
besides Exception as e:
state.scratch.append(f"MATH-ERROR: {expr} ({e})")
return "route"
def node_write(state: State, mannequin) -> str:
immediate = f"""Write the ultimate reply.
Activity: {state.job}
Use the proof and any math outcomes under, cite inline like [1],[2].
Proof:n{chr(10).be a part of(f'[{i+1}] '+e for i,e in enumerate(state.proof))}
Notes:n{chr(10).be a part of(state.scratch[-5:])}
Return a concise, structured reply."""
draft = call_llm(mannequin, immediate, temperature=0.3)
state.outcome = draft
state.scratch.append("DRAFT:n"+draft)
return "critic"
def node_critic(state: State, mannequin) -> str:
immediate = f"""Critique and enhance the reply for factuality, lacking steps, and readability.
If repair wanted, return improved reply. Else return 'OK'.
Reply:n{state.outcome}nCriteria:n{state.plan}"""
crit = call_llm(mannequin, immediate)
if crit.strip().higher() != "OK" and len(crit) > 30:
state.outcome = crit.strip()
state.scratch.append("REVISED")
state.finished = True
return "finish"
NODES: Dict[str, Callable[[State, Any], str]] = {
"plan": node_plan, "route": node_route, "analysis": node_research,
"math": node_math, "write": node_write, "critic": node_critic
}
def run_graph(job: str, api_key: str) -> State:
mannequin = make_model(api_key)
state = State(job=job)
cur = "plan"
max_steps = 12
whereas not state.finished and state.step plan -> route -> (analysis route) & (math route) -> write -> critic -> END
"""
We outline a typed State dataclass to persist the duty, plan, proof, scratch notes, and management flags because the graph executes. We implement node features, a planner, a router, analysis, math, a author, and a critic. These features mutate the state and return the following node label. We then register them in NODES and iterate in run_graph till finished. We additionally expose ascii_graph() to visualise the management circulate we comply with as we route between analysis/math and finalize with a critique. Take a look at the FULL CODES right here.
if __name__ == "__main__":
key = os.getenv("GEMINI_API_KEY") or getpass.getpass("🔐 Enter GEMINI_API_KEY: ")
job = enter("📝 Enter your job: ").strip() or "Evaluate photo voltaic vs wind for reliability; compute 5*7."
t0 = time.time()
state = run_graph(job, key)
dt = time.time() - t0
print("n=== GRAPH ===", ascii_graph())
print(f"n✅ Lead to {dt:.2f}s:n{state.outcome}n")
print("---- Proof ----")
print("n".be a part of(state.proof))
print("n---- Scratch (final 5) ----")
print("n".be a part of(state.scratch[-5:]))We outline this system’s entry level: we securely learn the Gemini API key, take a job as enter, after which run the graph by means of run_graph. We measure execution time, print the ASCII graph of the workflow, show the ultimate outcome, and likewise output supporting proof and the previous few scratch notes for transparency. Take a look at the FULL CODES right here.
In conclusion, we exhibit how a graph-structured agent allows the design of deterministic workflows round a probabilistic LLM. We observe how the planner node enforces job decomposition, the router dynamically selects between analysis and math, and the critic gives iterative enchancment for factuality and readability. Gemini acts because the central reasoning engine, whereas the graph nodes provide construction, security checks, and clear state administration. We conclude with a totally useful agent that showcases the advantages of mixing graph orchestration with a contemporary LLM, enabling extensions reminiscent of customized toolchains, multi-turn reminiscence, or parallel node execution in additional advanced deployments.
Take a look at the FULL CODES right here. Be at liberty to take a look at our GitHub Web page for Tutorials, Codes and Notebooks. Additionally, be at liberty to comply with us on Twitter and don’t overlook to affix our 100k+ ML SubReddit and Subscribe to our Publication.
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.
