Month: November 2025

Amazon Bedrock AgentCore is an agentic platform for building, deploying, and operating effective agents securely at scale. Amazon Bedrock AgentCore Runtime is a fully managed service of Bedrock AgentCore, which provides low latency serverless environments to deploy agents and tools. It provides session isolation, supports multiple agent frameworks including popular open-source frameworks, and handles multimodal workloads and long-running agents.
AgentCore Runtime supports container based deployments where the container definition is provided in a Dockerfile, and the agent is built as a container image. Customers who have container build and deploy pipelines benefit from this method, where agent deployment can be integrated into existing pipelines. 
Today, AgentCore Runtime has launched a second method to deploy agents – direct code deployment (for Python). Agent code and its dependencies can be packaged as a zip archive, alleviating the need for Docker definition and ECR dependencies. This makes it straightforward for developers to prototype and iterate faster. This method is a good fit for customers who prefer not to worry about Docker expertise and container infrastructure when deploying agents.
In this post, we’ll demonstrate how to use direct code deployment (for Python).
Introducing AgentCore Runtime direct code deployment
With the container deployment method, developers create a Dockerfile, build ARM-compatible containers, manage ECR repositories, and upload containers for code changes. This works well where container DevOps pipelines have already been established to automate deployments. 
However, customers looking for fully managed deployments can benefit from direct code deployment, which can significantly improve developer time and productivity. Direct code deployment provides a secure and scalable path forward for rapid prototyping agent capabilities to deploying production workloads at scale.
We’ll discuss the strengths of each deployment option to help you choose the right approach for your use case. 

With direct code deployment, developers create a zip archive of code and dependencies, upload to Amazon S3, and configure the bucket in the agent configuration. When using the AgentCore starter toolkit, the toolkit handles dependency detection, packaging, and upload which provides a much-simplified developer experience. Direct code deployment is also supported using the API.
Let’s compare the deployment steps at a high level between the two methods:
Container-based deployment
The container-based deployment method involves the following steps:

Create a Dockerfile
Build ARM-compatible container
Create ECR repository
Upload to ECR
Deploy to AgentCore Runtime

Direct code deployment
The direct code deployment method involves the following steps:

Package your code and dependencies into a zip archive
Upload it to S3
Configure the bucket in agent configuration
Deploy to AgentCore Runtime

How to use direct code deployment
Let’s illustrate how direct code deployment works with an agent created with Strands Agents SDK and using the AgentCore starter-toolkit to deploy the agent.
Prerequisites
Before you begin, make sure you have the following:

Any of the versions of Python 3.10 to 3.13
Your preferred package manager installed. For example, we use uv package manager.
AWS account for creating and deploying agents
Amazon Bedrock model access to Anthropic Claude Sonnet 4.0

Step 1: Initialize your project
Set up a new Python project using the uv package manager, then navigate into the project directory:

uv init <project> –python 3.13
cd <project>

Step 2: Add the dependencies for the project
Install the required Bedrock AgentCore libraries and development tools for your project. In this example, dependencies are added using .toml file, alternatively they can be specified in requirements.txt file:

uv add bedrock-agentcore strands-agents strands-agents-tools
uv add –dev bedrock-agentcore-starter-toolkit
source .venv/bin/activate

Step 3: Create an agent.py file
Create the main agent implementation file that defines your AI agent’s behavior:

from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent, tool
from strands_tools import calculator
from strands.models import BedrockModel
import logging

app = BedrockAgentCoreApp(debug=True)

# Logging setup
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Create a custom tool
@tool
def weather():
“”” Get weather “””
return “sunny”

model_id = “us.anthropic.claude-sonnet-4-20250514-v1:0″
model = BedrockModel(
model_id=model_id,
)

agent = Agent(
model=model,
tools=[calculator, weather],
system_prompt=”You’re a helpful assistant. You can do simple math calculation, and tell the weather.”
)

@app.entrypoint
def invoke(payload):
“””Your AI agent function”””
user_input = payload.get(“prompt”, “Hello! How can I help you today?”)
logger.info(“n User input: %s”, user_input)
response = agent(user_input)
logger.info(“n Agent result: %s “, response.message)
return response.message[‘content’][0][‘text’]

if __name__ == “__main__”:
app.run()

Step 4: Deploy to AgentCore Runtime
Configure and deploy your agent to the AgentCore Runtime environment:

agentcore configure –entrypoint agent.py –name <agent-name>

This will launch an interactive session where you configure the S3 bucket to upload the zip deployment package to and choose a deployment configuration type (as shown in the following configuration). To opt for direct code deployment, choose option 1 – Code Zip.
Deployment Configuration
Select deployment type:

Code Zip (recommended) – Simple, serverless, no Docker required
Container – For custom runtimes or complex dependencies

agentcore launch

This command creates a zip deployment package, uploads it to the specified S3 bucket, and launches the agent in the AgentCore Runtime environment, making it ready to receive and process requests.
To test the solution, let’s prompt the agent to see how the weather is:

agentcore invoke ‘{“prompt”:”How is the weather today?”}’

The first deployment takes approximately 30 seconds to complete, but subsequent updates to the agent benefit from the streamlined direct code deployment process and should take less than half the time, supporting faster iteration cycles during development.
When to choose direct code instead of container-based deployment
Let’s look at some of the dimensions and see how the direct code and container-based deployment options are different. This will help you choose the option that’s right for you:

Deployment process: Direct code deploys agents as zip files with no Docker, ECR, or CodeBuild required. Container-based deployment uses Docker and ECR with full Dockerfile control.
Deployment time: Although there is not much difference during first deployment of an agent, subsequent updates to the agent are significantly faster with direct code deployment (from an average of 30 seconds for containers to about 10 seconds for direct code deployment).
Artifact storage: Direct code stores ZIP packages in an S3 bucket. Container-based deployment stores Docker images in Amazon ECR. Direct code deployment incurs storage costs at standard S3 storage rates (starting February 27th  2026) as artifacts are stored in the service account. Container-based deployment incurs Amazon ECR charges in your account.
Customization: Direct code deployment supports custom dependencies through ZIP-based packaging, while container based depends on a Dockerfile.
Package size: Direct code deployment limits the package size to 250MB whereas container-based packages can be up to 2GB in size.
Language Support: Direct code currently supports Python 3.10, 3.11, 3.12, and 3.13. Container-based deployment supports many languages and runtimes.

Our general guidance is:
Container-based deployment is the right choice when your package exceeds 250MB, you have existing container CI/CD pipelines, or you need highly specialized dependencies and custom packaging requirements. Choose containers if you require multi-language support, custom system dependencies or direct control over artifact storage and versioning in your account.
Direct code deployment is the right choice when your package is under 250MB, you use Python 3.10-3.13 with common frameworks like LangGraph, Strands, or CrewAI, and you need rapid prototyping with fast iteration cycles. Choose direct code if your build process is straightforward without complex dependencies, and you want to remove the Docker/ECR/CodeBuild setup.
A hybrid approach works well for many teams, use direct code for rapid prototyping and experimentation where fast iteration and simple setup accelerate development, then graduate to containers for production when package size, multi-language requirements, or specialized build processes demand it.
Conclusion
Amazon Bedrock AgentCore direct code deployment makes iterative agent development cycles even faster, while still benefiting from enterprise security and scale of deployments. Developers can now rapidly prototype and iterate by deploying their code directly, without having to create a container. To get started with Amazon Bedrock AgentCore direct code deployment, visit the AWS documentation.

About the authors
Chaitra Mathur is as a GenAI Specialist Solutions Architect at AWS. She works with customers across industries in building scalable generative AI platforms and operationalizing them. Throughout her career, she has shared her expertise at numerous conferences and has authored several blogs in the Machine Learning and Generative AI domains.
Qingwei Li is a Machine Learning Specialist at Amazon Web Services. He received his Ph.D. in Operations Research after he broke his advisor’s research grant account and failed to deliver the Nobel Prize he promised. Currently he helps customers in the financial service and insurance industry build machine learning solutions on AWS. In his spare time, he likes reading and teaching.
Kosti Vasilakakis is a Principal PM at AWS on the Agentic AI team, where he has led the design and development of several Bedrock AgentCore services from the ground up, including Runtime, Browser, Code Interpreter, and Identity. He previously worked on Amazon SageMaker since its early days, launching AI/ML capabilities now used by thousands of companies worldwide. Earlier in his career, Kosti was a data scientist. Outside of work, he builds personal productivity automations, plays tennis, and enjoys life with his wife and kids.

How can AI teams run Tinker style reinforcement learning on large language models using their own infrastructure with a single unified engine? Anyscale and NovaSky (UC Berkeley) Team releases SkyRL tx v0.1.0 that gives developers a way to run a Tinker compatible training and inference engine directly on their own hardware, while keeping the same minimal API that Tinker exposes in the managed service.

The research team describes SkyRL tx as a unified training and inference engine that implements the Tinker API and allows people to run a Tinker like service on their own infrastructure. This v0.1.0 version is the first of its series that supports reinforcement learning end to end, and it also makes sampling significantly faster.

Tinker API in brief

Tinker from Thinking Machines is a training API built around four core functions. forward_backward performs a forward pass and a backward pass and accumulates gradients. optim_step updates model weights based on those gradients. sample generates tokens for interaction, evaluation or RL actions. save_state writes checkpoints for resuming training.

Instead of a full task specific fine tuning abstraction, Tinker exposes these low level primitives so that users can implement their own supervised or reinforcement learning loops in regular Python code, while the service handles GPU scheduling and distributed execution.

SkyRL tx targets this exact API and implements an open backend that users can deploy locally. It keeps the Tinker programming model, while removing the need to rely only on the hosted environment.

Where SkyRL tx fits inside SkyRL

SkyRL is a full stack reinforcement learning library for large language models that includes skyrl-agent for long horizon agents, skyrl-train for training, and skyrl-gym for tool use environments such as math, coding, search and SQL.

Within this stack, skyrl-tx is marked as an experimental cross platform library that exposes a local Tinker like REST API for model post training. SkyRL tx therefore becomes the system layer that connects RL logic, environments and training code to concrete GPU resources through the Tinker interface.

Architecture, inference engine that also trains

The SkyRL tx architecture is described as an inference engine that also supports backward passes. It has four main components:

REST API server that processes incoming requests from different users.

Database that tracks metadata about models, checkpoints, requests and futures, and also acts as a job queue. The current implementation uses SQLite behind an interface that also supports other SQL databases such as Postgres.

Engine that schedules and batches requests across users. Each engine instance serves a single base model and can attach many LoRA adapters.

Worker that executes forward and backward passes and holds model definitions and optimizer states. Multiple workers would be enabling more advanced multi node sharding in upcoming versions

What v0.1.0 adds?

The v0.1.0 release focuses on reinforcement learning support and performance improvements. The official release highlights several concrete changes:

Sampling is now much faster, since it is jitted and properly batched and sharded in the engine.

Different sampling parameters per request, per request seeds and stop tokens are now supported, which is useful when many experiments share a base model.

After several fixes, the RL loop now runs properly through the engine.

Gradient checkpointing support and micro batching for sampling are implemented.

Postgres is now supported as a database backend, next to SQLite.

Running RL end to end on 8 H100 GPUs

The official release contains a specific code recipe for running reinforcement learning end to end on a cluster with 8 H100 GPUs.

First, users clone the SkyRL repository and in the skyrl-tx folder start the engine with:

Copy CodeCopiedUse a different Browseruv run –extra gpu –extra tinker -m tx.tinker.api
–base-model Qwen/Qwen3-4B
–max-lora-adapters 3
–max-lora-rank 1
–tensor-parallel-size 8
–train-micro-batch-size 8 > out.log

Then they clone the Tinker Cookbook from the Thinking Machines team and in the tinker_cookbook/recipes folder run:

Copy CodeCopiedUse a different Browserexport TINKER_API_KEY=dummy
export WANDB_API_KEY=<your key>
uv run –with wandb –with tinker rl_loop.py
base_url=http://localhost:8000
model_name=”Qwen/Qwen3-4B”
lora_rank=1
max_length=1024
save_every=100

This produces a reward curve that confirms the RL loop runs correctly through the local SkyRL tx backend.

Key Takeaways

SkyRL tx v0.1.0 implements a local, Tinker compatible engine that unifies training and inference for LLM post training.

The system exposes Tinker primitives, forward_backward, optim_step, sample and save_state over REST, while handling batching, LoRA adapters and device placement internally.

Architecture is split into API server, SQL database, scheduling engine and workers that execute forward and backward passes for a single base model with multiple LoRA adapters.

v0.1.0 adds end to end reinforcement learning support, faster jitted and sharded sampling, per request sampling parameters, gradient checkpointing, micro batching and Postgres support.

Editorial Comments

SkyRL tx v0.1.0 is a practical step for dev teams that want Tinker style reinforcement learning on their own clusters with a consistent Tinker API surface. The design that treats the system as an inference engine that also runs backward passes is clean and reduces stack divergence. Support for LoRA, gradient checkpointing, micro batching and Postgres is a concrete systems upgrade. Overall, this release turns Tinker compatibility into an actionable local RL backend for LLM

Check out the Repo and Official Release. Feel free to check out our GitHub Page for Tutorials, Codes and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post Anyscale and NovaSky Team Releases SkyRL tx v0.1.0: Bringing Tinker Compatible Reinforcement Learning RL Engine To Local GPU Clusters appeared first on MarkTechPost.

In this tutorial, we explore how to build an intelligent agent that remembers, learns, and adapts to us over time. We implement a Persistent Memory & Personalisation system using simple, rule-based logic to simulate how modern Agentic AI frameworks store and recall contextual information. As we progress, we see how the agent’s responses evolve with experience, how memory decay helps prevent overload, and how personalisation improves performance. We aim to understand, step by step, how persistence transforms a static chatbot into a context-aware, evolving digital companion. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserimport math, time, random
from typing import List

class MemoryItem:
def __init__(self, kind:str, content:str, score:float=1.0):
self.kind = kind
self.content = content
self.score = score
self.t = time.time()

class MemoryStore:
def __init__(self, decay_half_life=1800):
self.items: List[MemoryItem] = []
self.decay_half_life = decay_half_life

def _decay_factor(self, item:MemoryItem):
dt = time.time() – item.t
return 0.5 ** (dt / self.decay_half_life)

We established the foundation for our agent’s long-term memory. We define the MemoryItem class to hold each piece of information and build a MemoryStore with an exponential decay mechanism. We begin laying the foundation for storing and aging information just like a human’s memory. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browser def add(self, kind:str, content:str, score:float=1.0):
self.items.append(MemoryItem(kind, content, score))

def search(self, query:str, topk=3):
scored = []
for it in self.items:
decay = self._decay_factor(it)
sim = len(set(query.lower().split()) & set(it.content.lower().split()))
final = (it.score * decay) + sim
scored.append((final, it))
scored.sort(key=lambda x: x[0], reverse=True)
return [it for _, it in scored[:topk] if _ > 0]

def cleanup(self, min_score=0.1):
new = []
for it in self.items:
if it.score * self._decay_factor(it) > min_score:
new.append(it)
self.items = new

We expand the memory system by adding methods to insert, search, and clean old memories. We implement a simple similarity function and a decay-based cleanup routine, enabling the agent to remember relevant facts while automatically forgetting weak or outdated ones. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserclass Agent:
def __init__(self, memory:MemoryStore, name=”PersonalAgent”):
self.memory = memory
self.name = name

def _llm_sim(self, prompt:str, context:List[str]):
base = “OK. ”
if any(“prefers short” in c for c in context):
base = “”
reply = base + f”I considered {len(context)} past notes. ”
if “summarize” in prompt.lower():
return reply + “Summary: ” + ” | “.join(context[:2])
if “recommend” in prompt.lower():
if any(“cybersecurity” in c for c in context):
return reply + “Recommended: write more cybersecurity articles.”
if any(“rag” in c for c in context):
return reply + “Recommended: build an agentic RAG demo next.”
return reply + “Recommended: continue with your last topic.”
return reply + “Here’s my response to: ” + prompt

def perceive(self, user_input:str):
ui = user_input.lower()
if “i like” in ui or “i prefer” in ui:
self.memory.add(“preference”, user_input, 1.5)
if “topic:” in ui:
self.memory.add(“topic”, user_input, 1.2)
if “project” in ui:
self.memory.add(“project”, user_input, 1.0)
def act(self, user_input:str):
mems = self.memory.search(user_input, topk=4)
ctx = [m.content for m in mems]
answer = self._llm_sim(user_input, ctx)
self.memory.add(“dialog”, f”user said: {user_input}”, 0.6)
self.memory.cleanup()
return answer, ctx

We design an intelligent agent that utilizes memory to inform its responses. We create a mock language model simulator that adapts replies based on stored preferences and topics. At the same time, the perception function enables the agent to dynamically capture new insights about the user. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserdef evaluate_personalisation(agent:Agent):
agent.memory.add(“preference”, “User likes cybersecurity articles”, 1.6)
q = “Recommend what to write next”
ans_personal, _ = agent.act(q)
empty_mem = MemoryStore()
cold_agent = Agent(empty_mem)
ans_cold, _ = cold_agent.act(q)
gain = len(ans_personal) – len(ans_cold)
return ans_personal, ans_cold, gain

Now we give our agent the ability to act and evaluate itself. We allow it to recall memories to shape contextual answers and add a small evaluation loop to compare personalised responses versus a memory-less baseline, quantifying how much the memory helps. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browsermem = MemoryStore(decay_half_life=60)
agent = Agent(mem)

print(“=== Demo: teaching the agent about yourself ===”)
inputs = [
“I prefer short answers.”,
“I like writing about RAG and agentic AI.”,
“Topic: cybersecurity, phishing, APTs.”,
“My current project is to build an agentic RAG Q&A system.”
]
for inp in inputs:
agent.perceive(inp)

print(“n=== Now ask the agent something ===”)
user_q = “Recommend what to write next in my blog”
ans, ctx = agent.act(user_q)
print(“USER:”, user_q)
print(“AGENT:”, ans)
print(“USED MEMORY:”, ctx)

print(“n=== Evaluate personalisation benefit ===”)
p, c, g = evaluate_personalisation(agent)
print(“With memory :”, p)
print(“Cold start :”, c)
print(“Personalisation gain (chars):”, g)

print(“n=== Current memory snapshot ===”)
for it in agent.memory.items:
print(f”- {it.kind} | {it.content[:60]}… | score~{round(it.score,2)}”)

Finally, we run the full demo to see our agent in action. We feed it user inputs, observe how it recommends personalised actions, and check its memory snapshot. We witness the emergence of adaptive behaviour, proof that persistent memory transforms a static script into a learning companion.

In conclusion, we demonstrate how adding memory and personalisation makes our agent more human-like, capable of remembering preferences, adapting plans, and forgetting outdated details naturally. We observe that even simple mechanisms such as decay and retrieval significantly improve the agent’s relevance and response quality. By the end, we realize that persistent memory is the foundation of next-generation Agentic AI, one that learns continuously, tailors experiences intelligently, and maintains context dynamically in a fully local, offline setup.

Check out the FULL CODES here. Feel free to check out our GitHub Page for Tutorials, Codes and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post How to Design a Persistent Memory and Personalized Agentic AI System with Decay and Self-Evaluation? appeared first on MarkTechPost.

In this tutorial, we develop a comprehensive benchmarking framework to evaluate various types of agentic AI systems on real-world enterprise software tasks. We design a suite of diverse challenges, from data transformation and API integration to workflow automation and performance optimization, and assess how various agents, including rule-based, LLM-powered, and hybrid ones, perform across these domains. By running structured benchmarks and visualizing key performance metrics, such as accuracy, execution time, and success rate, we gain a deeper understanding of each agent’s strengths and trade-offs in enterprise environments. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserimport json
import time
import random
from typing import Dict, List, Any, Callable
from dataclasses import dataclass, asdict
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns

@dataclass
class Task:
id: str
name: str
description: str
category: str
complexity: int
expected_output: Any

@dataclass
class BenchmarkResult:
task_id: str
agent_name: str
success: bool
execution_time: float
accuracy: float
error_message: str = “”

class EnterpriseTaskSuite:
def __init__(self):
self.tasks = self._create_tasks()

def _create_tasks(self) -> List[Task]:
return [
Task(“data_transform”, “CSV Data Transformation”,
“Transform customer data by aggregating sales”, “data_processing”, 3,
{“total_sales”: 15000, “avg_order”: 750}),
Task(“api_integration”, “REST API Integration”,
“Parse API response and extract key metrics”, “integration”, 2,
{“status”: “success”, “active_users”: 1250}),
Task(“workflow_automation”, “Multi-Step Workflow”,
“Execute data validation -> processing -> reporting”, “automation”, 4,
{“validated”: True, “processed”: 100, “report_generated”: True}),
Task(“error_handling”, “Error Recovery”,
“Handle malformed data gracefully”, “reliability”, 3,
{“errors_caught”: 5, “recovery_success”: True}),
Task(“optimization”, “Query Optimization”,
“Optimize database query performance”, “performance”, 5,
{“execution_time_ms”: 45, “rows_scanned”: 1000}),
Task(“data_validation”, “Schema Validation”,
“Validate data against business rules”, “validation”, 2,
{“valid_records”: 95, “invalid_records”: 5}),
Task(“reporting”, “Executive Dashboard”,
“Generate KPI summary report”, “analytics”, 3,
{“revenue”: 125000, “growth”: 0.15, “customer_count”: 450}),
Task(“integration_test”, “System Integration”,
“Test end-to-end integration flow”, “testing”, 4,
{“all_systems_connected”: True, “latency_ms”: 120}),
]

def get_task(self, task_id: str) -> Task:
return next((t for t in self.tasks if t.id == task_id), None)

We define the core data structures for our benchmarking system. We create the Task and BenchmarkResult data classes and initialize the EnterpriseTaskSuite, which holds multiple enterprise-relevant tasks such as data transformation, reporting, and integration. We laid the foundation for consistently evaluating different types of agents across these tasks. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserclass BaseAgent:
def __init__(self, name: str):
self.name = name

def execute(self, task: Task) -> Dict[str, Any]:
raise NotImplementedError

class RuleBasedAgent(BaseAgent):
def execute(self, task: Task) -> Dict[str, Any]:
time.sleep(random.uniform(0.1, 0.3))
if task.category == “data_processing”:
return {“total_sales”: 15000 + random.randint(-500, 500),
“avg_order”: 750 + random.randint(-50, 50)}
elif task.category == “integration”:
return {“status”: “success”, “active_users”: 1250}
elif task.category == “automation”:
return {“validated”: True, “processed”: 98, “report_generated”: True}
else:
return task.expected_output

We introduce the base agent structure and implement the RuleBasedAgent, which mimics traditional automation logic using predefined rules. We simulate how such agents execute tasks deterministically while maintaining speed and reliability, giving us a baseline for comparison with more advanced agents. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserclass LLMAgent(BaseAgent):
def execute(self, task: Task) -> Dict[str, Any]:
time.sleep(random.uniform(0.2, 0.5))
accuracy_boost = 0.95 if task.complexity >= 4 else 0.90
result = {}
for key, value in task.expected_output.items():
if isinstance(value, (int, float)):
variation = value * (1 – accuracy_boost)
result[key] = value + random.uniform(-variation, variation)
else:
result[key] = value
return result

class HybridAgent(BaseAgent):
def execute(self, task: Task) -> Dict[str, Any]:
time.sleep(random.uniform(0.15, 0.35))
if task.complexity <= 2:
return task.expected_output
else:
result = {}
for key, value in task.expected_output.items():
if isinstance(value, (int, float)):
variation = value * 0.03
result[key] = value + random.uniform(-variation, variation)
else:
result[key] = value
return result

We develop two intelligent agent types, the LLMAgent, representing reasoning-based AI systems, and the HybridAgent, which combines rule-based precision with LLM adaptability. We design these agents to show how learning-based methods improve task accuracy, especially for complex enterprise workflows. Check out the Full Codes here.

Copy CodeCopiedUse a different Browserclass BenchmarkEngine:
def __init__(self, task_suite: EnterpriseTaskSuite):
self.task_suite = task_suite
self.results: List[BenchmarkResult] = []

def run_benchmark(self, agent: BaseAgent, iterations: int = 3):
print(f”n{‘=’*60}”)
print(f”Benchmarking Agent: {agent.name}”)
print(f”{‘=’*60}”)
for task in self.task_suite.tasks:
print(f”nTask: {task.name} (Complexity: {task.complexity}/5)”)
for i in range(iterations):
result = self._execute_task(agent, task, i+1)
self.results.append(result)
status = “✓ PASS” if result.success else “✗ FAIL”
print(f” Run {i+1}: {status} | Time: {result.execution_time:.3f}s | Accuracy: {result.accuracy:.2%}”)

Here, we build the core of our benchmarking engine, which manages agent evaluation across the defined task suite. We implement methods to run each agent multiple times per task, log results, and measure key parameters like execution time and accuracy. This creates a systematic and repeatable benchmarking loop. Check out the Full Codes here.

Copy CodeCopiedUse a different Browser def _execute_task(self, agent: BaseAgent, task: Task, run_num: int) -> BenchmarkResult:
start_time = time.time()
try:
output = agent.execute(task)
execution_time = time.time() – start_time
accuracy = self._calculate_accuracy(output, task.expected_output)
success = accuracy >= 0.85
return BenchmarkResult(task_id=task.id, agent_name=agent.name, success=success,
execution_time=execution_time, accuracy=accuracy)
except Exception as e:
execution_time = time.time() – start_time
return BenchmarkResult(task_id=task.id, agent_name=agent.name, success=False,
execution_time=execution_time, accuracy=0.0, error_message=str(e))

def _calculate_accuracy(self, output: Dict, expected: Dict) -> float:
if not output:
return 0.0
scores = []
for key, expected_val in expected.items():
if key not in output:
scores.append(0.0)
continue
actual_val = output[key]
if isinstance(expected_val, bool):
scores.append(1.0 if actual_val == expected_val else 0.0)
elif isinstance(expected_val, (int, float)):
diff = abs(actual_val – expected_val)
tolerance = abs(expected_val * 0.1)
score = max(0, 1 – (diff / (tolerance + 1e-9)))
scores.append(score)
else:
scores.append(1.0 if actual_val == expected_val else 0.0)
return np.mean(scores) if scores else 0.0

We define the task execution logic and the accuracy computation. We measure each agent’s performance by comparing their outputs against expected results using a scoring mechanism. This step ensures our benchmarking process is quantitative and fair, providing insights into how closely agents align with business expectations. Check out the Full Codes here.

Copy CodeCopiedUse a different Browser def generate_report(self):
df = pd.DataFrame([asdict(r) for r in self.results])
print(f”n{‘=’*60}”)
print(“BENCHMARK REPORT”)
print(f”{‘=’*60}n”)
for agent_name in df[‘agent_name’].unique():
agent_df = df[df[‘agent_name’] == agent_name]
print(f”{agent_name}:”)
print(f” Success Rate: {agent_df[‘success’].mean():.1%}”)
print(f” Avg Execution Time: {agent_df[‘execution_time’].mean():.3f}s”)
print(f” Avg Accuracy: {agent_df[‘accuracy’].mean():.2%}n”)
return df

def visualize_results(self, df: pd.DataFrame):
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
fig.suptitle(‘Enterprise Agent Benchmarking Results’, fontsize=16, fontweight=’bold’)
success_rate = df.groupby(‘agent_name’)[‘success’].mean()
axes[0, 0].bar(success_rate.index, success_rate.values, color=[‘#3498db’, ‘#e74c3c’, ‘#2ecc71’])
axes[0, 0].set_title(‘Success Rate by Agent’, fontweight=’bold’)
axes[0, 0].set_ylabel(‘Success Rate’)
axes[0, 0].set_ylim(0, 1.1)
for i, v in enumerate(success_rate.values):
axes[0, 0].text(i, v + 0.02, f'{v:.1%}’, ha=’center’, fontweight=’bold’)
time_data = df.groupby(‘agent_name’)[‘execution_time’].mean()
axes[0, 1].bar(time_data.index, time_data.values, color=[‘#3498db’, ‘#e74c3c’, ‘#2ecc71’])
axes[0, 1].set_title(‘Average Execution Time’, fontweight=’bold’)
axes[0, 1].set_ylabel(‘Time (seconds)’)
for i, v in enumerate(time_data.values):
axes[0, 1].text(i, v + 0.01, f'{v:.3f}s’, ha=’center’, fontweight=’bold’)
df.boxplot(column=’accuracy’, by=’agent_name’, ax=axes[1, 0])
axes[1, 0].set_title(‘Accuracy Distribution’, fontweight=’bold’)
axes[1, 0].set_xlabel(‘Agent’)
axes[1, 0].set_ylabel(‘Accuracy’)
plt.sca(axes[1, 0])
plt.xticks(rotation=15)
task_complexity = {t.id: t.complexity for t in self.task_suite.tasks}
df[‘complexity’] = df[‘task_id’].map(task_complexity)
complexity_perf = df.groupby([‘agent_name’, ‘complexity’])[‘accuracy’].mean().unstack()
complexity_perf.plot(kind=’line’, ax=axes[1, 1], marker=’o’, linewidth=2)
axes[1, 1].set_title(‘Accuracy by Task Complexity’, fontweight=’bold’)
axes[1, 1].set_xlabel(‘Task Complexity’)
axes[1, 1].set_ylabel(‘Accuracy’)
axes[1, 1].legend(title=’Agent’, loc=’best’)
axes[1, 1].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

if __name__ == “__main__”:
print(“Enterprise Software Benchmarking for Agentic Agents”)
print(“=”*60)
task_suite = EnterpriseTaskSuite()
benchmark = BenchmarkEngine(task_suite)
agents = [RuleBasedAgent(“Rule-Based Agent”), LLMAgent(“LLM Agent”), HybridAgent(“Hybrid Agent”)]
for agent in agents:
benchmark.run_benchmark(agent, iterations=3)
results_df = benchmark.generate_report()
benchmark.visualize_results(results_df)
results_df.to_csv(‘agent_benchmark_results.csv’, index=False)
print(“nResults exported to: agent_benchmark_results.csv”)

We generate detailed reports and create visual analytics for performance comparison. We analyze metrics such as success rate, execution time, and accuracy across agents and task complexities. Finally, we export the results to CSV file, completing a full enterprise-grade evaluation workflow.

In conclusion, we implemented a robust, extensible benchmarking system that enables us to measure and compare the efficiency, adaptability, and accuracy of multiple agentic AI approaches. We observed how different architectures excel at different levels of task complexity and how visual analytics highlight performance trends. This process enables us to evaluate existing agents and provides a strong foundation for next-generation enterprise AI agents, optimized for reliability and intelligence.

Check out the Full Codes here. Feel free to check out our GitHub Page for Tutorials, Codes and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post A Coding Implementation of a Comprehensive Enterprise AI Benchmarking Framework to Evaluate Rule-Based LLM, and Hybrid Agentic AI Systems Across Real-World Tasks appeared first on MarkTechPost.

Most agent frameworks still run a predefined Reason, Act, Observe loop, so the agent can only use the tools that are injected in the prompt. This works for small tasks, but it fails when the toolset is large, when the task is long, and when the agent must change strategy in the middle of reasoning. The team from Renmin University of China and Xiaohongshu proposes DeepAgent as an end to end deep reasoning agent that keeps all of this inside one coherent reasoning process.

https://arxiv.org/pdf/2510.21618

Unified Reasoning With On Demand Tool Discovery

DeepAgent lets the model output four action types directly in text, internal thought, tool search, tool call, and memory fold. When the agent decides to search, it queries a dense index that contains tool descriptions from large registries, for example 16,000 plus RapidAPI tools and 3,912 ToolHop tools, then it receives only the top ranked tools back in context. This makes tool access dynamic, the model does not depend on a front loaded tool list, and it stays aligned with real environments where tools change.

Autonomous Memory Folding for Long Horizon Tasks

Long sequences of tool calls, web results, and code responses will overflow the context. DeepAgent solves this with an autonomous memory folding step. When the model emits the fold token, an auxiliary LLM compresses the full history into three memories, Episodic Memory that records task events, Working Memory that records the current sub goal and recent issues, and Tool Memory that records tool names, arguments, and outcomes. These memories are fed back as structured text, so the agent continues from a compact but information rich state.

ToolPO, Reinforcement Learning for Tool Use

Supervised traces do not teach robust tool use, because correct tool calls are only a few tokens inside a long generation. The research team introduce Tool Policy Optimization, ToolPO, to fix this. ToolPO runs rollouts on LLM simulated APIs, so training is stable and cheap, then it attributes reward to the exact tool call tokens, this is tool call advantage attribution, and it trains with a clipped PPO style objective. This is how the agent learns not only to call tools, but also to decide when to search and when to fold memory.

https://arxiv.org/pdf/2510.21618

Benchmarks, Labeled Tools vs Open Set Tools

The research team evaluates on 5 general tool use benchmarks, ToolBench, API Bank, TMDB, Spotify, ToolHop, and on 4 downstream tasks, ALFWorld, WebShop, GAIA, HLE. In the labeled tool setting, where every method is given the exact tools it needs, DeepAgent 32B RL with a QwQ 32B backbone reports 69.0 on ToolBench, 75.3 on API Bank, 89.0 on TMDB, 75.4 on Spotify, and 51.3 on ToolHop, which is the strongest 32B level result across all 5 datasets. Workflow baselines such as ReAct and CodeAct can match single datasets, for example ReAct with strong models is high on TMDB and Spotify, but none of them stay high on all 5, so the fair summary is that DeepAgent is more uniform, not that others are always low.

In the open set retrieval setting, which is the realistic one, DeepAgent must first find tools and then call them. Here DeepAgent 32B RL reaches 64.0 on ToolBench and 40.6 on ToolHop, while the strongest workflow baselines reach 55.0 on ToolBench and 36.2 on ToolHop, so the end to end agent still holds the lead. The research team also shows that autonomous tool retrieval itself lifts workflow agents, but DeepAgent gains more, which confirms that the architecture and the training are matched to large toolsets.

https://arxiv.org/pdf/2510.21618

Downstream Environments

On ALFWorld, WebShop, GAIA, and HLE, all under a 32B reasoning model, DeepAgent reports 91.8 percent success on ALFWorld, 34.4 percent success and 56.3 score on WebShop, 53.3 on GAIA, and a higher score than workflow agents on HLE. These tasks are longer and noisier, so the combination of memory folding and ToolPO is the likely source of the gap.

Key Takeaways

DeepAgent keeps the whole agent loop inside one reasoning stream, the model can think, search tools, call them, and continue, so it is not limited to a fixed ReAct style workflow.

It uses dense retrieval over large tool registries, 16,000 plus RapidAPI tools and about 3,900 ToolHop tools, so tools do not have to be pre listed in the prompt, they are discovered on demand.

The autonomous memory folding module compresses long interaction histories into episodic, working, and tool memories, which prevents context overflow and keeps long horizon reasoning stable.

Tool Policy Optimization, ToolPO, trains tool use end to end with simulated APIs and token level advantage attribution, so the agent learns to issue correct tool calls, not only to reach the final answer.

On 5 tool benchmarks and 4 downstream tasks, DeepAgent at 32B scale is more consistent than workflow baselines in both labeled tool and open set settings, especially on ToolBench and ToolHop where tool discovery matters most.

https://arxiv.org/pdf/2510.21618

Editorial Comments

DeepAgent is a practical step toward agent architectures that do not depend on fixed tool prompts, because it unifies autonomous thinking, dense tool retrieval over 16,000 plus RapidAPIs and 3,900 plus ToolHop tools, structured tool calling, and memory folding in one loop. The use of LLM simulated APIs in ToolPO is an engineering choice, but it solves the latency and instability problem that hurts prior tool agents. The evaluation shows consistent 32B level gains in both labeled tool and open set settings, not isolated peaks. This release makes large toolspaces actually usable for LLM agents. Overall, DeepAgent confirms that end to end tool agents with memory and RL are emerging as the default pattern.

Check out the Paper and GitHub Repo. Feel free to check out our GitHub Page for Tutorials, Codes and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post DeepAgent: A Deep Reasoning AI Agent that Performs Autonomous Thinking, Tool Discovery, and Action Execution within a Single Reasoning Process appeared first on MarkTechPost.

How do you tell whether a model is actually noticing its own internal state instead of just repeating what training data said about thinking? In a latest Anthropic’s research study ‘Emergent Introspective Awareness in Large Language Models‘ asks whether current Claude models can do more than talk about their abilities, it asks whether they can notice real changes inside their network. To remove guesswork, the research team does not test on text alone, they directly edit the model’s internal activations and then ask the model what happened. This lets them tell apart genuine introspection from fluent self description.

Method, concept injection as activation steering

The core method is concept injection, described in the Transformer Circuits write up as an application of activation steering. The researchers first capture an activation pattern that corresponds to a concept, for example an all caps style or a concrete noun, then they add that vector into the activations of a later layer while the model is answering. If the model then says, there is an injected thought that matches X, that answer is causally grounded in the current state, not in prior internet text. Anthropic research team reports that this works best in later layers and with tuned strength.

https://transformer-circuits.pub/2025/introspection/index.html

Main result, about 20 percent success with zero false positives in controls

Claude Opus 4 and Claude Opus 4.1 show the clearest effect. When the injection is done in the correct layer band and with the right scale, the models correctly report the injected concept in about 20 percent of trials. On control runs with no injection, production models do not falsely claim to detect an injected thought over 100 runs, which makes the 20 percent signal meaningful.

Separating internal concepts from user text

A natural objection is that the model could be importing the injected word into the text channel. Anthropic researchers tests this. The model receives a normal sentence, the researchers inject an unrelated concept such as bread on the same tokens, and then they ask the model to name the concept and to repeat the sentence. The stronger Claude models can do both, they keep the user text intact and they name the injected thought, which shows that internal concept state can be reported separately from the visible input stream. For agent style systems, this is the interesting part, because it shows that a model can talk about the extra state that tool calls or agents may depend on.

Prefill, using introspection to tell what was intended

Another experiment targets an evaluation problem. Anthropic prefilled the assistant message with content the model did not plan. By default Claude says that the output was not intended. When the researchers retroactively inject the matching concept into earlier activations, the model now accepts the prefilled output as its own and can justify it. This shows that the model is consulting an internal record of its previous state to decide authorship, not only the final text. That is a concrete use of introspection.

Key Takeaways

Concept injection gives causal evidence of introspection: Anthropic shows that if you take a known activation pattern, inject it into Claude’s hidden layers, and then ask the model what is happening, advanced Claude variants can sometimes name the injected concept. This separates real introspection from fluent roleplay.

Best models succeed only in a narrow regime: Claude Opus 4 and 4.1 detect injected concepts only when the vector is added in the right layer band and with tuned strength, and the reported success rate is around the same scale Anthropic stated, while production runs show 0 false positives in controls, so the signal is real but small.

Models can keep text and internal ‘thoughts’ separate: In experiments where an unrelated concept is injected on top of normal input text, the model can both repeat the user sentence and report the injected concept, which means the internal concept stream is not just leaking into the text channel.

Introspection supports authorship checks: When Anthropic prefilled outputs that the model did not intend, the model disavowed them, but if the matching concept was retroactively injected, the model accepted the output as its own. This shows the model can consult past activations to decide whether it meant to say something.

This is a measurement tool, not a consciousness claim: The research team frame the work as functional, limited introspective awareness that could feed future transparency and safety evaluations, including ones about evaluation awareness, but they do not claim general self awareness or stable access to all internal features.

Editorial Comments

Anthropic’s ‘Emergent Introspective Awareness in LLMs‘ research is a useful measurement advance, not a grand metaphysical claim. The setup is clean, inject a known concept into hidden activations using activation steering, then query the model for a grounded self report. Claude variants sometimes detect and name the injected concept, and they can keep injected ‘thoughts’ distinct from input text, which is operationally relevant for agent debugging and audit trails. The research team also shows limited intentional control of internal states. Constraints remain strong, effects are narrow, and reliability is modest, so downstream use should be evaluative, not safety critical.

Check out the Paper and Technical details. Feel free to check out our GitHub Page for Tutorials, Codes and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post Anthropic’s New Research Shows Claude can Detect Injected Concepts, but only in Controlled Layers appeared first on MarkTechPost.

OpenAI has released a research preview of gpt-oss-safeguard, two open weight safety reasoning models that let developers apply custom safety policies at inference time. The models come in two sizes, gpt-oss-safeguard-120b and gpt-oss-safeguard-20b, both fine tuned from gpt-oss, both licensed under Apache 2.0, and both available on Hugging Face for local use.

https://openai.com/index/introducing-gpt-oss-safeguard/

Why Policy-Conditioned Safety Matters?

Conventional moderation models are trained on a single fixed policy. When that policy changes, the model must be retrained or replaced. gpt-oss-safeguard reverses this relationship. It takes the developer authored policy as input together with the user content, then reasons step by step to decide whether the content violates the policy. This turns safety into a prompt and evaluation task, which is better suited for fast changing or domain specific harms such as fraud, biology, self harm or game specific abuse.

Same Pattern as OpenAI’s Internal Safety Reasoner

OpenAI states that gpt-oss-safeguard is an open weight implementation of the Safety Reasoner used internally across systems like GPT 5, ChatGPT Agent and Sora 2. In production settings OpenAI already runs small high recall filters first, then escalates uncertain or sensitive items to a reasoning model, and in recent launches up to 16 percent of total compute was spent on safety reasoning. The open release lets external teams reproduce this defense in depth pattern instead of guessing how OpenAI’s stack works.

Model Sizes and Hardware Fit

The large model, gpt-oss-safeguard-120b, has 117B parameters with 5.1B active parameters and is sized to fit on a single 80GB H100 class GPU. The smaller gpt-oss-safeguard-20b has 21B parameters with 3.6B active parameters and targets lower latency or smaller GPUs, including 16GB setups. Both models were trained on the harmony response format, so prompts must follow that structure otherwise results will degrade. The license is Apache 2.0, the same as the parent gpt-oss models, so commercial local deployment is permitted.

https://openai.com/index/introducing-gpt-oss-safeguard/

Evaluation Results

OpenAI evaluated the models on internal multi policy tests and on public datasets. In multi policy accuracy, where the model must correctly apply several policies at once, gpt-oss-safeguard and OpenAI’s internal Safety Reasoner outperform gpt-5-thinking and the open gpt-oss baselines. On the 2022 moderation dataset the new models slightly outperform both gpt-5-thinking and the internal Safety Reasoner, however OpenAI specifies that this gap is not statistically significant, so it should not be oversold. On ToxicChat, the internal Safety Reasoner still leads, with gpt-oss-safeguard close behind. This places the open models in the competitive range for real moderation tasks.

Recommended Deployment Pattern

OpenAI is explicit that pure reasoning on every request is expensive. The recommended setup is to run small, fast, high recall classifiers on all traffic, then send only uncertain or sensitive content to gpt-oss-safeguard, and when user experience requires fast responses, to run the reasoner asynchronously. This mirrors OpenAI’s own production guidance and reflects the fact that dedicated task specific classifiers can still win when there is a large high quality labeled dataset.

Key Takeaways

gpt-oss-safeguard is a research preview of two open weight safety reasoning models, 120b and 20b, that classify content using developer supplied policies at inference time, so policy changes do not require retraining.

The models implement the same Safety Reasoner pattern OpenAI uses internally across GPT 5, ChatGPT Agent and Sora 2, where a first fast filter routes only risky or ambiguous content to a slower reasoning model.

Both models are fine tuned from gpt-oss, keep the harmony response format, and are sized for real deployments, the 120b model fits on a single H100 class GPU, the 20b model targets 16GB level hardware, and both are Apache 2.0 on Hugging Face.

On internal multi policy evaluations and on the 2022 moderation dataset, the safeguard models outperform gpt-5-thinking and the gpt-oss baselines, but OpenAI notes that the small margin over the internal Safety Reasoner is not statistically significant.

OpenAI recommends using these models in a layered moderation pipeline, together with community resources such as ROOST, so platforms can express custom taxonomies, audit the chain of thought, and update policies without touching weights.

Editorial Comments

OpenAI is taking an internal safety pattern and making it reproducible, which is the most important part of this launch. The models are open weight, policy conditioned and Apache 2.0, so platforms can finally apply their own taxonomies instead of accepting fixed labels. The fact that gpt-oss-safeguard matches and sometimes slightly exceeds the internal Safety Reasoner on the 2022 moderation dataset, while outperforming gpt-5-thinking on multi policy accuracy, but with a non statistically significant margin, shows the approach is already usable. The recommended layered deployment is realistic for production.
The post OpenAI Releases Research Preview of ‘gpt-oss-safeguard’: Two Open-Weight Reasoning Models for Safety Classification Tasks appeared first on MarkTechPost.

In this tutorial, we build an Agentic Data and Infrastructure Strategy system using the lightweight Qwen2.5-0.5B-Instruct model for efficient execution. We begin by creating a flexible LLM agent framework and then develop specialized agents that handle different layers of data management, from ingestion and quality analysis to infrastructure optimization. We integrate these agents into an orchestrator that coordinates their interactions, ensuring smooth multi-agent collaboration across the data pipeline. Through hands-on examples like e-commerce and IoT pipelines, we explore how autonomous decision-making can streamline complex data operations. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browser!pip install -q transformers torch accelerate datasets huggingface_hub
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
import json, time
from typing import List, Dict, Any
from dataclasses import dataclass
from datetime import datetime
import pandas as pd

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”
print(f”Loading {model_name} for {role} agent on {self.device}…”)
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

We start by setting up the lightweight LLM agent infrastructure using the Qwen2.5-0.5B-Instruct model. We load the model and tokenizer, and define a base agent class capable of handling contextual conversations and generating intelligent responses. This forms the core foundation upon which our specialized agents operate efficiently within Colab. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserclass 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()}

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”

We design the Data Ingestion and Data Quality agents to focus on structured analysis of data pipelines. We let the ingestion agent determine the best approach to data flow, while the quality agent evaluates data completeness, consistency, and issues to provide actionable insights. Together, they establish the first two layers of autonomous data management. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserclass 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”

We develop the Infrastructure Optimization Agent to continuously analyze key metrics like CPU, memory, and storage utilization. We use it to generate intelligent optimization suggestions, helping us maintain high performance and resource efficiency. This agent ensures that our infrastructure remains responsive and scalable during data operations. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserclass AgenticDataOrchestrator:
def __init__(self):
print(“n” + “=”*70)
print(“Initializing Agentic Data Infrastructure System”)
print(“=”*70 + “n”)
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”: []}
print(“n[Stage 1] Data Ingestion Analysis”)
ingestion_result = self.ingestion_agent.analyze_data_source(pipeline_config.get(“source”, {}))
print(f”Strategy: {ingestion_result[‘strategy’][:150]}…”)
results[“stages”].append({“stage”: “ingestion”, “result”: ingestion_result})
print(“n[Stage 2] Data Quality Assessment”)
quality_result = self.quality_agent.assess_data_quality(pipeline_config.get(“quality_metrics”, {}))
print(f”Assessment: {quality_result[‘assessment’][:150]}…”)
print(f”Severity: {quality_result[‘severity’]}”)
results[“stages”].append({“stage”: “quality”, “result”: quality_result})
print(“n[Stage 3] Infrastructure Optimization”)
optimization_result = self.optimization_agent.optimize_resources(pipeline_config.get(“infrastructure_metrics”, {}))
print(f”Recommendations: {optimization_result[‘recommendations’][:150]}…”)
print(f”Priority: {optimization_result[‘priority’]}”)
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)

We built an Agentic Data Orchestrator to coordinate all specialized agents under a unified workflow. We use it to manage end-to-end pipeline execution, triggering ingestion, quality checks, and optimization sequentially. By doing this, we bring structure, collaboration, and automation to the entire multi-agent system. Check out the FULL CODES here.

Copy CodeCopiedUse a different Browserdef main():
orchestrator = AgenticDataOrchestrator()
print(“n” + “=”*70)
print(“EXAMPLE 1: E-commerce Data Pipeline”)
print(“=”*70)
ecommerce_pipeline = {
“id”: “ecommerce_pipeline_001”,
“source”: {“type”: “REST API”, “volume”: “10GB/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)
print(“nn” + “=”*70)
print(“EXAMPLE 2: IoT Sensor Data Pipeline”)
print(“=”*70)
iot_pipeline = {
“id”: “iot_pipeline_002”,
“source”: {“type”: “Message Queue (Kafka)”, “volume”: “50GB/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)
print(“nn” + “=”*70)
print(“EXECUTION SUMMARY REPORT”)
print(“=”*70 + “n”)
summary_df = orchestrator.generate_summary_report()
print(summary_df.to_string(index=False))
print(“n” + “=”*70)
print(“Tutorial Complete!”)
print(“=”*70)
print(“nKey Concepts Demonstrated:”)
print(“✓ Lightweight LLM agent architecture”)
print(“✓ Specialized agents for different data tasks”)
print(“✓ Multi-agent orchestration”)
print(“✓ Infrastructure monitoring and optimization”)
print(“✓ Autonomous decision-making in data pipelines”)

if __name__ == “__main__”:
main()

We demonstrate our complete system through two real-world examples, an e-commerce and an IoT data pipeline. We observe how each agent performs its role autonomously while contributing to a shared objective. Finally, we generate a summary report, confirming the orchestration’s efficiency and the power of lightweight agentic intelligence.

In conclusion, we design and execute an intelligent, multi-agent data infrastructure framework powered by a compact open-source model. We witness how independent yet cooperative agents can autonomously analyze, assess, and optimize real-world data systems. The entire setup demonstrates how lightweight LLMs can efficiently handle infrastructure intelligence, while also highlighting how agentic orchestration transforms traditional data workflows into adaptive, self-optimizing systems ready for scalable enterprise applications.

Check out the FULL CODES here. Feel free to check out our GitHub Page for Tutorials, Codes and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post How to Design an Autonomous Multi-Agent Data and Infrastructure Strategy System Using Lightweight Qwen Models for Efficient Pipeline Intelligence? appeared first on MarkTechPost.