Building a Sophisticated Agentic AI with a Control-Plane Architecture
This guide demonstrates how to develop a robust Agentic AI system by leveraging the control-plane architectural pattern. We break down each integral component, illustrating how the control plane acts as the central coordinator-managing tool interactions, enforcing safety protocols, and orchestrating the reasoning workflow. Alongside, we implement a compact retrieval mechanism, modular tool definitions, and an agentic reasoning layer that dynamically formulates and executes plans. Ultimately, the system functions as a disciplined, tool-aware AI capable of knowledge retrieval, comprehension evaluation, learner profile updates, and comprehensive interaction logging within a scalable framework.
Setting Up Dependencies and Initializing the Knowledge Base
First, we ensure all necessary libraries are installed and imported, including packages for AI interaction, numerical computation, and similarity measurement. We then create foundational data structures representing our knowledge base. To simulate semantic search, we generate mock embeddings for documents, enabling lightweight similarity comparisons. This setup lays the groundwork for retrieval-driven reasoning in subsequent modules.
import subprocess
import sys
def installdependencies():
requiredpackages = ['anthropic', 'numpy', 'scikit-learn']
for package in requiredpackages:
subprocess.checkcall([sys.executable, '-m', 'pip', 'install', '-q', package])
try:
import anthropic
except ImportError:
installdependencies()
import anthropic
import json
import numpy as np
from sklearn.metrics.pairwise import cosinesimilarity
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime
@dataclass
class Document:
id: str
content: str
metadata: Dict[str, any]
embedding: Optional[np.ndarray] = None
class SimpleRetriever:
def init(self):
self.documents = self.createknowledgebase()
def createknowledgebase(self) -> List[Document]:
docs = [
Document("py101", "Introduction to Python: Variables hold data. Assign with x=10 or name='Bob'. Use print() to display output.", {"topic": "python", "level": "beginner"}),
Document("py201", "Functions in Python encapsulate logic. Define with def functionname(params): and invoke with functionname(args).", {"topic": "python", "level": "intermediate"}),
Document("py301", "Advanced OOP concepts: Classes define blueprints. Use init for initialization.", {"topic": "python", "level": "advanced"}),
Document("math201", "Linear algebra essentials: Vectors represent ordered data. Matrix multiplication combines linear transformations.", {"topic": "math", "level": "intermediate"}),
Document("ml101", "Basics of machine learning: Training models on labeled data to predict outcomes.", {"topic": "ml", "level": "beginner"}),
Document("ml301", "Deep learning: Neural networks consist of layers applying weights and activation functions.", {"topic": "ml", "level": "advanced"}),
]
for idx, doc in enumerate(docs):
doc.embedding = np.random.rand(128)
doc.embedding[idx20:(idx+1)20] += 2 # Boost segment for uniqueness
return docs
def retrieve(self, query: str, topk: int = 2) -> List[Document]:
queryembedding = np.random.rand(128)
similarityscores = [cosinesimilarity([queryembedding], [doc.embedding])[0][0] for doc in self.documents]
topindices = np.argsort(similarityscores)[-topk:][::-1]
return [self.documents[i] for i in topindices]
Designing a Modular Tool Registry for Agent Interaction
Next, we develop a tool registry that encapsulates the AI’s capabilities. This registry includes functions for knowledge search, learner assessment, profile management, and interaction logging. It maintains a persistent user state to track progress and preferences. Each tool acts as an independent module, allowing the control plane to delegate tasks efficiently and maintain a clean separation of concerns.
class ToolRegistry:
def init(self, retriever: SimpleRetriever):
self.retriever = retriever
self.interactionlog = []
self.userprofile = {"level": "beginner", "topicscompleted": []}
def searchknowledge(self, query: str, filters: Optional[Dict] = None) -> Dict:
results = self.retriever.retrieve(query, topk=2)
if filters:
results = [doc for doc in results if all(doc.metadata.get(k) == v for k, v in filters.items())]
return {
"tool": "searchknowledge",
"results": [{"content": doc.content, "metadata": doc.metadata} for doc in results],
"count": len(results)
}
def evaluatecomprehension(self, topic: str) -> Dict:
questionbank = {
"python": ["Which keyword is used to define a function?", "How do you assign a value to a variable?"],
"ml": ["What defines supervised learning?", "List two common machine learning algorithms."],
"math": ["Define a vector.", "Describe matrix multiplication."]
}
return {
"tool": "evaluatecomprehension",
"topic": topic,
"questions": questionbank.get(topic, ["Please explain the topic in your own words."])
}
def updateprofile(self, topic: str, level: str) -> Dict:
if topic not in self.userprofile["topicscompleted"]:
self.userprofile["topicscompleted"].append(topic)
self.userprofile["level"] = level
return {
"tool": "updateprofile",
"status": "profileupdated",
"profile": self.userprofile.copy()
}
def recordinteraction(self, event: str, details: Dict) -> Dict:
entry = {
"timestamp": datetime.utcnow().isoformat(),
"event": event,
"details": details
}
self.interactionlog.append(entry)
return {"tool": "recordinteraction", "status": "logged", "entryid": len(self.interactionlog)}
Implementing the Control Plane for Safe and Coordinated Execution
The control plane serves as the command center, validating requests against safety policies, routing actions to the appropriate tools, and maintaining an execution history. It enforces constraints such as limiting the number of tools invoked per request and restricting permissible actions, ensuring the AI operates within defined boundaries. This layer guarantees predictable, secure, and auditable agent behavior.
class ControlPlane:
def init(self, toolregistry: ToolRegistry):
self.tools = toolregistry
self.safetypolicies = {
"maxtoolsperrequest": 4,
"permittedtools": ["searchknowledge", "evaluatecomprehension", "updateprofile", "recordinteraction"]
}
self.executionhistory = []
def execute(self, plan: Dict[str, any]) -> Dict[str, any]:
if not self.isrequestsafe(plan):
return {"error": "Request failed safety validation", "plan": plan}
action = plan.get("action")
params = plan.get("parameters", {})
outcome = self.dispatchaction(action, params)
self.executionhistory.append({
"timestamp": datetime.utcnow().isoformat(),
"plan": plan,
"outcome": outcome
})
return {
"success": True,
"action": action,
"result": outcome,
"metadata": {
"executionscount": len(self.executionhistory),
"safetypassed": True
}
}
def isrequestsafe(self, plan: Dict) -> bool:
action = plan.get("action")
if action not in self.safetypolicies["permittedtools"]:
return False
if len(self.executionhistory) >= 100:
return False
return True
def dispatchaction(self, action: str, params: Dict) -> any:
actionmap = {
"searchknowledge": self.tools.searchknowledge,
"evaluatecomprehension": self.tools.evaluatecomprehension,
"updateprofile": self.tools.updateprofile,
"recordinteraction": self.tools.recordinteraction
}
func = actionmap.get(action)
if func:
return func(*params)
return {"error": f"Action '{action}' is not recognized."}
Creating the Tutor Agent: Planning, Executing, and Responding
The TutorAgent acts as the intelligent interface, interpreting student queries, devising multi-step action plans, and synthesizing comprehensive responses. It interacts with the control plane to execute planned tasks such as knowledge retrieval, comprehension checks, and logging. This agent dynamically adapts its strategy based on query content, ensuring relevant and coherent feedback for learners.
class TutorAgent:
def init(self, controlplane: ControlPlane, apikey: str):
self.controlplane = controlplane
self.client = anthropic.Anthropic(apikey=apikey)
self.history = []
def teach(self, studentinput: str) -> str:
actionplan = self.generateplan(studentinput)
results = []
for step in actionplan:
result = self.controlplane.execute(step)
results.append(result)
response = self.composeresponse(studentinput, results)
self.history.append({
"input": studentinput,
"plan": actionplan,
"results": results,
"response": response
})
return response
def generateplan(self, query: str) -> List[Dict]:
plan = []
lowerquery = query.lower()
if any(keyword in lowerquery for keyword in ["explain", "how", "what", "teach"]):
plan.append({
"action": "searchknowledge",
"parameters": {"query": query},
"context": {"purpose": "knowledgeretrieval"}
})
if any(keyword in lowerquery for keyword in ["quiz", "test", "assess", "evaluate"]):
topic = "python" if "python" in lowerquery else "ml"
plan.append({
"action": "evaluatecomprehension",
"parameters": {"topic": topic},
"context": {"purpose": "assessment"}
})
plan.append({
"action": "recordinteraction",
"parameters": {"event": "queryhandled", "details": {"query": query}},
"context": {"purpose": "logging"}
})
return plan
def composeresponse(self, query: str, results: List[Dict]) -> str:
responselines = [f"Student Query: {query}n"]
for res in results:
if res.get("success") and "result" in res:
content = res["result"]
if res["action"] == "searchknowledge":
responselines.append("n📚 Retrieved Information:")
for doc in content.get("results", []):
responselines.append(f" • {doc['content']}")
elif res["action"] == "evaluatecomprehension":
responselines.append("n✅ Assessment Questions:")
for question in content.get("questions", []):
responselines.append(f" • {question}")
return "n".join(responselines)
Demonstration: Running the AI Tutor System
To illustrate the system in action, we initialize all components and process a series of example student queries. The demo showcases how the agent coordinates retrieval and logging, while the control plane enforces safety and tracks execution. This end-to-end example highlights the seamless integration of all modules within a realistic tutoring scenario.
def rundemo():
print("=" 70)
print("Agentic AI Tutor Demo Using Control-Plane Architecture")
print("=" * 70)
APIKEY = "your-api-key-here"
retriever = SimpleRetriever()
tools = ToolRegistry(retriever)
controlplane = ControlPlane(tools)
print("System initialized successfully")
print(f"Available tools: {len(controlplane.safetypolicies['permittedtools'])}")
print(f"Knowledge base size: {len(retriever.documents)} documents")
try:
tutoragent = TutorAgent(controlplane, APIKEY)
except Exception:
print("Running in mock mode without API access")
tutoragent = None
samplequeries = [
"Can you explain how Python functions work?",
"I want to understand the basics of machine learning.",
"Please test my knowledge on Python fundamentals."
]
for query in samplequeries:
print("n--- New Query ---")
if tutoragent:
print(tutoragent.teach(query))
else:
plan = [
{"action": "searchknowledge", "parameters": {"query": query}},
{"action": "recordinteraction", "parameters": {"event": "queryreceived", "details": {}}}
]
print(query)
for step in plan:
outcome = controlplane.execute(step)
print(f"{step['action']} executed: {outcome.get('success', False)}")
print("nSummary:")
print(f"Total executions: {len(controlplane.executionhistory)}")
print(f"Total logged interactions: {len(tools.interactionlog)}")
print(f"Current user profile: {tools.userprofile}")
if name == "main":
rundemo()
Final Thoughts: Advantages of the Control-Plane Pattern in AI Tutors
Through this exploration, we observe how the control-plane design pattern streamlines orchestration, enhances safety, and cleanly separates reasoning from tool execution. The integration of a retrieval system, modular tool registry, and agentic planning layer culminates in a coherent AI tutor that intelligently addresses student inquiries. Experimenting with the demo reveals how tasks are routed, rules enforced, and insights synthesized, all within a flexible and extensible architecture.
