On this tutorial, you’ll construct a self-validating DataOps AIAgent that may mechanically plan, execute, and check knowledge operations utilizing a neighborhood Hugging Face mannequin. Brokers are designed with three clever roles. A planner to create an execution technique, an executor to put in writing and run the code utilizing pandas, and a tester to confirm the accuracy and consistency of the outcomes. Through the use of Microsoft’s Phi-2 mannequin domestically in Google Colab, we display how LLM can automate advanced knowledge processing duties end-to-end whereas making certain workflow effectivity, reproducibility, and privateness safety. Please verify Full code here.
!pip set up -q transformers speed up bitsandbytes scipy
import json, pandas as pd, numpy as np, torch
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline, BitsAndBytesConfig
MODEL_NAME = "microsoft/phi-2"
class LocalLLM:
def __init__(self, model_name=MODEL_NAME, use_8bit=False):
print(f"Loading mannequin: {model_name}")
self.tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
model_kwargs = {"device_map": "auto", "trust_remote_code": True}
if use_8bit and torch.cuda.is_available():
model_kwargs["quantization_config"] = BitsAndBytesConfig(load_in_8bit=True)
else:
model_kwargs["torch_dtype"] = torch.float32 if not torch.cuda.is_available() else torch.float16
self.mannequin = AutoModelForCausalLM.from_pretrained(model_name, **model_kwargs)
self.pipe = pipeline("text-generation", mannequin=self.mannequin, tokenizer=self.tokenizer,
max_new_tokens=512, do_sample=True, temperature=0.3, top_p=0.9,
pad_token_id=self.tokenizer.eos_token_id)
print("✓ Mannequin loaded efficiently!n")
def generate(self, immediate, system_prompt="", temperature=0.3):
if system_prompt:
full_prompt = f"Instruct: {system_prompt}nn{immediate}nOutput:"
else:
full_prompt = f"Instruct: {immediate}nOutput:"
output = self.pipe(full_prompt, temperature=temperature, do_sample=temperature>0,
return_full_text=False, eos_token_id=self.tokenizer.eos_token_id)
end result = output[0]['generated_text'].strip()
if "Instruct:" in end result:
end result = end result.cut up("Instruct:")[0].strip()
return end resultSet up the required libraries and use Hugging Face Transformers to load the Phi-2 mannequin domestically. Create a LocalLLM class that initializes the tokenizer and mannequin, helps non-compulsory quantization, and defines a technology technique that produces textual content output. It ensures that your mannequin runs easily on each CPU and GPU, making it ultimate to be used with Colab. Please verify Full code here.
PLANNER_PROMPT = """You're a Knowledge Operations Planner. Create an in depth execution plan as legitimate JSON.
Return ONLY a JSON object (no different textual content) with this construction:
{"steps": ["step 1","step 2"],"expected_output":"description","validation_criteria":["criteria 1","criteria 2"]}"""
EXECUTOR_PROMPT = """You're a Knowledge Operations Executor. Write Python code utilizing pandas.
Necessities:
- Use pandas (imported as pd) and numpy (imported as np)
- Retailer closing lead to variable 'end result'
- Return ONLY Python code, no explanations or markdown"""
TESTER_PROMPT = """You're a Knowledge Operations Tester. Confirm execution outcomes.
Return ONLY a JSON object (no different textual content) with this construction:
{"handed":true,"points":["any issues found"],"suggestions":["suggestions"]}"""
class DataOpsAgent:
def __init__(self, llm=None):
self.llm = llm or LocalLLM()
self.historical past = []
def _extract_json(self, textual content):
attempt:
return json.masses(textual content)
besides:
begin, finish = textual content.discover('{'), textual content.rfind('}')+1
if begin >= 0 and finish > begin:
attempt:
return json.masses(textual content[start:end])
besides:
cross
return NoneOutline system prompts for DataOps agent planner, executor, and tester roles. Subsequent, initialize the DataOpsAgent class utilizing helper strategies and a JSON extraction utility to parse the structured response. Put together the muse for the agent’s inference and execution pipeline. Please verify Full code here.
def plan(self, job, data_info):
print("n" + "="*60)
print("PHASE 1: PLANNING")
print("="*60)
immediate = f"Job: {job}nnData Data:n{data_info}nnCreate an execution plan as JSON with steps, expected_output, and validation_criteria."
plan_text = self.llm.generate(immediate, PLANNER_PROMPT, temperature=0.2)
self.historical past.append(("PLANNER", plan_text))
plan = self._extract_json(plan_text) or {"steps":[task],"expected_output":"Processed knowledge","validation_criteria":["Result generated","No errors"]}
print(f"n📋 Plan Created:")
print(f" Steps: {len(plan.get('steps', []))}")
for i, step in enumerate(plan.get('steps', []), 1):
print(f" {i}. {step}")
print(f" Anticipated: {plan.get('expected_output', 'N/A')}")
return plan
def execute(self, plan, data_context):
print("n" + "="*60)
print("PHASE 2: EXECUTION")
print("="*60)
steps_text="n".be part of(f"{i}. {s}" for i, s in enumerate(plan.get('steps', []), 1))
immediate = f"Job Steps:n{steps_text}nnData out there: DataFrame 'df'n{data_context}nnWrite Python code to execute these steps. Retailer closing lead to 'end result' variable."
code = self.llm.generate(immediate, EXECUTOR_PROMPT, temperature=0.1)
self.historical past.append(("EXECUTOR", code))
if "```python" in code: code = code.cut up("```python")[1].cut up("```")[0]
elif "```" in code: code = code.cut up("```")[1].cut up("```")[0]
strains = []
for line in code.cut up('n'):
s = line.strip()
if s and (not s.startswith('#') or 'import' in s):
strains.append(line)
code="n".be part of(strains).strip()
print(f"n💻 Generated Code:n" + "-"*60)
for i, line in enumerate(code.cut up('n')[:15],1):
print(f"{i:2}. {line}")
if len(code.cut up('n'))>15: print(f" ... ({len(code.cut up('n'))-15} extra strains)")
print("-"*60)
return codeImplement the agent’s planning and execution phases. Let Planner create detailed job steps and validation standards, and Executor will generate the corresponding Python code primarily based on Pandas to execute the duty. Visualize an agent autonomously transferring from inference to producing executable code. Please verify Full code here.
def check(self, plan, end result, execution_error=None):
print("n" + "="*60)
print("PHASE 3: TESTING & VERIFICATION")
print("="*60)
result_desc = f"EXECUTION ERROR: {execution_error}" if execution_error else f"End result sort: {sort(end result).__name__}n"
if not execution_error:
if isinstance(end result, pd.DataFrame):
result_desc += f"Form: {end result.form}nColumns: {record(end result.columns)}nSample:n{end result.head(3).to_string()}"
elif isinstance(end result, (int,float,str)):
result_desc += f"Worth: {end result}"
else:
result_desc += f"Worth: {str(end result)[:200]}"
criteria_text="n".be part of(f"- {c}" for c in plan.get('validation_criteria', []))
immediate = f"Validation Standards:n{criteria_text}nnExpected: {plan.get('expected_output', 'N/A')}nnActual End result:n{result_desc}nnEvaluate if end result meets standards. Return JSON with handed (true/false), points, and proposals."
test_result = self.llm.generate(immediate, TESTER_PROMPT, temperature=0.2)
self.historical past.append(("TESTER", test_result))
test_json = self._extract_json(test_result) or {"handed":execution_error is None,"points":["Could not parse test result"],"suggestions":["Review manually"]}
print(f"n✓ Take a look at Outcomes:n Standing: {'✅ PASSED' if test_json.get('handed') else '❌ FAILED'}")
if test_json.get('points'):
print(" Points:")
for situation in test_json['issues'][:3]:
print(f" • {situation}")
if test_json.get('suggestions'):
print(" Suggestions:")
for rec in test_json['recommendations'][:3]:
print(f" • {rec}")
return test_json
def run(self, job, df=None, data_info=None):
print("n🤖 SELF-VERIFYING DATA-OPS AGENT (Native HF Mannequin)")
print(f"Job: {job}n")
if data_info is None and df just isn't None:
data_info = f"Form: {df.form}nColumns: {record(df.columns)}nSample:n{df.head(2).to_string()}"
plan = self.plan(job, data_info)
code = self.execute(plan, data_info)
end result, error = None, None
attempt:
local_vars = {'pd': pd, 'np': np, 'df': df}
exec(code, local_vars)
end result = local_vars.get('end result')
besides Exception as e:
error = str(e)
print(f"n⚠️ Execution Error: {error}")
test_result = self.check(plan, end result, error)
return {'plan': plan,'code': code,'end result': end result,'check': test_result,'historical past': self.historical past}We give attention to the testing and validation part of the workflow. Have the agent consider its output in opposition to predefined validation standards and summarize the outcomes as structured JSON. It then integrates all three phases of planning, execution, and testing right into a single self-validating pipeline, making certain full automation. Please verify Full code here.
def demo_basic(agent):
print("n" + "#"*60)
print("# DEMO 1: Gross sales Knowledge Aggregation")
print("#"*60)
df = pd.DataFrame({'product':['A','B','A','C','B','A','C'],
'gross sales':[100,150,200,80,130,90,110],
'area':['North','South','North','East','South','West','East']})
job = "Calculate complete gross sales by product"
output = agent.run(job, df)
if output['result'] just isn't None:
print(f"n📊 Remaining End result:n{output['result']}")
return output
def demo_advanced(agent):
print("n" + "#"*60)
print("# DEMO 2: Buyer Age Evaluation")
print("#"*60)
df = pd.DataFrame({'customer_id':vary(1,11),
'age':[25,34,45,23,56,38,29,41,52,31],
'purchases':[5,12,8,3,15,7,9,11,6,10],
'spend':[500,1200,800,300,1500,700,900,1100,600,1000]})
job = "Calculate common spend by age group: younger (beneath 35) and mature (35+)"
output = agent.run(job, df)
if output['result'] just isn't None:
print(f"n📊 Remaining End result:n{output['result']}")
return output
if __name__ == "__main__":
print("🚀 Initializing Native LLM...")
print("Utilizing CPU mode for max compatibilityn")
attempt:
llm = LocalLLM(use_8bit=False)
agent = DataOpsAgent(llm)
demo_basic(agent)
print("nn")
demo_advanced(agent)
print("n" + "="*60)
print("✅ Tutorial Full!")
print("="*60)
print("nKey Options:")
print(" • 100% Native - No API calls required")
print(" • Makes use of Phi-2 from Microsoft (2.7B params)")
print(" • Self-verifying 3-phase workflow")
print(" • Runs on free Google Colab CPU/GPU")
besides Exception as e:
print(f"n❌ Error: {e}")
print("Troubleshooting:n1. pip set up -q transformers speed up scipyn2. Restart runtimen3. Strive a distinct mannequin")We constructed two demo samples to check agent performance utilizing a easy gross sales dataset and a buyer dataset. Initialize fashions, run Knowledge-Ops workflows, and observe your entire cycle from planning to validation. We conclude the tutorial by summarizing the principle advantages and inspiring additional experimentation with native fashions.
In conclusion, we’ve got created a completely autonomous, self-validating DataOps system that leverages a neighborhood Hugging Face mannequin. We’ll expertise how the planning, execution, and testing phases work collectively seamlessly to supply dependable outcomes with out counting on cloud APIs. This workflow highlights the power of native LLMs reminiscent of Phi-2 for light-weight automation and conjures up us to increase this structure for extra superior knowledge pipelines, validation frameworks, and multi-agent knowledge methods sooner or later.
Please verify Full code here. Please be happy to test it out GitHub page for tutorials, code, and notebooks. Additionally, be happy to comply with us Twitter Do not forget to affix us 100,000+ ML subreddits and subscribe our newsletter. cling on! Are you on telegram? You can now also participate by telegram.
Asif Razzaq is the CEO of Marktechpost Media Inc. Asif is a visionary entrepreneur and engineer dedicated to harnessing the potential of synthetic intelligence for social good. His newest endeavor is the launch of Marktechpost, a man-made intelligence media platform. It stands out for its thorough protection of machine studying and deep studying information, which is technically sound and simply understood by a large viewers. The platform boasts over 2 million views monthly, demonstrating its recognition amongst viewers.
