On this tutorial, we discover how we design and run a full agentic AI orchestration pipeline powered by semantic routing, symbolic guardrails, and self-correction loops utilizing Gemini. We stroll via how we construction brokers, dispatch duties, implement constraints, and refine outputs utilizing a clear, modular structure. As we progress via every snippet, we see how the system intelligently chooses the fitting agent, validates its output, and improves itself via iterative reflection. Take a look at the Full Codes right here.
import os
import json
import time
import typing
from dataclasses import dataclass, asdict
from google import genai
from google.genai import sorts
API_KEY = os.environ.get("GEMINI_API_KEY", "API Key")
shopper = genai.Shopper(api_key=API_KEY)
@dataclass
class AgentMessage:
supply: str
goal: str
content material: str
metadata: dict
timestamp: float = time.time()We arrange our core setting by importing important libraries, defining the API key, and initializing the Gemini shopper. We additionally set up the AgentMessage construction, which acts because the shared communication format between brokers. Take a look at the Full Codes right here.
class CognitiveEngine:
@staticmethod
def generate(immediate: str, system_instruction: str, json_mode: bool = False) -> str:
config = sorts.GenerateContentConfig(
temperature=0.1,
response_mime_type="utility/json" if json_mode else "textual content/plain"
)
attempt:
response = shopper.fashions.generate_content(
mannequin="gemini-2.0-flash",
contents=immediate,
config=config
)
return response.textual content
besides Exception as e:
elevate 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:
immediate = f"""
You're a Grasp Dispatcher. Analyze the consumer request and map it to the ONE finest agent.
AVAILABLE AGENTS:
{json.dumps(self.registry, indent=2)}
USER REQUEST: "{user_query}"
Return ONLY a JSON object: {{"selected_agent": "agent_name", "reasoning": "transient purpose"}}
"""
response_text = CognitiveEngine.generate(immediate, "You're a routing system.", json_mode=True)
attempt:
resolution = json.hundreds(response_text)
print(f" [Router] Chosen: {resolution['selected_agent']} (Cause: {resolution['reasoning']})")
return resolution['selected_agent']
besides:
return "general_agent"We construct the cognitive layer utilizing Gemini, permitting us to generate each textual content and JSON outputs relying on the instruction. We additionally implement the semantic router, which analyzes queries and selects essentially the most appropriate agent. Take a look at the Full Codes right here.
class Agent:
def __init__(self, identify: str, instruction: str):
self.identify = identify
self.instruction = instruction
def execute(self, message: AgentMessage) -> str:
return CognitiveEngine.generate(
immediate=f"Enter: {message.content material}",
system_instruction=self.instruction
)
class Orchestrator:
def __init__(self):
self.agents_info = {
"analyst_bot": "Analyzes knowledge, logic, and math. Returns structured JSON summaries.",
"creative_bot": "Writes poems, tales, and inventive textual content. Returns plain textual content.",
"coder_bot": "Writes Python code snippets."
}
self.staff = {
"analyst_bot": Agent("analyst_bot", "You're a Information Analyst. output strict JSON."),
"creative_bot": Agent("creative_bot", "You're a Artistic Author."),
"coder_bot": Agent("coder_bot", "You're a Python Skilled. Return solely code.")
}
self.router = SemanticRouter(self.agents_info)We assemble the employee brokers and the central orchestrator. Every agent receives a transparent position, analyst, artistic, or coder, and we configure the orchestrator to handle them. As we evaluation this part, we see how we outline the agent ecosystem and put together it for clever process delegation. Take a look at the Full Codes right here.
def validate_constraint(self, content material: str, constraint_type: str) -> tuple[bool, str]:
if constraint_type == "json_only":
attempt:
json.hundreds(content material)
return True, "Legitimate JSON"
besides:
return False, "Output was not legitimate JSON."
if constraint_type == "no_markdown":
if "```" in content material:
return False, "Output comprises Markdown code blocks, that are forbidden."
return True, "Legitimate Textual content"
return True, "Go"
def run_task(self, user_input: str, constraint: str = None, max_retries: int = 2):
print(f"n--- New Activity: {user_input} ---")
target_name = self.router.route(user_input)
employee = self.staff.get(target_name)
current_input = user_input
historical past = []
for try in vary(max_retries + 1):
attempt:
msg = AgentMessage(supply="Person", goal=target_name, content material=current_input, metadata={})
print(f" [Exec] {employee.identify} working... (Try {try+1})")
consequence = employee.execute(msg)
if constraint:
is_valid, error_msg = self.validate_constraint(consequence, constraint)
if not is_valid:
print(f" [Guardrail] VIOLATION: {error_msg}")
current_input = f"Your earlier reply failed a test.nOriginal Request: {user_input}nYour Reply: {consequence}nError: {error_msg}nFIX IT instantly."
proceed
print(f" [Success] Last Output:n{consequence[:100]}...")
return consequence
besides 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 implement constraints like strict JSON or no Markdown. We run iterative refinement every time outputs violate necessities, permitting our brokers to repair their very own errors. Take a look at the Full Codes right here.
if __name__ == "__main__":
orchestrator = Orchestrator()
orchestrator.run_task(
"Examine the GDP of France and Germany in 2023.",
constraint="json_only"
)
orchestrator.run_task(
"Write a Python perform for Fibonacci numbers.",
constraint="no_markdown"
)We execute two full situations, showcasing routing, agent execution, and constraint validation in motion. We run a JSON-enforced analytical process and a coding process with Markdown restrictions to watch the reflexive conduct.Â
In conclusion, we now see how a number of elements, routing, employee brokers, guardrails, and self-correction, come collectively to create a dependable and clever agentic system. We witness how every half contributes to strong process execution, guaranteeing that outputs stay correct, aligned, and constraint-aware. As we mirror on the structure, we acknowledge how simply we will broaden it with new brokers, richer constraints, or extra superior reasoning methods.
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 happy to observe us on Twitter and don’t overlook to hitch our 100k+ ML SubReddit and Subscribe to our E-newsletter. Wait! are you on telegram? now you may be a part of us on telegram as properly.
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.
