How to Design an Autonomous Multi-Agent Data and Infrastructure Strategy System Using Lightweight Qwen Models for Efficient Pipeline Intelligence

This tutorial outlines the development of an Agentic Data and Infrastructure Strategy system utilizing the lightweight Qwen2.5-0.5B-Instruct model to enhance data pipeline efficiency. We will create a flexible LLM agent framework and specialize agents for various data management layers, including ingestion, quality analysis, and infrastructure optimization. These agents will be integrated into an orchestrator to ensure seamless collaboration throughout the data pipeline. We will illustrate the application of this system with examples from e-commerce and IoT pipelines, demonstrating how autonomous decision-making can streamline complex data operations.

Target Audience Analysis

The primary audience for this tutorial includes:

• Data Engineers: Professionals tasked with building and maintaining data pipelines.
• Data Scientists: Individuals focused on analyzing and interpreting complex data sets.
• AI/ML Practitioners: Experts interested in implementing machine learning models for data management.
• Business Analysts: Stakeholders who rely on accurate data for decision-making.

Pain Points: Difficulty in managing complex data pipelines, ensuring data quality, and optimizing infrastructure resources.

Goals: Streamlining data ingestion processes, improving data quality assessments, and enhancing overall pipeline efficiency.

Interests: Innovations in AI-driven data management, practical applications of multi-agent systems, and effective data infrastructure strategies.

Communication Preferences: Clear, concise technical documentation with practical examples, supported by code snippets and visual aids.

Building the Agentic Data and Infrastructure Strategy System

We will set up a lightweight LLM agent infrastructure using the Qwen2.5-0.5B-Instruct model. The initial step involves loading the model and tokenizer, followed by defining a base agent class capable of managing contextual conversations and generating intelligent responses. This structure serves as the foundation for our specialized agents that will operate efficiently within a collaborative environment.

Lightweight LLM Agent Class

class LightweightLLMAgent:
   def __init__(self, role: str, model_name: str = "Qwen/Qwen2.5-0.5B-Instruct"):
       self.role = role
       self.model_name = model_name
       self.device = "cuda" if torch.cuda.is_available() else "cpu"
       self.tokenizer = AutoTokenizer.from_pretrained(model_name)
       self.model = AutoModelForCausalLM.from_pretrained(
           model_name,
           torch_dtype=torch.float16 if self.device == "cuda" else torch.float32,
           device_map="auto"
       )
       self.conversation_history = []

   def generate_response(self, prompt: str, max_tokens: int = 150) -> str:
       messages = [
           {"role": "system", "content": f"You are a {self.role} agent in a data infrastructure system."},
           {"role": "user", "content": prompt}
       ]
       text = self.tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
       model_inputs = self.tokenizer([text], return_tensors="pt").to(self.device)
       with torch.no_grad():
           generated_ids = self.model.generate(
               model_inputs.input_ids,
               max_new_tokens=max_tokens,
               temperature=0.7,
               do_sample=True,
               top_p=0.95
           )
       generated_ids = [output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)]
       response = self.tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
       self.conversation_history.append({"prompt": prompt, "response": response})
       return response

Data Ingestion Agent

class DataIngestionAgent(LightweightLLMAgent):
   def __init__(self):
       super().__init__(role="Data Ingestion Specialist")
   def analyze_data_source(self, source_info: Dict) -> Dict:
       prompt = f"""Analyze this data source and provide ingestion strategy:
Source Type: {source_info.get('type', 'unknown')}
Volume: {source_info.get('volume', 'unknown')}
Frequency: {source_info.get('frequency', 'unknown')}
Provide a brief strategy focusing on: 1) Ingestion method, 2) Key considerations."""
       strategy = self.generate_response(prompt, max_tokens=100)
       return {"source": source_info, "strategy": strategy, "timestamp": datetime.now().isoformat()}

Data Quality Agent

class DataQualityAgent(LightweightLLMAgent):
   def __init__(self):
       super().__init__(role="Data Quality Analyst")
   def assess_data_quality(self, data_sample: Dict) -> Dict:
       prompt = f"""Assess data quality for this sample:
Completeness: {data_sample.get('completeness', 'N/A')}%
Consistency: {data_sample.get('consistency', 'N/A')}%
Issues Found: {data_sample.get('issues', 0)}
Provide brief quality assessment and top 2 recommendations."""
       assessment = self.generate_response(prompt, max_tokens=100)
       return {"assessment": assessment, "severity": self._calculate_severity(data_sample), "timestamp": datetime.now().isoformat()}
   def _calculate_severity(self, data_sample: Dict) -> str:
       completeness = data_sample.get('completeness', 100)
       consistency = data_sample.get('consistency', 100)
       avg_score = (completeness + consistency) / 2
       if avg_score >= 90: return "LOW"
       elif avg_score >= 70: return "MEDIUM"
       else: return "HIGH"

Infrastructure Optimization Agent

class InfrastructureOptimizationAgent(LightweightLLMAgent):
   def __init__(self):
       super().__init__(role="Infrastructure Optimization Specialist")
   def optimize_resources(self, metrics: Dict) -> Dict:
       prompt = f"""Analyze infrastructure metrics and suggest optimizations:
CPU Usage: {metrics.get('cpu_usage', 0)}%
Memory Usage: {metrics.get('memory_usage', 0)}%
Storage: {metrics.get('storage_used', 0)}GB / {metrics.get('storage_total', 0)}GB
Query Latency: {metrics.get('query_latency', 0)}ms
Provide 2 optimization recommendations."""
       recommendations = self.generate_response(prompt, max_tokens=100)
       return {"current_metrics": metrics, "recommendations": recommendations, "priority": self._calculate_priority(metrics), "timestamp": datetime.now().isoformat()}
   def _calculate_priority(self, metrics: Dict) -> str:
       cpu = metrics.get('cpu_usage', 0)
       memory = metrics.get('memory_usage', 0)
       if cpu > 85 or memory > 85: return "CRITICAL"
       elif cpu > 70 or memory > 70: return "HIGH"
       else: return "NORMAL"

Agentic Data Orchestrator

class AgenticDataOrchestrator:
   def __init__(self):
       self.ingestion_agent = DataIngestionAgent()
       self.quality_agent = DataQualityAgent()
       self.optimization_agent = InfrastructureOptimizationAgent()
       self.execution_log = []
   def process_data_pipeline(self, pipeline_config: Dict) -> Dict:
       results = {"pipeline_id": pipeline_config.get("id", "unknown"), "start_time": datetime.now().isoformat(), "stages": []}
       ingestion_result = self.ingestion_agent.analyze_data_source(pipeline_config.get("source", {}))
       results["stages"].append({"stage": "ingestion", "result": ingestion_result})
       quality_result = self.quality_agent.assess_data_quality(pipeline_config.get("quality_metrics", {}))
       results["stages"].append({"stage": "quality", "result": quality_result})
       optimization_result = self.optimization_agent.optimize_resources(pipeline_config.get("infrastructure_metrics", {}))
       results["stages"].append({"stage": "optimization", "result": optimization_result})
       results["end_time"] = datetime.now().isoformat()
       results["status"] = "completed"
       self.execution_log.append(results)
       return results
   def generate_summary_report(self) -> pd.DataFrame:
       if not self.execution_log: return pd.DataFrame()
       summary_data = []
       for log in self.execution_log:
           summary_data.append({"Pipeline ID": log["pipeline_id"], "Start Time": log["start_time"], "Status": log["status"], "Stages Completed": len(log["stages"])})
       return pd.DataFrame(summary_data)

Example Implementations

We demonstrate the complete system through two real-world examples, an e-commerce and an IoT data pipeline. Each agent performs its role autonomously while contributing to a shared objective.

E-commerce Data Pipeline

ecommerce_pipeline = {
   "id": "ecommerce_pipeline_001",
   "source": {"type": "REST API", "volume": "10 GB/day", "frequency": "real-time"},
   "quality_metrics": {"completeness": 87, "consistency": 92, "issues": 15},
   "infrastructure_metrics": {"cpu_usage": 78, "memory_usage": 82, "storage_used": 450, "storage_total": 1000, "query_latency": 250}
}
result1 = orchestrator.process_data_pipeline(ecommerce_pipeline)

IoT Sensor Data Pipeline

iot_pipeline = {
   "id": "iot_pipeline_002",
   "source": {"type": "Message Queue (Kafka)", "volume": "50 GB/day", "frequency": "streaming"},
   "quality_metrics": {"completeness": 95, "consistency": 88, "issues": 8},
   "infrastructure_metrics": {"cpu_usage": 65, "memory_usage": 71, "storage_used": 780, "storage_total": 2000, "query_latency": 180}
}
result2 = orchestrator.process_data_pipeline(iot_pipeline)

Execution Summary Report

The orchestrator generates a summary report, confirming the orchestration’s efficiency and the power of lightweight agentic intelligence.

In conclusion, we have designed and executed an intelligent, multi-agent data infrastructure framework powered by an open-source model. This setup demonstrates how independent yet cooperative agents can autonomously analyze, assess, and optimize real-world data systems, transforming traditional data workflows into adaptive, self-optimizing systems ready for scalable enterprise applications.

Check out the FULL CODES here. You can also explore more tutorials and resources on our GitHub Page. Join our community on Twitter and subscribe to our Newsletter. If you’re on Telegram, join us there as well.