How to Design a Fully Local Multi-Agent Orchestration System Using TinyLlama for Intelligent Task Decomposition and Autonomous Collaboration

!pip set up transformers torch speed up 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
   standing: str = "pending"
   end result: Any = None
   dependencies: List[str] = None
  
   def __post_init__(self):
       if self.dependencies is None:
           self.dependencies = []


@dataclass
class Agent:
   identify: str
   position: str
   experience: str
   system_prompt: str

AGENT_REGISTRY = {
   "researcher": Agent(
       identify="researcher",
       position="Research Specialist",
       experience="Information gathering, evaluation, and synthesis",
       system_prompt="You are a analysis specialist. Provide thorough analysis on subjects."
   ),
   "coder": Agent(
       identify="coder",
       position="Software Engineer",
       experience="Writing clear, environment friendly code with greatest practices",
       system_prompt="You are an skilled programmer. Write clear, well-documented code."
   ),
   "author": Agent(
       identify="author",
       position="Content Writer",
       experience="Clear communication and documentation",
       system_prompt="You are a skilled author. Create clear, partaking content material."
   ),
   "analyst": Agent(
       identify="analyst",
       position="Data Analyst",
       experience="Data interpretation and insights",
       system_prompt="You are a information analyst. Provide clear insights from information."
   )
}


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.mannequin = 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, immediate: str, max_tokens: int = 300) -> str:
       formatted_prompt = f"<|system|>nYou are a useful AI assistant.</s>n<|person|>n{immediate}</s>n<|assistant|>n"
       inputs = self.tokenizer(
           formatted_prompt,
           return_tensors="pt",
           truncation=True,
           max_length=1024,
           padding=True
       )
       inputs = {okay: v.to(self.mannequin.gadget) for okay, v in inputs.objects()}
       with torch.no_grad():
           outputs = self.mannequin.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.break up("<|assistant|>")[-1].strip()
       return full_response[len(formatted_prompt):].strip()

class ManagerAgent:
   def __init__(self, model_name: str = "TinyLlama/TinyLlama-1.1B-Chat-v1.0"):
       self.llm = LocalLLM(model_name)
       self.brokers = AGENT_REGISTRY
       self.duties: 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, objective: str) -> List[Task]:
       self.log(f" Decomposing objective: {objective}")
       agent_info = "n".be part of([f"- {name}: {agent.expertise}" for name, agent in self.agents.items()])
       immediate = f"""Break down this objective into 3 particular subtasks. Assign every to the very best agent.


Goal: {objective}


Available brokers:
{agent_info}


Respond ONLY with a JSON array."""
       response = self.llm.generate(immediate, max_tokens=250)
       attempt:
           json_match = re.search(r'[s*{.*?}s*]', response, re.DOTALL)
           if json_match:
               tasks_data = json.hundreds(json_match.group())
           else:
               elevate ValueError("No JSON discovered")
       besides:
           tasks_data = self._create_default_tasks(objective)
      
       duties = []
       for i, task_data in enumerate(tasks_data[:3]):
           process = Task(
               id=task_data.get('id', f'task_{i+1}'),
               description=task_data.get('description', f'Work on: {objective}'),
               assigned_to=task_data.get('assigned_to', listing(self.brokers.keys())[i % len(self.agents)]),
               dependencies=task_data.get('dependencies', [] if i == 0 else [f'task_{i}'])
           )
           self.duties[task.id] = process
           duties.append(process)
           self.log(f"  ✓ {process.id}: {process.description[:50]}... → {process.assigned_to}")
      
       return duties

 def _create_default_tasks(self, objective: str) -> List[Dict]:
       if any(phrase in objective.decrease() for phrase 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: {objective}", "assigned_to": "coder", "dependencies": ["task_1"]},
               {"id": "task_3", "description": f"Create documentation and examples", "assigned_to": "author", "dependencies": ["task_2"]}
           ]
       return [
           {"id": "task_1", "description": f"Research: {goal}", "assigned_to": "researcher", "dependencies": []},
           {"id": "task_2", "description": f"Analyze findings and construction content material", "assigned_to": "analyst", "dependencies": ["task_1"]},
           {"id": "task_3", "description": f"Write complete response", "assigned_to": "author", "dependencies": ["task_2"]}
       ]
  
   def execute_task(self, process: Task, context: Dict[str, Any] = None) -> str:
       self.log(f" Executing {process.id} with {process.assigned_to}")
       process.standing = "in_progress"
       agent = self.brokers[task.assigned_to]
       context_str = ""
       if context and process.dependencies:
           context_str = "nnContext from earlier duties:n"
           for dep_id in process.dependencies:
               if dep_id in context:
                   context_str += f"- {context[dep_id][:150]}...n"
      
       immediate = f"""{agent.system_prompt}


Task: {process.description}{context_str}


Provide a clear, concise response:"""
       end result = self.llm.generate(immediate, max_tokens=250)
       process.end result = end result
       process.standing = "accomplished"
       self.log(f"  ✓ Completed {process.id}")
       return end result

def synthesize_results(self, objective: str, outcomes: Dict[str, str]) -> str:
       self.log(" Synthesizing last outcomes")
       results_text = "nn".be part of([f"Task {tid}:n{res[:200]}" for tid, res in outcomes.objects()])
       immediate = f"""Combine these process outcomes into one last coherent reply.


Original Goal: {objective}


Task Results:
{results_text}


Final complete reply:"""
       return self.llm.generate(immediate, max_tokens=350)
  
   def execute_goal(self, objective: str) -> Dict[str, Any]:
       self.log(f"n{'='*60}n Starting Manager Agentn{'='*60}")
       duties = self.decompose_goal(objective)
       outcomes = {}
       accomplished = set()
       max_iterations = len(duties) * 2
       iteration = 0
      
       whereas len(accomplished) < len(duties) and iteration < max_iterations:
           iteration += 1
           for process in duties:
               if process.id in accomplished:
                   proceed
               deps_met = all(dep in accomplished for dep in process.dependencies)
               if deps_met:
                   end result = self.execute_task(process, outcomes)
                   outcomes[task.id] = end result
                   accomplished.add(process.id)
      
       final_output = self.synthesize_results(objective, outcomes)
       self.log(f"n{'='*60}n Execution Complete!n{'='*60}n")
      
       return {
           "objective": objective,
           "duties": [asdict(task) for task in tasks],
           "final_output": final_output,
           "execution_log": self.execution_log
       }

def demo_basic():
   supervisor = ManagerAgent()
   objective = "Explain binary search algorithm with a easy instance"
   end result = supervisor.execute_goal(objective)
   print("n" + "="*60)
   print("FINAL OUTPUT")
   print("="*60)
   print(end result["final_output"])
   return end result


def demo_coding():
   supervisor = ManagerAgent()
   objective = "Implement a perform to discover the utmost factor in a listing"
   end result = supervisor.execute_goal(objective)
   print("n" + "="*60)
   print("FINAL OUTPUT")
   print("="*60)
   print(end result["final_output"])
   return end result


def demo_custom(custom_goal: str):
   supervisor = ManagerAgent()
   end result = supervisor.execute_goal(custom_goal)
   print("n" + "="*60)
   print("FINAL OUTPUT")
   print("="*60)
   print(end result["final_output"])
   return end result


if __name__ == "__main__":
   print(" Manager Agent Tutorial - APIless Local Version")
   print("="*60)
   print("Using TinyLlama (1.1B) - Fast & environment friendly!n")
   end result = demo_basic()
   print("nn Try extra:")
   print("  - demo_coding()")
   print("  - demo_custom('your objective right here')")