Coding Guide for Building Agentic AI System uses a secure, modular, and scanable organizational workflow
In this tutorial, we build an advanced agentic ae that uses the flight control pattern, and walk through each component step by step as we implement it. We manage the control plane as a central orchestrator that coordinates tools, manages security rules, and structures loop detection. Also, set up a small retrieval system, modified defined tools, and combine the layer of the aventic lemming with the dynamic layer and extract the actions. Finally, we look at how the entire system behaves as a guided, instrumental AI that is able to retrieve information, assess understanding, update student profiles, and incorporate all interactions into a unified, focused design. Look Full codes here.
import subprocess
import sys
def install_deps():
deps = ['anthropic', 'numpy', 'scikit-learn']
for dep in deps:
subprocess.check_call([sys.executable, '-m', 'pip', 'install', '-q', dep])
try:
import anthropic
except ImportError:
install_deps()
import anthropic
import json
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from dataclasses import dataclass, asdict
from typing import List, Dict, Any, Optional
from datetime import datetime
@dataclass
class Document:
id: str
content: str
metadata: Dict[str, Any]
embedding: Optional[np.ndarray] = None
class SimpleRAGRetriever:
def __init__(self):
self.documents = self._init_knowledge_base()
def _init_knowledge_base(self) -> List[Document]:
docs = [
Document("cs101", "Python basics: Variables store data. Use x=5 for integers, name="Alice" for strings. Print with print().", {"topic": "python", "level": "beginner"}),
Document("cs102", "Functions encapsulate reusable code. Define with def func_name(params): and call with func_name(args).", {"topic": "python", "level": "intermediate"}),
Document("cs103", "Object-oriented programming uses classes. class MyClass: defines structure, __init__ initializes instances.", {"topic": "python", "level": "advanced"}),
Document("math101", "Linear algebra: Vectors are ordered lists of numbers. Matrix multiplication combines transformations.", {"topic": "math", "level": "intermediate"}),
Document("ml101", "Machine learning trains models on data to make predictions. Supervised learning uses labeled examples.", {"topic": "ml", "level": "beginner"}),
Document("ml102", "Neural networks are composed of layers. Each layer applies weights and activation functions to transform inputs.", {"topic": "ml", "level": "advanced"}),
]
for i, doc in enumerate(docs):
doc.embedding = np.random.rand(128)
doc.embedding[i*20:(i+1)*20] += 2
return docs
def retrieve(self, query: str, top_k: int = 2) -> List[Document]:
query_embedding = np.random.rand(128)
scores = [cosine_similarity([query_embedding], [doc.embedding])[0][0] for doc in self.documents]
top_indices = np.argsort(scores)[-top_k:][::-1]
return [self.documents[i] for i in top_indices]We set up all the dependencies, import the libraries we depend on, and initialize the data structures of our database. We define a simple retrieval and generate a mock embedding to simulate the same search in a lightweight way. As we work on this block, we are preparing everything necessary to restore the latest parts. Look Full codes here.
class ToolRegistry:
def __init__(self, retriever: SimpleRAGRetriever):
self.retriever = retriever
self.interaction_log = []
self.user_state = {"level": "beginner", "topics_covered": []}
def search_knowledge(self, query: str, filters: Optional[Dict] = None) -> Dict:
docs = self.retriever.retrieve(query, top_k=2)
if filters:
docs = [d for d in docs if all(d.metadata.get(k) == v for k, v in filters.items())]
return {
"tool": "search_knowledge",
"results": [{"content": d.content, "metadata": d.metadata} for d in docs],
"count": len(docs)
}
def assess_understanding(self, topic: str) -> Dict:
questions = {
"python": ["What keyword defines a function?", "How do you create a variable?"],
"ml": ["What is supervised learning?", "Name two types of ML algorithms."],
"math": ["What is a vector?", "Explain matrix multiplication."]
}
return {
"tool": "assess_understanding",
"topic": topic,
"questions": questions.get(topic, ["General comprehension check."])
}
def update_learner_profile(self, topic: str, level: str) -> Dict:
if topic not in self.user_state["topics_covered"]:
self.user_state["topics_covered"].append(topic)
self.user_state["level"] = level
return {
"tool": "update_learner_profile",
"status": "updated",
"profile": self.user_state.copy()
}
def log_interaction(self, event: str, details: Dict) -> Dict:
log_entry = {
"timestamp": datetime.now().isoformat(),
"event": event,
"details": details
}
self.interaction_log.append(log_entry)
return {"tool": "log_interaction", "status": "logged", "entry_id": len(self.interaction_log)}We create a registry of tools that our agent uses while interacting with the system. We describe tools such as information search, screening, profile updates, and logging, and maintain a persistent user dictionary. As we use this layer, we see how each tool becomes a control over how the control plane can move. Look Full codes here.
class ControlPlane:
def __init__(self, tool_registry: ToolRegistry):
self.tools = tool_registry
self.safety_rules = {
"max_tools_per_request": 4,
"allowed_tools": ["search_knowledge", "assess_understanding",
"update_learner_profile", "log_interaction"]
}
self.execution_log = []
def execute(self, plan: Dict[str, Any]) -> Dict[str, Any]:
if not self._validate_request(plan):
return {"error": "Safety validation failed", "plan": plan}
action = plan.get("action")
params = plan.get("parameters", {})
result = self._route_and_execute(action, params)
self.execution_log.append({
"timestamp": datetime.now().isoformat(),
"plan": plan,
"result": result
})
return {
"success": True,
"action": action,
"result": result,
"metadata": {
"execution_count": len(self.execution_log),
"safety_checks_passed": True
}
}
def _validate_request(self, plan: Dict) -> bool:
action = plan.get("action")
if action not in self.safety_rules["allowed_tools"]:
return False
if len(self.execution_log) >= 100:
return False
return True
def _route_and_execute(self, action: str, params: Dict) -> Any:
tool_map = {
"search_knowledge": self.tools.search_knowledge,
"assess_understanding": self.tools.assess_understanding,
"update_learner_profile": self.tools.update_learner_profile,
"log_interaction": self.tools.log_interaction
}
tool_func = tool_map.get(action)
if tool_func:
return tool_func(**params)
return {"error": f"Unknown action: {action}"}We use a control plane that executes tools to manage tools, check security rules, and manage permissions. We verify all requests, route actions to the appropriate device, and keep a record of face execution. As we run this snippet, we see how the control plane becomes a control system that ensures predictable and safe aventic performance. Look Full codes here.
class TutorAgent:
def __init__(self, control_plane: ControlPlane, api_key: str):
self.control_plane = control_plane
self.client = anthropic.Anthropic(api_key=api_key)
self.conversation_history = []
def teach(self, student_query: str) -> str:
plan = self._plan_actions(student_query)
results = []
for action_plan in plan:
result = self.control_plane.execute(action_plan)
results.append(result)
response = self._synthesize_response(student_query, results)
self.conversation_history.append({
"query": student_query,
"plan": plan,
"results": results,
"response": response
})
return response
def _plan_actions(self, query: str) -> List[Dict]:
plan = []
query_lower = query.lower()
if any(kw in query_lower for kw in ["what", "how", "explain", "teach"]):
plan.append({
"action": "search_knowledge",
"parameters": {"query": query},
"context": {"intent": "knowledge_retrieval"}
})
if any(kw in query_lower for kw in ["test", "quiz", "assess", "check"]):
topic = "python" if "python" in query_lower else "ml"
plan.append({
"action": "assess_understanding",
"parameters": {"topic": topic},
"context": {"intent": "assessment"}
})
plan.append({
"action": "log_interaction",
"parameters": {"event": "query_processed", "details": {"query": query}},
"context": {"intent": "logging"}
})
return plan
def _synthesize_response(self, query: str, results: List[Dict]) -> str:
response_parts = [f"Student Query: {query}n"]
for result in results:
if result.get("success") and "result" in result:
tool_result = result["result"]
if result["action"] == "search_knowledge":
response_parts.append("n📚 Retrieved Knowledge:")
for doc in tool_result.get("results", []):
response_parts.append(f" • {doc['content']}")
elif result["action"] == "assess_understanding":
response_parts.append("n✅ Assessment Questions:")
for q in tool_result.get("questions", []):
response_parts.append(f" • {q}")
return "n".join(response_parts)We use Tutorutorutent, which organizes the actions, communicates with the control plane, and agrees on the final responses. We analyze the questions, generate multi-step plans, and synthesize the tools that show meaningful student responses. As we release this snippet, we see the agent behave intelligently by coordinating retrieval, testing, and logging. Look Full codes here.
def run_demo():
print("=" * 70)
print("Control Plane as a Tool: RAG AI Tutor Demo")
print("=" * 70)
API_KEY = "your-api-key-here"
retriever = SimpleRAGRetriever()
tool_registry = ToolRegistry(retriever)
control_plane = ControlPlane(tool_registry)
print("System initialized")
print(f"Tools: {len(control_plane.safety_rules['allowed_tools'])}")
print(f"Knowledge base: {len(retriever.documents)} documents")
try:
tutor = TutorAgent(control_plane, API_KEY)
except:
print("Mock mode enabled")
tutor = None
demo_queries = [
"Explain Python functions to me",
"I want to learn about machine learning",
"Test my understanding of Python basics"
]
for query in demo_queries:
print("n--- Query ---")
if tutor:
print(tutor.teach(query))
else:
plan = [
{"action": "search_knowledge", "parameters": {"query": query}},
{"action": "log_interaction", "parameters": {"event": "query", "details": {}}}
]
print(query)
for action in plan:
result = control_plane.execute(action)
print(f"{action['action']}: {result.get('success', False)}")
print("Summary")
print(f"Executions: {len(control_plane.execution_log)}")
print(f"Logs: {len(tool_registry.interaction_log)}")
print(f"Profile: {tool_registry.user_state}")
if __name__ == "__main__":
run_demo()We use a complete demo that starts everything, sample procedures for student questions, and Print System State Sumfaries. We watch agent by agent retrieving and cutting logs while the control plane enforces the rules and follows the history of the execution of the tracks. As we complete this block, we get a clear picture of how all the constructs work together in the loop you are teaching.
In conclusion, we gain a clear understanding of how the flight control pattern enables safety, reinforces safety, and creates a clean separation between consultation and instrument execution. Now we see how the retrieval system, the tool register, and the agentic programming layer come together to create a coherent AI tutor that responds intelligently to student tests. As we tried out the demo, we looked at how the system runs tasks, implements rules, and incorporates useful insights from tooltips, all while remaining a mod and being used for development.
Look Full codes here. Feel free to take a look at ours GitHub page for tutorials, code and notebooks. Also, feel free to follow us Kind of stubborn and don't forget to join ours 100K + ML Subreddit and sign up Our newsletter. Wait! Do you telegraph? Now you can join us by telegraph.
AsifAzzaq is the CEO of MarktechPost Media Inc.. as a visionary entrepreneur and developer, Asifi is committed to harnessing the power of social intelligence for good. His latest effort is the launch of a media intelligence platform, MarktechPpost, which stands out for its deep understanding of machine learning and deep learning stories that are technically sound and easily understood by a wide audience. The platform sticks to more than two million monthly views, which shows its popularity among the audience.
Follow Marktechpost: Add us as a favorite source on Google.
