In this tutorial, we explore how we design and run a complete agentic AI orchestration pipeline using Gemini, powered by semantic routing, symbolic guardrails, and self-correcting loops. We walk through how we structure agents, dispatch tasks, enforce constraints, and refine the output using a clean, modular architecture. As we move through each snippet, we see how the system intelligently chooses the right agent, validates its outputs, and improves itself through iterative reflection. check it out full code 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 have set up our core environment by importing the required libraries, defining the API key, and initializing the Gemini client. We also establish the AgentMessage structure, which serves as a shared communication format between agents. check it out full code 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, which allows us to generate both text and JSON output based on instructions. We also implement a semantic router, which analyzes queries and selects the most appropriate agent. check it out full code 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 build worker agents and central orchestrators. Each agent gets a clear role, analyst, creative, or coder, and we configure Orchestrator to manage them. As we review this section, we look at how we define the agent ecosystem and prepare it for intelligent task delegation. check it out full code 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-correcting loop to enforce constraints like strict JSON or no markdown. We run iterative refinements whenever the output violates requirements, allowing our agents to fix their mistakes. check it out full code 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 full scenarios, demonstrating routing, agent execution, and constraint verification in action. We run a coding task with a JSON-enforced analytical task and Markdown constraints to observe 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 agent system. We see how each part contributes to robust performance, ensuring outputs remain accurate, aligned, and constraint-aware. As we consider the architecture, we recognize how easily we can extend it with new agents, richer constraints, or more advanced reasoning strategies.
check it out full code hereFeel free to check us out GitHub page for tutorials, code, and notebooksAlso, feel free to follow us Twitter And don’t forget to join us 100k+ ml subreddit and subscribe our newsletterwait! Are you on Telegram? Now you can also connect with us on Telegram.
Asif Razzaq Marktechpost Media Inc. Is the CEO of. As a visionary entrepreneur and engineer, Asif is committed to harnessing the potential of Artificial Intelligence for social good. Their most recent endeavor is the launch of MarketTechPost, an Artificial Intelligence media platform, known for its in-depth coverage of Machine Learning and Deep Learning news that is technically robust and easily understood by a wide audience. The platform boasts of over 2 million monthly views, which shows its popularity among the audience.
