A Full Code Implementation to Design a Graph-Structured AI Agent with Gemini for Task Planning, Retrieval, Computation, and Self-Critique

import os, json, time, ast, math, getpass
from dataclasses import dataclass, area
from typing import Dict, Listing, Callable, Any
import google.generativeai as genai


attempt:
   import networkx as nx
besides ImportError:
   nx = None

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're GraphAgent, a principled planner-executor. "
       "Desire 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()

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 test(n):
       if not isinstance(n, allowed): elevate ValueError("Unsafe expression")
       for c in ast.iter_child_nodes(n): test(c)
   test(node)
   return str(eval(compile(node, "<math>", "eval"), {"__builtins__": {}}, {}))


DOCS = [
   "Solar panels convert sunlight to electricity; capacity factor ~20%.",
   "Wind turbines harvest kinetic energy; onshore capacity factor ~35%.",
   "RAG = retrieval-augmented generation joins search with prompting.",
   "LangGraph enables cyclic graphs of agents; good for tool orchestration.",
]
def search_docs(q: str, ok: int = 3) -> Listing[str]:
   ql = q.decrease()
   scored = sorted(DOCS, key=lambda d: -sum(w in d.decrease() for w in ql.break up()))
   return scored[:k]

@dataclass
class State:
   activity: str
   plan: str = ""
   scratch: Listing[str] = area(default_factory=record)
   proof: Listing[str] = area(default_factory=record)
   outcome: str = ""
   step: int = 0
   executed: bool = False


def node_plan(state: State, mannequin) -> str:
   immediate = f"""Plan step-by-step to unravel the consumer activity.
Activity: {state.activity}
Return JSON: {{"subtasks": ["..."], "instruments": {{"search": true/false, "math": true/false}}, "success_criteria": ["..."]}}"""
   js = call_llm(mannequin, immediate)
   attempt:
       plan = json.hundreds(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.activity}"""
   alternative = call_llm(mannequin, immediate).decrease()
   if "math" in alternative and any(ch.isdigit() for ch in state.activity):
       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 centered search queries for:
Activity: {state.activity}
Return as JSON record of strings."""
   qjson = call_llm(mannequin, immediate)
   attempt:
       queries = json.hundreds(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.prolong(search_docs(q, ok=2))
   state.proof.prolong(record(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 activity:n"+state.activity
   expr = call_llm(mannequin, immediate)
   expr = "".be a part of(ch for ch in expr if ch in "0123456789+-*/().%^ ")
   attempt:
       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.activity}
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.executed = 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(activity: str, api_key: str) -> State:
   mannequin = make_model(api_key)
   state = State(activity=activity)
   cur = "plan"
   max_steps = 12
   whereas not state.executed and state.step < max_steps:
       state.step += 1
       nxt = NODES[cur](state, mannequin)
       if nxt == "finish": break
       cur = nxt
   return state


def ascii_graph():
   return """
START -> plan -> route -> (analysis <-> route) & (math <-> route) -> write -> critic -> END
"""

if __name__ == "__main__":
   key = os.getenv("GEMINI_API_KEY") or getpass.getpass(" Enter GEMINI_API_KEY: ")
   activity = enter(" Enter your activity: ").strip() or "Evaluate photo voltaic vs wind for reliability; compute 5*7."
   t0 = time.time()
   state = run_graph(activity, key)
   dt = time.time() - t0
   print("n=== GRAPH ===", ascii_graph())
   print(f"n End in {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:]))