كيف تبني AI Agent مستقل: من الصفر إلى التنفيذ الذاتي دون تدخل بشري

# Model Cascading Implementation
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

class ModelCascade:
 def __init__(self):
 # Small model for fast tasks
 self.small_model = AutoModelForCausalLM.from_pretrained(
 "mistralai/Mistral-7B-v0.1",
 torch_dtype=torch.float16,
 device_map="auto"
 )
 self.small_tokenizer = AutoTokenizer.from_pretrained("mistralai/Mistral-7B-v0.1")
 
 # Large model for complex tasks
 self.large_model = AutoModelForCausalLM.from_pretrained(
 "meta-llama/Meta-Llama-3-70B",
 torch_dtype=torch.float16,
 device_map="auto"
 )
 self.large_tokenizer = AutoTokenizer.from_pretrained("meta-llama/Meta-Llama-3-70B")
 
 def decide_model(self, task_description: str) -> str:
 """Determine which model to use based on task complexity"""
 prompt = f"Analyze the following task and respond with either 'small' or 'large':\n{task_description}\nModel:"
 inputs = self.small_tokenizer(prompt, return_tensors="pt").to("cuda")
 output = self.small_model.generate(**inputs, max_new_tokens=1)
 decision = self.small_tokenizer.decode(output[0], skip_special_tokens=True).strip()
 return "large" if decision == "large" else "small"
 
 def process_task(self, task_description: str) -> str:
 model_type = self.decide_model(task_description)
 model = self.large_model if model_type == "large" else self.small_model
 tokenizer = self.large_tokenizer if model_type == "large" else self.small_tokenizer
 
 inputs = tokenizer(task_description, return_tensors="pt").to("cuda")
 output = model.generate(**inputs, max_new_tokens=500)
 return tokenizer.decode(output[0], skip_special_tokens=True)

# Usage
cascade = ModelCascade()
task = "Analyze this server log for potential security breaches: [log data]"
result = cascade.process_task(task)
print(result)

# Memory Compression System using Vector DB and Summarization
from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

class MemorySystem:
 def __init__(self, max_memories=100):
 self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
 self.index = faiss.IndexFlatL2(384) # Dimension of all-MiniLM-L6-v2
 self.max_memories = max_memories
 self.memory_buffer = []
 
 def add_memory(self, text: str, metadata: dict = None):
 """Add a new memory with optional metadata"""
 embedding = self.embedding_model.encode(text)
 self.memory_buffer.append({
 "text": text,
 "embedding": embedding,
 "metadata": metadata or {}
 })
 
 # Compress if we exceed max memories
 if len(self.memory_buffer) > self.max_memories:
 self._compress_memories()
 
 def _compress_memories(self):
 """Compress memories by summarizing similar ones"""
 if not self.memory_buffer:
 return
 
 # Convert to numpy array for FAISS
 embeddings = np.array([m["embedding"] for m in self.memory_buffer])
 self.index.add(embeddings)
 
 # Cluster similar memories
 n_clusters = max(1, len(self.memory_buffer) // 10)
 kmeans = faiss.Kmeans(384, n_clusters, niter=20, verbose=False)
 kmeans.train(embeddings)
 _, labels = kmeans.index.search(embeddings, 1)
 
 # Summarize each cluster
 new_memories = []
 for cluster_id in range(n_clusters):
 cluster_memories = [m for m, l in zip(self.memory_buffer, labels) if l == cluster_id]
 if not cluster_memories:
 continue
 
 # Combine texts and summarize
 combined_text = "\n".join([m["text"] for m in cluster_memories])
 summary = self._summarize_text(combined_text)
 
 # Keep metadata from most recent memory in cluster
 most_recent = max(cluster_memories, key=lambda x: x["metadata"].get("timestamp", 0))
 new_memories.append({
 "text": summary,
 "embedding": self.embedding_model.encode(summary),
 "metadata": most_recent["metadata"]
 })
 
 self.memory_buffer = new_memories
 self.index.reset()
 
 def _summarize_text(self, text: str) -> str:
 """Simple summarization using sentence embeddings"""
 sentences = text.split('\n')
 if len(sentences) <= 3:
 return text
 
 embeddings = self.embedding_model.encode(sentences)
 centroid = np.mean(embeddings, axis=0)
 
 # Find sentences closest to centroid (most representative)
 distances = np.linalg.norm(embeddings - centroid, axis=1)
 top_indices = np.argsort(distances)[:3]
 
 return "\n".join([sentences[i] for i in sorted(top_indices)])
 
 def search_memories(self, query: str, k: int = 3) -> list:
 """Search memories using semantic similarity"""
 query_embedding = self.embedding_model.encode(query)
 distances, indices = self.index.search(np.array([query_embedding]), k)
 return [self.memory_buffer[i] for i in indices[0]]

# Perception Throttling System for AI Agents
import time
import requests
from collections import deque
import numpy as np

class PerceptionSystem:
 def __init__(self, initial_rate=1.0, max_rate=10.0, min_rate=0.1):
 self.current_rate = initial_rate # Requests per second
 self.max_rate = max_rate
 self.min_rate = min_rate
 self.rate_history = deque(maxlen=100)
 self.last_request_time = 0
 self.success_history = deque(maxlen=20)
 
 def _adjust_rate(self, success: bool):
 """Adjust the request rate based on success/failure"""
 self.success_history.append(success)
 
 # Calculate success ratio over last 20 requests
 if len(self.success_history) >= 5:
 success_ratio = sum(self.success_history) / len(self.success_history)
 
 # Multiplicative increase, additive decrease
 if success_ratio > 0.8:
 self.current_rate = min(self.current_rate * 1.2, self.max_rate)
 elif success_ratio < 0.5:
 self.current_rate = max(self.current_rate * 0.8, self.min_rate)
 
 def _get_delay(self) -> float:
 """Calculate delay based on current rate"""
 now = time.time()
 time_since_last = now - self.last_request_time
 target_delay = 1.0 / self.current_rate
 delay = max(0, target_delay - time_since_last)
 self.last_request_time = now + delay
 return delay
 
 def fetch_data(self, url: str, params: dict = None) -> dict:
 """Fetch data with adaptive rate limiting"""
 delay = self._get_delay()
 if delay > 0:
 time.sleep(delay)
 
 try:
 resp requests.get(url, params=params, timeout=10)
 response.raise_for_status()
 self._adjust_rate(True)
 return response.json()
 except (requests.RequestException, ValueError) as e:
 self._adjust_rate(False)
 raise Exception(f"Failed to fetch data: {str(e)}")
 
 def get_current_stats(self) -> dict:
 """Get current perception system stats"""
 return {
 "current_rate": round(self.current_rate, 2),
 "success_ratio": round(sum(self.success_history)/len(self.success_history), 2) 
 if self.success_history else 1.0
 }

# Example usage
perception = PerceptionSystem(initial_rate=2.0)
# This will automatically adjust the rate based on success/failure
for _ in range(50):
 try:
 data = perception.fetch_data("https://api.example.com/stock-prices", {"symbol": "AAPL"})
 print(f"Fetched data at rate: {perception.get_current_stats()['current_rate']} req/s")
 except Exception as e:
 print(f"Error: {e}")

# Decision Sandbox System for AI Agents
from typing import Callable, Dict, Any
import random
import time

class DecisionSandbox:
 def __init__(self, real_executor: Callable, simulation_executor: Callable):
 self.real_executor = real_executor # Function to execute decision in real world
 self.simulati simulation_executor # Function to simulate decision
 self.decision_history = []
 self.confidence_threshold = 0.85
 
 def _evaluate_decision(self, decision: Dict[str, Any], simulation_result: Dict[str, Any]) -> float:
 """Evaluate decision quality based on simulation results"""
 # Custom evaluation logic based on your use case
 # This is a simplified example
 score = 0.0
 
 # Example: Check if simulation was successful
 if simulation_result.get("success", False):
 score += 0.4
 
 # Example: Check if resources were used efficiently
 if simulation_result.get("resource_usage", 1.0) < 0.7:
 score += 0.3
 
 # Example: Check if risk was minimized
 if simulation_result.get("risk_level", 1.0) < 0.5:
 score += 0.3
 
 return min(max(score, 0.0), 1.0)
 
 def _should_execute_real(self, confidence: float) -> bool:
 """Determine if decision should be executed in real world"""
 # Gradually increase confidence threshold as agent learns
 dynamic_threshold = self.confidence_threshold * (1 - 0.1 * len(self.decision_history))
 return confidence >= dynamic_threshold
 
 def make_decision(self, decision: Dict[str, Any]) -> Dict[str, Any]:
 """Make a decision with safety checks"""
 start_time = time.time()
 
 # Step 1: Simulate the decision
 try:
 simulation_result = self.simulation_executor(decision)
 confidence = self._evaluate_decision(decision, simulation_result)
 except Exception as e:
 print(f"Simulation failed: {str(e)}")
 return {"status": "failed", "error": str(e)}
 
 # Step 2: Decide whether to execute in real world
 if self._should_execute_real(confidence):
 try:
 real_result = self.real_executor(decision)
 result = {
 "status": "executed",
 "confidence": confidence,
 "simulation_result": simulation_result,
 "real_result": real_result,
 "execution_time": time.time() - start_time
 }
 except Exception as e:
 result = {
 "status": "real_execution_failed",
 "confidence": confidence,
 "simulation_result": simulation_result,
 "error": str(e)
 }
 else:
 result = {
 "status": "simulated_only",
 "confidence": confidence,
 "simulation_result": simulation_result,
 "execution_time": time.time() - start_time
 }
 
 # Step 3: Record decision for future learning
 self.decision_history.append({
 "decision": decision,
 "result": result,
 "timestamp": time.time()
 })
 
 return result

# Example usage
def real_executor(decision: Dict[str, Any]) -> Dict[str, Any]:
 """Execute decision in real world (simplified example)"""
 print(f"Executing in real world: {decision}")
 # Simulate real execution
 time.sleep(0.5)
 return {"success": True, "details": "Executed successfully"}

def simulation_executor(decision: Dict[str, Any]) -> Dict[str, Any]:
 """Simulate decision execution (simplified example)"""
 print(f"Simulating decision: {decision}")
 # Simulate different outcomes based on decision
 if decision.get("type") == "high_risk":
 return {
 "success": random.random() > 0.3,
 "resource_usage": random.random(),
 "risk_level": random.random()
 }
 else:
 return {
 "success": random.random() > 0.1,
 "resource_usage": random.random() * 0.5,
 "risk_level": random.random() * 0.3
 }

sandbox = DecisionSandbox(real_executor, simulation_executor)

decision = {
 "type": "allocate_resources",
 "amount": 1000,
 "project": "AI Development"
}

result = sandbox.make_decision(decision)
print(f"Decision result: {result['status']}, Confidence: {result['confidence']:.2f}")

# Minimum Viable AI Agent for Pull Request Review
from github import Github
from transformers import pipeline
import os

class PRReviewAgent:
 def __init__(self, github_token: str, repo_name: str):
 self.github = Github(github_token)
 self.repo = self.github.get_repo(repo_name)
 self.reviewer = pipeline(
 "text-generation",
 model="HuggingFaceH4/zephyr-7b-beta",
 device_map="auto"
 )
 
 def _generate_review(self, pr_diff: str) -> str:
 """Generate code review using LLM"""
 prompt = f"""You are an expert code reviewer. Analyze the following Git diff and provide:
 1. Summary of changes
 2. Potential issues
 3. Suggestions for improvement
 
 Git Diff:\n{pr_diff[:3000]} # Limit to first 3000 chars
 
 Review:"""
 
 resp self.reviewer(prompt, max_new_tokens=500)
 return response[0]['generated_text']
 
 def review_open_prs(self):
 """Review all open pull requests"""
 for pr in self.repo.get_pulls(state='open'):
 print(f"Reviewing PR #{pr.number}: {pr.title}")
 
 # Get PR diff
 diff = pr.get_files()[0].patch if pr.get_files() else "No changes"
 
 # Generate review
 review = self._generate_review(diff)
 
 # Post review as comment
 pr.create_issue_comment(f" AI Code Review:\n\n{review}")
 print(f"Posted review for PR #{pr.number}")

# Usage
agent = PRReviewAgent(
 github_token=os.getenv("GITHUB_TOKEN"),
 repo_name="your-username/your-repo"
)
agent.review_open_prs()