In this tutorial, we explore how we design and run a full agentic AI orchestration pipeline powered by semantic routing, symbolic guardrails, and self-correction loops using Gemini. We walk through how we structure agents, dispatch tasks, enforce constraints, and refine outputs using a clean, modular architecture. As we progress through each snippet, we see how the system intelligently chooses the right agent, validates its output, and improves itself through iterative reflection. Check out the Full Codes here.
import os
import json
import time
import typing
from dataclasses import dataclass, asdict
from google import genai
from google.genai import types
API_KEY = os.environ.get("GEMINI_API_KEY", "API Key")
client = genai.Client(api_key=API_KEY)
@dataclass
class AgentMessage:
source: str
target: str
content: str
metadata: dict
timestamp: float = time.time()We set up our core environment by importing essential libraries, defining the API key, and initializing the Gemini client. We also establish the AgentMessage structure, which acts as the shared communication format between agents. Check out the Full Codes here.
class CognitiveEngine:
@staticmethod
def generate(prompt: str, system_instruction: str, json_mode: bool = False) -> str:
config = types.GenerateContentConfig(
temperature=0.1,
response_mime_type="application/json" if json_mode else "text/plain"
)
try:
response = client.models.generate_content(
model="gemini-2.0-flash",
contents=prompt,
config=config
)
return response.text
except Exception as e:
raise ConnectionError(f"Gemini API Error: {e}")
class SemanticRouter:
def __init__(self, agents_registry: dict):
self.registry = agents_registry
def route(self, user_query: str) -> str:
prompt = f"""
You are a Master Dispatcher. Analyze the user request and map it to the ONE best agent.
AVAILABLE AGENTS:
{json.dumps(self.registry, indent=2)}
USER REQUEST: "{user_query}"
Return ONLY a JSON object: {{"selected_agent": "agent_name", "reasoning": "brief reason"}}
"""
response_text = CognitiveEngine.generate(prompt, "You are a routing system.", json_mode=True)
try:
decision = json.loads(response_text)
print(f" [Router] Selected: {decision['selected_agent']} (Reason: {decision['reasoning']})")
return decision['selected_agent']
except:
return "general_agent"We build the cognitive layer using Gemini, allowing us to generate both text and JSON outputs depending on the instruction. We also implement the semantic router, which analyzes queries and selects the most suitable agent. Check out the Full Codes here.
class Agent:
def __init__(self, name: str, instruction: str):
self.name = name
self.instruction = instruction
def execute(self, message: AgentMessage) -> str:
return CognitiveEngine.generate(
prompt=f"Input: {message.content}",
system_instruction=self.instruction
)
class Orchestrator:
def __init__(self):
self.agents_info = {
"analyst_bot": "Analyzes data, logic, and math. Returns structured JSON summaries.",
"creative_bot": "Writes poems, stories, and creative text. Returns plain text.",
"coder_bot": "Writes Python code snippets."
}
self.workers = {
"analyst_bot": Agent("analyst_bot", "You are a Data Analyst. output strict JSON."),
"creative_bot": Agent("creative_bot", "You are a Creative Writer."),
"coder_bot": Agent("coder_bot", "You are a Python Expert. Return only code.")
}
self.router = SemanticRouter(self.agents_info)We construct the worker agents and the central orchestrator. Each agent receives a clear role, analyst, creative, or coder, and we configure the orchestrator to manage them. As we review this section, we see how we define the agent ecosystem and prepare it for intelligent task delegation. Check out the Full Codes here.
def validate_constraint(self, content: str, constraint_type: str) -> tuple[bool, str]:
if constraint_type == "json_only":
try:
json.loads(content)
return True, "Valid JSON"
except:
return False, "Output was not valid JSON."
if constraint_type == "no_markdown":
if "```" in content:
return False, "Output contains Markdown code blocks, which are forbidden."
return True, "Valid Text"
return True, "Pass"
def run_task(self, user_input: str, constraint: str = None, max_retries: int = 2):
print(f"\n--- New Task: {user_input} ---")
target_name = self.router.route(user_input)
worker = self.workers.get(target_name)
current_input = user_input
history = []
for attempt in range(max_retries + 1):
try:
msg = AgentMessage(source="User", target=target_name, content=current_input, metadata={})
print(f" [Exec] {worker.name} working... (Attempt {attempt+1})")
result = worker.execute(msg)
if constraint:
is_valid, error_msg = self.validate_constraint(result, constraint)
if not is_valid:
print(f" [Guardrail] VIOLATION: {error_msg}")
current_input = f"Your previous answer failed a check.\nOriginal Request: {user_input}\nYour Answer: {result}\nError: {error_msg}\nFIX IT immediately."
continue
print(f" [Success] Final Output:\n{result[:100]}...")
return result
except Exception as e:
print(f" [System Error] {e}")
time.sleep(1)
print(" [Failed] Max retries reached or self-correction failed.")
return NoneWe implement symbolic guardrails and a self-correction loop to enforce constraints like strict JSON or no Markdown. We run iterative refinement whenever outputs violate requirements, allowing our agents to fix their own mistakes. Check out the Full Codes here.
if __name__ == "__main__":
orchestrator = Orchestrator()
orchestrator.run_task(
"Compare the GDP of France and Germany in 2023.",
constraint="json_only"
)
orchestrator.run_task(
"Write a Python function for Fibonacci numbers.",
constraint="no_markdown"
)We execute two complete scenarios, showcasing routing, agent execution, and constraint validation in action. We run a JSON-enforced analytical task and a coding task with Markdown restrictions to observe the reflexive behavior.
In conclusion, we now see how multiple components, routing, worker agents, guardrails, and self-correction, come together to create a reliable and intelligent agentic system. We witness how each part contributes to robust task execution, ensuring that outputs remain accurate, aligned, and constraint-aware. As we reflect on the architecture, we recognize how easily we can expand it with new agents, richer constraints, or more advanced reasoning strategies.
Check out the Full Codes here. Feel free to check out our GitHub Page for Tutorials, Codes and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is committed to harnessing the potential of Artificial Intelligence for social good. His most recent endeavor is the launch of an Artificial Intelligence Media Platform, Marktechpost, which stands out for its in-depth coverage of machine learning and deep learning news that is both technically sound and easily understandable by a wide audience. The platform boasts of over 2 million monthly views, illustrating its popularity among audiences.
