In diesem Tutorial erstellen wir einen selbstverifizierenden DataOps AIAgent, der Datenoperationen mithilfe lokaler Hugging Face-Modelle automatisch planen, ausführen und testen kann. Wir entwerfen den Agenten mit drei intelligenten Rollen: einem Planer, der eine Ausführungsstrategie erstellt, einem Executor, der Code mit Pandas schreibt und ausführt, und einem Tester, der die Ergebnisse auf Genauigkeit und Konsistenz validiert. Durch die lokale Verwendung des Phi-2-Modells von Microsoft in Google Colab stellen wir sicher, dass der Workflow effizient, reproduzierbar und datenschutzkonform bleibt, und demonstrieren gleichzeitig, wie LLMs komplexe Datenverarbeitungsaufgaben durchgängig automatisieren können. Schauen Sie sich das an VOLLSTÄNDIGE CODES hier.
!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)
outcome = output(0)('generated_text').strip()
if "Instruct:" in outcome:
outcome = outcome.break up("Instruct:")(0).strip()
return outcomeWir installieren die erforderlichen Bibliotheken und laden das Phi-2-Modell lokal mithilfe von Hugging Face Transformers. Wir erstellen eine LocalLLM-Klasse, die den Tokenizer und das Modell initialisiert, optionale Quantisierung unterstützt und eine Generiermethode definiert, um Textausgaben zu erzeugen. Wir stellen sicher, dass das Modell sowohl auf der CPU als auch auf der GPU reibungslos läuft, sodass es sich superb für den Einsatz auf Colab eignet. Schauen Sie sich das an VOLLSTÄNDIGE CODES hier.
PLANNER_PROMPT = """You're a Information 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":("standards 1","standards 2")}"""
EXECUTOR_PROMPT = """You're a Information Operations Executor. Write Python code utilizing pandas.
Necessities:
- Use pandas (imported as pd) and numpy (imported as np)
- Retailer remaining end in variable 'outcome'
- Return ONLY Python code, no explanations or markdown"""
TESTER_PROMPT = """You're a Information Operations Tester. Confirm execution outcomes.
Return ONLY a JSON object (no different textual content) with this construction:
{"handed":true,"points":("any points discovered"),"suggestions":("recommendations")}"""
class DataOpsAgent:
def __init__(self, llm=None):
self.llm = llm or LocalLLM()
self.historical past = ()
def _extract_json(self, textual content):
strive:
return json.hundreds(textual content)
besides:
begin, finish = textual content.discover('{'), textual content.rfind('}')+1
if begin >= 0 and finish > begin:
strive:
return json.hundreds(textual content(begin:finish))
besides:
cross
return NoneWir definieren die Systemeingabeaufforderungen für die Rollen Planner, Executor und Tester unseres DataOps-Agenten. Anschließend initialisieren wir die DataOpsAgent-Klasse mit Hilfsmethoden und einem JSON-Extraktionsdienstprogramm, um strukturierte Antworten zu analysieren. Wir bereiten die Grundlage für die Argumentations- und Ausführungspipeline des Agenten. Schauen Sie sich das an VOLLSTÄNDIGE CODES hier.
def plan(self, process, data_info):
print("n" + "="*60)
print("PHASE 1: PLANNING")
print("="*60)
immediate = f"Activity: {process}nnData Info: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":(process),"expected_output":"Processed knowledge","validation_criteria":("Consequence 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 a part of(f"{i}. {s}" for i, s in enumerate(plan.get('steps', ()), 1))
immediate = f"Activity Steps:n{steps_text}nnData out there: DataFrame 'df'n{data_context}nnWrite Python code to execute these steps. Retailer remaining end in 'outcome' variable."
code = self.llm.generate(immediate, EXECUTOR_PROMPT, temperature=0.1)
self.historical past.append(("EXECUTOR", code))
if "```python" in code: code = code.break up("```python")(1).break up("```")(0)
elif "```" in code: code = code.break up("```")(1).break up("```")(0)
traces = ()
for line in code.break up('n'):
s = line.strip()
if s and (not s.startswith('#') or 'import' in s):
traces.append(line)
code="n".be a part of(traces).strip()
print(f"n💻 Generated Code:n" + "-"*60)
for i, line in enumerate(code.break up('n')(:15),1):
print(f"{i:2}. {line}")
if len(code.break up('n'))>15: print(f" ... ({len(code.break up('n'))-15} extra traces)")
print("-"*60)
return codeWir implementieren die Planungs- und Ausführungsphasen des Agenten. Wir lassen den Planer detaillierte Aufgabenschritte und Validierungskriterien erstellen, und dann generiert der Ausführende den entsprechenden Python-Code basierend auf Pandas, um die Aufgabe auszuführen. Wir visualisieren, wie der Agent autonom von der Argumentation zur Generierung umsetzbaren Codes übergeht. Schauen Sie sich das an VOLLSTÄNDIGE CODES hier.
def check(self, plan, outcome, 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"Consequence kind: {kind(outcome).__name__}n"
if not execution_error:
if isinstance(outcome, pd.DataFrame):
result_desc += f"Form: {outcome.form}nColumns: {checklist(outcome.columns)}nSample:n{outcome.head(3).to_string()}"
elif isinstance(outcome, (int,float,str)):
result_desc += f"Worth: {outcome}"
else:
result_desc += f"Worth: {str(outcome)(:200)}"
criteria_text="n".be a 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 Consequence:n{result_desc}nnEvaluate if outcome meets standards. Return JSON with handed (true/false), points, and suggestions."
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":("Couldn't parse check outcome"),"suggestions":("Overview manually")}
print(f"n✓ Check Outcomes:n Standing: {'✅ PASSED' if test_json.get('handed') else '❌ FAILED'}")
if test_json.get('points'):
print(" Points:")
for problem in test_json('points')(:3):
print(f" • {problem}")
if test_json.get('suggestions'):
print(" Suggestions:")
for rec in test_json('suggestions')(:3):
print(f" • {rec}")
return test_json
def run(self, process, df=None, data_info=None):
print("n🤖 SELF-VERIFYING DATA-OPS AGENT (Native HF Mannequin)")
print(f"Activity: {process}n")
if data_info is None and df is just not None:
data_info = f"Form: {df.form}nColumns: {checklist(df.columns)}nSample:n{df.head(2).to_string()}"
plan = self.plan(process, data_info)
code = self.execute(plan, data_info)
outcome, error = None, None
strive:
local_vars = {'pd': pd, 'np': np, 'df': df}
exec(code, local_vars)
outcome = local_vars.get('outcome')
besides Exception as e:
error = str(e)
print(f"n⚠️ Execution Error: {error}")
test_result = self.check(plan, outcome, error)
return {'plan': plan,'code': code,'outcome': outcome,'check': test_result,'historical past': self.historical past}Wir konzentrieren uns auf die Check- und Verifizierungsphase unseres Workflows. Wir lassen den Agenten seine eigene Ausgabe anhand vordefinierter Validierungskriterien bewerten und das Ergebnis als strukturiertes JSON zusammenfassen. Anschließend integrieren wir alle drei Phasen, Planung, Ausführung und Check, in einer einzigen selbstverifizierenden Pipeline, die eine vollständige Automatisierung gewährleistet. Schauen Sie sich das an VOLLSTÄNDIGE CODES hier.
def demo_basic(agent):
print("n" + "#"*60)
print("# DEMO 1: Gross sales Information 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')})
process = "Calculate whole gross sales by product"
output = agent.run(process, df)
if output('outcome') is just not None:
print(f"n📊 Closing Consequence:n{output('outcome')}")
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)})
process = "Calculate common spend by age group: younger (underneath 35) and mature (35+)"
output = agent.run(process, df)
if output('outcome') is just not None:
print(f"n📊 Closing Consequence:n{output('outcome')}")
return output
if __name__ == "__main__":
print("🚀 Initializing Native LLM...")
print("Utilizing CPU mode for optimum compatibilityn")
strive:
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. Attempt a distinct mannequin")Wir haben zwei Demobeispiele erstellt, um die Fähigkeiten des Agenten anhand einfacher Verkaufs- und Kundendatensätze zu testen. Wir initialisieren das Modell, führen den Information-Ops-Workflow aus und beobachten den gesamten Zyklus von der Planung bis zur Validierung. Wir schließen das Tutorial mit einer Zusammenfassung der wichtigsten Vorteile ab und ermutigen zu weiteren Experimenten mit lokalen Modellen.
Zusammenfassend haben wir ein vollständig autonomes und selbstverifizierendes DataOps-System erstellt, das auf einem lokalen Hugging Face-Modell basiert. Wir erleben, wie jede Part – Planung, Ausführung und Exams – nahtlos zusammenwirkt, um zuverlässige Ergebnisse zu erzielen, ohne auf Cloud-APIs angewiesen zu sein. Dieser Workflow unterstreicht die Stärke lokaler LLMs wie Phi-2 für eine leichte Automatisierung und inspiriert uns, diese Architektur in Zukunft für fortschrittlichere Datenpipelines, Validierungsframeworks und Multi-Agenten-Datensysteme zu erweitern.
Schauen Sie sich das an VOLLSTÄNDIGE CODES hier. Schauen Sie sich gerne bei uns um GitHub-Seite für Tutorials, Codes und Notebooks. Sie können uns auch gerne weiter folgen Twitter und vergessen Sie nicht, bei uns mitzumachen 100.000+ ML SubReddit und Abonnieren Unser Publication. Warten! Bist du im Telegram? Jetzt können Sie uns auch per Telegram kontaktieren.
Asif Razzaq ist CEO von Marktechpost Media Inc.. Als visionärer Unternehmer und Ingenieur setzt sich Asif dafür ein, das Potenzial der künstlichen Intelligenz für das soziale Wohl zu nutzen. Sein jüngstes Unterfangen ist die Einführung einer Medienplattform für künstliche Intelligenz, Marktechpost, die sich durch eine ausführliche Berichterstattung über maschinelles Lernen und Deep-Studying-Nachrichten auszeichnet, die sowohl technisch fundiert als auch für ein breites Publikum leicht verständlich ist. Die Plattform verfügt über mehr als 2 Millionen monatliche Aufrufe, was ihre Beliebtheit beim Publikum verdeutlicht.