In diesem Tutorial untersuchen wir, wie wir mithilfe von Gemini eine vollständige agentische KI-Orchestrierungspipeline entwerfen und ausführen, die auf semantischem Routing, symbolischen Leitplanken und Selbstkorrekturschleifen basiert. Wir zeigen Ihnen, wie wir mithilfe einer sauberen, modularen Architektur Agenten strukturieren, Aufgaben verteilen, Einschränkungen durchsetzen und Ausgaben verfeinern. Während wir jedes Snippet durchgehen, sehen wir, wie das System clever den richtigen Agenten auswählt, seine Ausgabe validiert und sich durch iterative Reflexion selbst verbessert. Schauen Sie sich das an Vollständige Codes hier.
import os
import json
import time
import typing
from dataclasses import dataclass, asdict
from google import genai
from google.genai import varieties
API_KEY = os.environ.get("GEMINI_API_KEY", "API Key")
consumer = genai.Consumer(api_key=API_KEY)
@dataclass
class AgentMessage:
supply: str
goal: str
content material: str
metadata: dict
timestamp: float = time.time()Wir richten unsere Kernumgebung ein, indem wir wichtige Bibliotheken importieren, den API-Schlüssel definieren und den Gemini-Consumer initialisieren. Wir erstellen außerdem die AgentMessage-Struktur, die als gemeinsames Kommunikationsformat zwischen Agenten fungiert. Schauen Sie sich das an Vollständige Codes hier.
class CognitiveEngine:
@staticmethod
def generate(immediate: str, system_instruction: str, json_mode: bool = False) -> str:
config = varieties.GenerateContentConfig(
temperature=0.1,
response_mime_type="software/json" if json_mode else "textual content/plain"
)
attempt:
response = consumer.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 greatest agent.
AVAILABLE AGENTS:
{json.dumps(self.registry, indent=2)}
USER REQUEST: "{user_query}"
Return ONLY a JSON object: {{"selected_agent": "agent_name", "reasoning": "transient motive"}}
"""
response_text = CognitiveEngine.generate(immediate, "You're a routing system.", json_mode=True)
attempt:
determination = json.hundreds(response_text)
print(f" (Router) Chosen: {determination('selected_agent')} (Motive: {determination('reasoning')})")
return determination('selected_agent')
besides:
return "general_agent"Wir bauen die kognitive Ebene mit Gemini auf, sodass wir je nach Anweisung sowohl Textual content- als auch JSON-Ausgaben generieren können. Außerdem implementieren wir den semantischen Router, der Anfragen analysiert und den am besten geeigneten Agenten auswählt. Schauen Sie sich das an Vollständige Codes hier.
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 artistic 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)Wir konstruieren die Employee-Brokers und den zentralen Orchestrator. Jeder Agent erhält eine klare Rolle: Analyst, Kreativer oder Programmierer, und wir konfigurieren den Orchestrator für die Verwaltung dieser Agenten. Während wir diesen Abschnitt durchgehen, sehen wir, wie wir das Agenten-Ökosystem definieren und es für die intelligente Aufgabendelegierung vorbereiten. Schauen Sie sich das an Vollständige Codes hier.
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 incorporates 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 Job: {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="Consumer", goal=target_name, content material=current_input, metadata={})
print(f" (Exec) {employee.identify} working... (Try {try+1})")
end result = employee.execute(msg)
if constraint:
is_valid, error_msg = self.validate_constraint(end result, constraint)
if not is_valid:
print(f" (Guardrail) VIOLATION: {error_msg}")
current_input = f"Your earlier reply failed a examine.nOriginal Request: {user_input}nYour Reply: {end result}nError: {error_msg}nFIX IT instantly."
proceed
print(f" (Success) Remaining Output:n{end result(:100)}...")
return end result
besides Exception as e:
print(f" (System Error) {e}")
time.sleep(1)
print(" (Failed) Max retries reached or self-correction failed.")
return NoneWir implementieren symbolische Leitplanken und eine Selbstkorrekturschleife, um Einschränkungen wie striktes JSON oder kein Markdown durchzusetzen. Wir führen eine iterative Verfeinerung durch, wenn Ausgaben gegen Anforderungen verstoßen, sodass unsere Agenten ihre eigenen Fehler beheben können. Schauen Sie sich das an Vollständige Codes hier.
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"
)Wir führen zwei vollständige Szenarien durch und zeigen Routing, Agentenausführung und Einschränkungsvalidierung in Aktion. Wir führen eine JSON-erzwungene Analyseaufgabe und eine Codierungsaufgabe mit Markdown-Einschränkungen aus, um das reflexive Verhalten zu beobachten.
Zusammenfassend sehen wir nun, wie mehrere Komponenten, Routing, Employee-Brokers, Leitplanken und Selbstkorrektur, zusammenkommen, um ein zuverlässiges und intelligentes Agentensystem zu schaffen. Wir erleben, wie jeder Teil zu einer robusten Aufgabenausführung beiträgt und sicherstellt, dass die Ergebnisse präzise, ausgerichtet und einschränkungsbewusst bleiben. Wenn wir über die Architektur nachdenken, erkennen wir, wie einfach wir sie mit neuen Agenten, umfassenderen Einschränkungen oder fortschrittlicheren Argumentationsstrategien erweitern können.
Schauen Sie sich das an Vollständige Codes hier. Schauen Sie sich gerne bei uns um GitHub-Seite für Tutorials, Codes und Notebooks. Sie können uns auch gerne weiter folgen Twitter und vergessen Sie nicht, bei uns mitzumachen 100.000+ ML SubReddit und Abonnieren Unser E-newsletter. Warten! Bist du im Telegram? Jetzt können Sie uns auch per Telegram kontaktieren.
Asif Razzaq ist CEO von Marktechpost Media Inc.. Als visionärer Unternehmer und Ingenieur setzt sich Asif dafür ein, das Potenzial der künstlichen Intelligenz für das soziale Wohl zu nutzen. Sein jüngstes Unterfangen ist die Einführung einer Medienplattform für künstliche Intelligenz, Marktechpost, die sich durch eine ausführliche Berichterstattung über maschinelles Lernen und Deep-Studying-Nachrichten auszeichnet, die sowohl technisch fundiert als auch für ein breites Publikum leicht verständlich ist. Die Plattform verfügt über mehr als 2 Millionen monatliche Aufrufe, was ihre Beliebtheit beim Publikum verdeutlicht.
