Hands on: the mini-AGI engine

pip install google-genai
export GEMINI_API_KEY="your-key-here"

import os
import subprocess
import sys
from google import genai
from google.genai import types

client = genai.Client()   # reads GEMINI_API_KEY from the environment


def run_python_code(code_string):
    """Execute python code in a child process and return output or error."""
    filename = "temp_agent_sandbox.py"
    with open(filename, "w") as f:
        f.write(code_string)

    try:
        result = subprocess.run(
            [sys.executable, filename],
            capture_output=True,
            text=True,
            timeout=10,           # a runaway loop must not hang the agent
        )
        if result.returncode != 0:
            return f"EXECUTION ERROR:\n{result.stderr}"
        return f"SUCCESS OUTPUT:\n{result.stdout}"
    except Exception as e:
        return f"SYSTEM EXCEPTION: {str(e)}"
    finally:
        if os.path.exists(filename):
            os.remove(filename)

def agi_cognitive_loop(goal: str, max_iterations: int = 3):
    """Plan -> execute -> evaluate -> self-correct -> achieve goal."""
    print(f"\n[AGI Goal Initialized]: {goal}\n" + "=" * 40)

    system_prompt = (
        "You are an autonomous AGI reasoning agent. Your job is to achieve the "
        "user's goal by writing Python code blocks. Respond in two parts:\n"
        "1. THOUGHT: explain your step-by-step logic.\n"
        "2. CODE: enclosed in a python fenced block, containing executable code "
        "that solves or tests the step.\n"
        "If a previous step failed, analyse the error output and write "
        "corrected code."
    )

    # Working memory: the growing record of what has been tried.
    conversation_history = f"Goal: {goal}"

    for iteration in range(1, max_iterations + 1):
        print(f"\n--- [Iteration {iteration} of {max_iterations}] ---")

        # DECIDE
        response = client.models.generate_content(
            model="gemini-2.5-flash",
            contents=conversation_history,
            config=types.GenerateContentConfig(
                system_instruction=system_prompt,
                temperature=0.2,   # low temperature: strict logical adherence
            ),
        )
        model_output = response.text
        print(f"Agent thought and plan:\n{model_output}\n")

        fence = chr(96) * 3 + "python"      # the python code fence marker
        if fence not in model_output:
            print("No executable code block found this turn.")
            conversation_history += (
                f"\n\nIteration {iteration}: no code provided. "
                "Write code to test your logic."
            )
            continue

        # ACT
        code_block = model_output.split(fence)[1].split(chr(96) * 3)[0].strip()
        print("Executing agent-generated code...")
        execution_result = run_python_code(code_block)
        print(execution_result)

        # OBSERVE — the result becomes context for the next decision
        conversation_history += (
            f"\n\nIteration {iteration} code output:\n{execution_result}"
        )

        if "SUCCESS OUTPUT" in execution_result:
            print("\n[AGI Status]: goal reached and verified via execution.")
            return

    print("\n[AGI Status]: max iterations reached without closure.")


if __name__ == "__main__":
    goal = (
        "Write a python script that calculates the first 10 numbers of the "
        "Fibonacci sequence, verifies mathematically that each number is the "
        "sum of the two preceding ones, and prints 'VERIFICATION PASSED'."
    )
    agi_cognitive_loop(goal)

[AGI Goal Initialized]: Write a python script that calculates...
========================================

--- [Iteration 1 of 3] ---
Agent thought and plan:
THOUGHT: I will generate the sequence iteratively, then assert the
recurrence relation for every element from index 2 onward...

Executing agent-generated code...
SUCCESS OUTPUT:
[0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
VERIFICATION PASSED

[AGI Status]: goal reached and verified via execution.