Multi-agent AI systems usually mean cloud APIs, per-token bills and opaque orchestration layers. This tutorial takes the opposite route: a team of specialized AI agents coordinated by a manager agent, running entirely on local hardware with TinyLlama, a compact 1.1-billion-parameter open model. Everything runs through the Hugging Face Transformers library — no external APIs — producing an offline, lightweight and fully inspectable multi-agent system covering structured task decomposition, inter-agent collaboration and autonomous logic loops.
How the system fits together
In plain terms, the architecture works like a small project team. A manager agent receives a high-level goal, breaks it into subtasks, and hands each subtask to a specialist agent. The specialists do their work using the local model, the manager respects dependencies between tasks (task C waits for A and B), and a final synthesis step merges all outputs into one coherent answer. Because every component is a few dozen lines of Python, the whole pipeline can be inspected, customized and extended.
Step 1: Imports and core data structures
!pip install transformers torch accelerate bitsandbytes -q
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
import json
import re
from typing import List, Dict, Any
from dataclasses import dataclass, asdict
from datetime import datetime
@dataclass
class Task:
id: str
description: str
assigned_to: str = None
status: str = "pending"
result: Any = None
dependencies: List(str) = None
def __post_init__(self):
if self.dependencies is None:
self.dependencies = ()
@dataclass
class Agent:
name: str
role: str
expertise: str
system_prompt: strThe first block sets up the key imports and defines the basic data structures for managing tasks and agents. Modeling tasks and agents as structured units gives every later component a consistent, reliable foundation to build on.
Step 2: Registering agents and the local LLM wrapper
AGENT_REGISTRY = {
"researcher": Agent(
name="researcher",
role="Research Specialist",
expertise="Information gathering, analysis, and synthesis",
system_prompt="You are a research specialist. Provide thorough research on topics."
),
"coder": Agent(
name="coder",
role="Software Engineer",
expertise="Writing clean, efficient code with best practices",
system_prompt="You are an expert programmer. Write clean, well-documented code."
),
"writer": Agent(
name="writer",
role="Content Writer",
expertise="Clear communication and documentation",
system_prompt="You are a professional writer. Create clear, engaging content."
),
"analyst": Agent(
name="analyst",
role="Data Analyst",
expertise="Data interpretation and insights",
system_prompt="You are a data analyst. Provide clear insights from data."
)
}
class LocalLLM:
def __init__(self, model_name: str = "TinyLlama/TinyLlama-1.1B-Chat-v1.0"):
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
quantization_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_compute_dtype=torch.float16
) if torch.cuda.is_available() else None
self.model = AutoModelForCausalLM.from_pretrained(
model_name,
quantization_config=quantization_config,
device_map="auto",
low_cpu_mem_usage=True
)
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
def generate(self, prompt: str, max_tokens: int = 300) -> str:
formatted_prompt = f"<|system|>nYou are a helpful AI assistant.n<|user|>n{prompt}n<|assistant|>n"
inputs = self.tokenizer(
formatted_prompt,
return_tensors="pt",
truncation=True,
max_length=1024,
padding=True
)
inputs = {k: v.to(self.model.device) for k, v in inputs.items()}
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=0.7,
do_sample=True,
top_p=0.9,
pad_token_id=self.tokenizer.pad_token_id,
eos_token_id=self.tokenizer.eos_token_id,
use_cache=True
)
full_response = self.tokenizer.decode(outputs(0), skip_special_tokens=True)
if "<|assistant|>" in full_response:
return full_response.split("<|assistant|>")(-1).strip()
return full_response(len(formatted_prompt):).strip()This section registers the specialist agents and implements the local LLM wrapper that powers the system. TinyLlama is loaded in efficient 4-bit mode, so the whole stack runs comfortably in Google Colab or on modest local hardware — a flexible, completely local way to generate responses for each agent.
Step 3: The manager agent and task decomposition
class ManagerAgent:
def __init__(self, model_name: str = "TinyLlama/TinyLlama-1.1B-Chat-v1.0"):
self.llm = LocalLLM(model_name)
self.agents = AGENT_REGISTRY
self.tasks: Dict(str, Task) = {}
self.execution_log = ()
def log(self, message: str):
timestamp = datetime.now().strftime("%H:%M:%S")
log_entry = f"({timestamp}) {message}"
self.execution_log.append(log_entry)
print(log_entry)
def decompose_goal(self, goal: str) -> List(Task):
self.log(f"🎯 Decomposing goal: {goal}")
agent_info = "n".join((f"- {name}: {agent.expertise}" for name, agent in self.agents.items()))
prompt = f"""Break down this goal into 3 specific subtasks. Assign each to the best agent.
Goal: {goal}
Available agents:
{agent_info}
Respond ONLY with a JSON array."""
response = self.llm.generate(prompt, max_tokens=250)
try:
json_match = re.search(r'(s*{.*?}s*)', response, re.DOTALL)
if json_match:
tasks_data = json.loads(json_match.group())
else:
raise ValueError("No JSON found")
except:
tasks_data = self._create_default_tasks(goal)
tasks = ()
for i, task_data in enumerate(tasks_data(:3)):
task = Task(
id=task_data.get('id', f'task_{i+1}'),
description=task_data.get('description', f'Work on: {goal}'),
assigned_to=task_data.get('assigned_to', list(self.agents.keys())(i % len(self.agents))),
dependencies=task_data.get('dependencies', () if i == 0 else (f'task_{i}'))
)
self.tasks(task.id) = task
tasks.append(task)
self.log(f" ✓ {task.id}: {task.description(:50)}... → {task.assigned_to}")
return tasksThe ManagerAgent class is where the intelligence lives: it decomposes a high-level goal into well-defined subtasks, generating structured JSON-based tasks and automatically assigning each to the right agent. The system organizes work step by step, much like a human project manager would.
Step 4: Execution flow and fallback logic
def _create_default_tasks(self, goal: str) -> List(Dict):
if any(word in goal.lower() for word in ('code', 'program', 'implement', 'algorithm')):
return (
{"id": "task_1", "description": f"Research and explain the concept: {goal}", "assigned_to": "researcher", "dependencies": ()},
{"id": "task_2", "description": f"Write code implementation for: {goal}", "assigned_to": "coder", "dependencies": ("task_1")},
{"id": "task_3", "description": f"Create documentation and examples", "assigned_to": "writer", "dependencies": ("task_2")}
)
return (
{"id": "task_1", "description": f"Research: {goal}", "assigned_to": "researcher", "dependencies": ()},
{"id": "task_2", "description": f"Analyze findings and structure content", "assigned_to": "analyst", "dependencies": ("task_1")},
{"id": "task_3", "description": f"Write comprehensive response", "assigned_to": "writer", "dependencies": ("task_2")}
)
def execute_task(self, task: Task, context: Dict(str, Any) = None) -> str:
self.log(f"🤖 Executing {task.id} with {task.assigned_to}")
task.status = "in_progress"
agent = self.agents(task.assigned_to)
context_str = ""
if context and task.dependencies:
context_str = "nnContext from previous tasks:n"
for dep_id in task.dependencies:
if dep_id in context:
context_str += f"- {context(dep_id)(:150)}...n"
prompt = f"""{agent.system_prompt}
Task: {task.description}{context_str}
Provide a clear, concise response:"""
result = self.llm.generate(prompt, max_tokens=250)
task.result = result
task.status = "completed"
self.log(f" ✓ Completed {task.id}")
return resultThis block defines the fallback task logic and the complete execution flow for each task. Each agent is guided by its own system prompt and supplied with relevant context, which keeps results consistent while tasks execute in dependency order.
Step 5: Synthesis and the orchestration loop
def synthesize_results(self, goal: str, results: Dict(str, str)) -> str:
self.log("🔄 Synthesizing final results")
results_text = "nn".join((f"Task {tid}:n{res(:200)}" for tid, res in results.items()))
prompt = f"""Combine these task results into one final coherent answer.
Original Goal: {goal}
Task Results:
{results_text}
Final comprehensive answer:"""
return self.llm.generate(prompt, max_tokens=350)
def execute_goal(self, goal: str) -> Dict(str, Any):
self.log(f"n{'='*60}n🎬 Starting Manager Agentn{'='*60}")
tasks = self.decompose_goal(goal)
results = {}
completed = set()
max_iterations = len(tasks) * 2
iteration = 0
while len(completed) < len(tasks) and iteration < max_iterations:
iteration += 1
for task in tasks:
if task.id in completed:
continue
deps_met = all(dep in completed for dep in task.dependencies)
if deps_met:
result = self.execute_task(task, results)
results(task.id) = result
completed.add(task.id)
final_output = self.synthesize_results(goal, results)
self.log(f"n{'='*60}n✅ Execution Complete!n{'='*60}n")
return {
"goal": goal,
"tasks": (asdict(task) for task in tasks),
"final_output": final_output,
"execution_log": self.execution_log
}Outputs from all subtasks are synthesized into an integrated final answer, and an orchestration loop ensures each task runs only after its dependencies are satisfied — bringing everything together into a seamless multi-step pipeline.
Step 6: Demonstration tasks
def demo_basic():
manager = ManagerAgent()
goal = "Explain binary search algorithm with a simple example"
result = manager.execute_goal(goal)
print("n" + "="*60)
print("FINAL OUTPUT")
print("="*60)
print(result("final_output"))
return result
def demo_coding():
manager = ManagerAgent()
goal = "Implement a function to find the maximum element in a list"
result = manager.execute_goal(goal)
print("n" + "="*60)
print("FINAL OUTPUT")
print("="*60)
print(result("final_output"))
return result
def demo_custom(custom_goal: str):
manager = ManagerAgent()
result = manager.execute_goal(custom_goal)
print("n" + "="*60)
print("FINAL OUTPUT")
print("="*60)
print(result("final_output"))
return result
if __name__ == "__main__":
print("🤖 Manager Agent Tutorial - APIless Local Version")
print("="*60)
print("Using TinyLlama (1.1B) - Fast & efficient!n")
result = demo_basic()
print("nn💡 Try more:")
print(" - demo_coding()")
print(" - demo_custom('your goal here')")Demonstration tasks make it easy to test the system with different goals and watch the manager decompose, execute and synthesize work in real time — an interactive way to understand the workflow before refining it.
Limitations and what to watch
A 1.1B-parameter model is the main constraint: TinyLlama handles structured, well-prompted subtasks but will not match larger models on open-ended reasoning, and JSON output from small models can be brittle — hence the fallback logic above. The design is educational rather than production-grade: there is no persistence, retry budget or safety layer, topics covered in why AI agent projects stall without governance. For heavier local workloads, the same pattern transfers to larger open models, and compression-native approaches like Apple’s CLaRa hint at where efficient local retrieval is heading.
The takeaway
A complete multi-agent orchestration system — manager, specialists, dependency resolution and synthesis — can run locally with minimal dependencies. Built from scratch, the pattern is modular and predictable, and it demystifies what commercial agent frameworks do behind their APIs.
