Committee

The AI industry is currently obsessed with ‘agents’—autonomous programs that do more than just chat. However, most current multi-agent systems rely on brittle, hard-coded heuristics that fail when the environment changes. Google DeepMind researchers have proposed a new solution. The research team argued that for the ‘agentic web’ to scale, agents must move beyond simple task-splitting and adopt human-like organizational principles such as authority, responsibility, and accountability. Defining ‘Intelligent’ Delegation In standard software, a subroutine is just ‘outsourced’. Intelligent delegation is different. It is a sequence of decisions where a delegator transfers authority and responsibility to a delegatee. This process involves risk assessment, capability matching, and establishing trust. The 5 Pillars of the Framework To build this, the research team identified 5 core requirements mapped to specific technical protocols: Framework Pillar Technical Implementation Core Function Dynamic Assessment Task Decomposition & Assignment Granularly inferring agent state and capacity. Adaptive Execution Adaptive Coordination Handling context shifts and runtime failures. Structural Transparency Monitoring & Verifiable Completion Auditing both the process and the final outcome. Scalable Market Trust & Reputation & Multi-objective Optimization Efficient, trusted coordination in open markets. Systemic Resilience Security & Permission Handling Preventing cascading failures and malicious use. Engineering Strategy: ‘Contract-First’ Decomposition The most significant shift is contract-first decomposition. Under this principle, a delegator only assigns a task if the outcome can be precisely verified. If a task is too subjective or complex to verify—like ‘write a compelling research paper’—the system must recursively decompose it. This continues until the sub-tasks match available verification tools, such as unit tests or formal mathematical proofs. Recursive Verification: The Chain of Custody In a delegation chain, such as 𝐴 → 𝐵 → 𝐶, accountability is transitive. Agent B is responsible for verifying the work of C. When Agent B returns the result to A, it must provide a full chain of cryptographically signed attestations. Agent A then performs a 2-stage check: verifying B’s direct work and verifying that B correctly verified C. Security: Tokens and Tunnels Scaling these chains introduces massive security risks, including Data Exfiltration, Backdoor Implanting, and Model Extraction. To protect the network, DeepMind team suggests Delegation Capability Tokens (DCTs). Based on technologies like Macaroons or Biscuits, these tokens use ‘cryptographic caveats’ to enforce the principle of least privilege. For example, an agent might receive a token that allows it to READ a specific Google Drive folder but forbids any WRITE operations. Evaluating Current Protocols The research team analyzed whether current industry standards are ready for this framework. While these protocols provide a base, they all have ‘missing pieces’ for high-stakes delegation. MCP (Model Context Protocol): Standardizes how models connect to tools. The Gap: It lacks a policy layer to govern permissions across deep delegation chains. A2A (Agent-to-Agent): Manages discovery and task lifecycles. The Gap: It lacks standardized headers for Zero-Knowledge Proofs (ZKPs) or digital signature chains. AP2 (Agent Payments Protocol): Authorizes agents to spend funds. The Gap: It cannot natively verify the quality of the work before releasing payment. UCP (Universal Commerce Protocol): Standardizes commercial transactions. The Gap: It is optimized for shopping/fulfillment, not abstract computational tasks. Key Takeaways Move Beyond Heuristics: Current AI delegations relies on simple, hard-coded heuristics that are brittle and cannot dynamically adapt to environmental changes or unexpected failures. Intelligent delegation requires an adaptive framework that incorporates transfer of authority, responsibility, and accountability. ‘Contract-First’ Task Decomposition: For complex goals, delegators should use a ‘contract-first’ approach, where tasks are decomposed until the sub-units match specific, automated verification capabilities, such as unit tests or formal proofs. Transitive Accountability in Chains: In long delegation chains (e.g., 𝐴 → 𝐵 → 𝐶), responsibility is transitive. Agent B is responsible for the work of C, and Agent A must verify both B’s direct work and that B correctly verified C’s attestations. Attenuated Security via Tokens: To prevent systemic breaches and the ‘confused deputy problem,’ agents should use Delegation Capability Tokens (DCTs) that provide attenuated authorization. This ensures agents operate under the principle of least privilege, with access restricted to specific subsets of resources and allowable operations. Check out the Paper here. 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 Google DeepMind Proposes New Framework for Intelligent AI Delegation to Secure the Emerging Agentic Web for Future Economies appeared first on MarkTechPost.

In this tutorial, we build a fully stateful personal tutor agent that moves beyond short-lived chat interactions and learns continuously over time. We design the system to persist user preferences, track weak learning areas, and selectively recall only relevant past context when responding. By combining durable storage, semantic retrieval, and adaptive prompting, we demonstrate how an agent can behave more like a long-term tutor than a stateless chatbot. Also, we focus on keeping the agent self-managed, context-aware, and able to improve its guidance without requiring the user to repeat information. Copy CodeCopiedUse a different Browser !pip -q install “langchain>=0.2.12” “langchain-openai>=0.1.20” “sentence-transformers>=3.0.1” “faiss-cpu>=1.8.0.post1” “pydantic>=2.7.0″ import os, json, sqlite3, uuid from datetime import datetime, timezone from typing import List, Dict, Any import numpy as np import faiss from pydantic import BaseModel, Field from sentence_transformers import SentenceTransformer from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.outputs import ChatGeneration, ChatResult DB_PATH=”/content/tutor_memory.db” STORE_DIR=”/content/tutor_store” INDEX_PATH=f”{STORE_DIR}/mem.faiss” META_PATH=f”{STORE_DIR}/mem_meta.json” os.makedirs(STORE_DIR, exist_ok=True) def now(): return datetime.now(timezone.utc).isoformat() def db(): return sqlite3.connect(DB_PATH) def init_db(): c=db(); cur=c.cursor() cur.execute(“””CREATE TABLE IF NOT EXISTS events( id TEXT PRIMARY KEY,user_id TEXT,session_id TEXT,role TEXT,content TEXT,ts TEXT)”””) cur.execute(“””CREATE TABLE IF NOT EXISTS memories( id TEXT PRIMARY KEY,user_id TEXT,kind TEXT,content TEXT,tags TEXT,importance REAL,ts TEXT)”””) cur.execute(“””CREATE TABLE IF NOT EXISTS weak_topics( user_id TEXT,topic TEXT,mastery REAL,last_seen TEXT,notes TEXT,PRIMARY KEY(user_id,topic))”””) c.commit(); c.close() We set up the execution environment and import all required libraries for building a stateful agent. We also define core paths and utility functions for time handling and database connections. It establishes the foundational infrastructure that the rest of the system relies on. Copy CodeCopiedUse a different Browser class MemoryItem(BaseModel): kind:str content:str tags:List[str]=Field(default_factory=list) importance:float=Field(0.5,ge=0,le=1) class WeakTopicSignal(BaseModel): topic:str signal:str evidence:str confidence:float=Field(0.5,ge=0,le=1) class Extracted(BaseModel): memories:List[MemoryItem]=Field(default_factory=list) weak_topics:List[WeakTopicSignal]=Field(default_factory=list) class FallbackTutorLLM(BaseChatModel): @property def _llm_type(self)->str: return “fallback_tutor” def _generate(self, messages, stop=None, run_manager=None, **kwargs)->ChatResult: last=messages[-1].content if messages else “” content=self._respond(last) return ChatResult(generations=[ChatGeneration(message=AIMessage(content=content))]) def _respond(self, text:str)->str: t=text.lower() if “extract_memories” in t: out={“memories”:[],”weak_topics”:[]} if “recursion” in t: out[“weak_topics”].append({“topic”:”recursion”,”signal”:”struggled”, “evidence”:”User indicates difficulty with recursion.”,”confidence”:0.85}) if “prefer” in t or “i like” in t: out[“memories”].append({“kind”:”preference”,”content”:”User prefers concise explanations with examples.”, “tags”:[“style”,”preference”],”importance”:0.55}) return json.dumps(out) if “generate_practice” in t: return “n”.join([ “Targeted Practice (Recursion):”, “1) Implement factorial(n) recursively, then iteratively.”, “2) Recursively sum a list; state the base case explicitly.”, “3) Recursive binary search; return index or -1.”, “4) Trace fibonacci(6) call tree; count repeated subcalls.”, “5) Recursively reverse a string; discuss time/space.”, “Mini-quiz: Why does missing a base case cause infinite recursion?” ]) return “Tell me what you’re studying and what felt hard; I’ll remember and adapt practice next time.” def get_llm(): key=os.environ.get(“OPENAI_API_KEY”,””).strip() if key: from langchain_openai import ChatOpenAI return ChatOpenAI(model=”gpt-4o-mini”,temperature=0.2) return FallbackTutorLLM() We define the database schema and initialize persistent storage for events, memories, and weak topics. We ensure that user interactions and long-term learning signals are stored reliably across sessions. It enables agent memory to be durable beyond a single run. Copy CodeCopiedUse a different Browser EMBED_MODEL=”sentence-transformers/all-MiniLM-L6-v2″ embedder=SentenceTransformer(EMBED_MODEL) def load_meta(): if os.path.exists(META_PATH): with open(META_PATH,”r”) as f: return json.load(f) return [] def save_meta(meta): with open(META_PATH,”w”) as f: json.dump(meta,f) def normalize(x): n=np.linalg.norm(x,axis=1,keepdims=True)+1e-12 return x/n def load_index(dim): if os.path.exists(INDEX_PATH): return faiss.read_index(INDEX_PATH) return faiss.IndexFlatIP(dim) def save_index(ix): faiss.write_index(ix, INDEX_PATH) EXTRACTOR_SYSTEM = ( “You are a memory extractor for a stateful personal tutor.n” “Return ONLY JSON with keys: memories (list of {kind,content,tags,importance}) ” “and weak_topics (list of {topic,signal,evidence,confidence}).n” “Store durable info only; do not store secrets.” ) llm=get_llm() init_db() dim=embedder.encode([“x”],convert_to_numpy=True).shape[1] ix=load_index(dim) meta=load_meta() We define the data models and the fallback language model used when no external API key is available. We formalize how memories and weak-topic signals are represented and extracted. It allows the agent to consistently convert raw conversations into structured, actionable memory. Copy CodeCopiedUse a different Browser def log_event(user_id, session_id, role, content): c=db(); cur=c.cursor() cur.execute(“INSERT INTO events VALUES (?,?,?,?,?,?)”, (str(uuid.uuid4()),user_id,session_id,role,content,now())) c.commit(); c.close() def upsert_weak(user_id, sig:WeakTopicSignal): c=db(); cur=c.cursor() cur.execute(“SELECT mastery,notes FROM weak_topics WHERE user_id=? AND topic=?”,(user_id,sig.topic)) row=cur.fetchone() delta=(-0.10 if sig.signal==”struggled” else 0.10 if sig.signal==”improved” else 0.0)*sig.confidence if row is None: mastery=float(np.clip(0.5+delta,0,1)); notes=sig.evidence cur.execute(“INSERT INTO weak_topics VALUES (?,?,?,?,?)”,(user_id,sig.topic,mastery,now(),notes)) else: mastery=float(np.clip(row[0]+delta,0,1)); notes=(row[1]+” | “+sig.evidence)[-2000:] cur.execute(“UPDATE weak_topics SET mastery=?,last_seen=?,notes=? WHERE user_id=? AND topic=?”, (mastery,now(),notes,user_id,sig.topic)) c.commit(); c.close() def store_memory(user_id, m:MemoryItem): mem_id=str(uuid.uuid4()) c=db(); cur=c.cursor() cur.execute(“INSERT INTO memories VALUES (?,?,?,?,?,?,?)”, (mem_id,user_id,m.kind,m.content,json.dumps(m.tags),float(m.importance),now())) c.commit(); c.close() v=embedder.encode([m.content],convert_to_numpy=True).astype(“float32”) v=normalize(v); ix.add(v) meta.append({“mem_id”:mem_id,”user_id”:user_id,”kind”:m.kind,”content”:m.content, “tags”:m.tags,”importance”:m.importance,”ts”:now()}) save_index(ix); save_meta(meta) We focus on embedding-based semantic memory using vector representations and similarity search. We encode memories, store them in a vector index, and persist metadata for later retrieval. It enables relevance-based recall rather than blindly loading all past context. Copy CodeCopiedUse a different Browser def extract(user_text)->Extracted: msg=”extract_memoriesnnUser message:n”+user_text r=llm.invoke([SystemMessage(content=EXTRACTOR_SYSTEM),HumanMessage(content=msg)]).content try: d=json.loads(r) return Extracted( memories=[MemoryItem(**x) for x in d.get(“memories”,[])], weak_topics=[WeakTopicSignal(**x) for x in d.get(“weak_topics”,[])] ) except: return Extracted() def recall(user_id, query, k=6): if ix.ntotal==0: return [] q=embedder.encode([query],convert_to_numpy=True).astype(“float32”) q=normalize(q) scores, idxs = ix.search(q,k) out=[] for s,i in zip(scores[0].tolist(), idxs[0].tolist()): if i<0 or i>=len(meta): continue m=meta[i] if m[“user_id”]!=user_id or s<0.25: continue out.append({**m,”score”:float(s)}) out.sort(key=lambda r: r[“score”]*(0.6+0.4*r[“importance”]), reverse=True) return out def weak_snapshot(user_id): c=db(); cur=c.cursor() cur.execute(“SELECT topic,mastery,last_seen FROM weak_topics WHERE user_id=? ORDER BY mastery ASC LIMIT 5”,(user_id,)) rows=cur.fetchall(); c.close() return [{“topic”:t,”mastery”:float(m),”last_seen”:ls} for t,m,ls in rows] def tutor_turn(user_id, session_id, user_text): log_event(user_id,session_id,”user”,user_text) ex=extract(user_text) for w in ex.weak_topics: upsert_weak(user_id,w) for m in ex.memories: store_memory(user_id,m) rel=recall(user_id,user_text,k=6) weak=weak_snapshot(user_id) prompt={ “recalled_memories”:[{“kind”:x[“kind”],”content”:x[“content”],”score”:x[“score”]} for x in rel], “weak_topics”:weak, “user_message”:user_text } gen = llm.invoke([SystemMessage(content=”You are a personal tutor. Use recalled_memories only if relevant.”), HumanMessage(content=”generate_practicenn”+json.dumps(prompt))]).content log_event(user_id,session_id,”assistant”,gen) return gen, rel, weak USER_ID=”user_demo” SESSION_ID=str(uuid.uuid4()) print(” Ready. Example run:n”) ans, rel, weak = tutor_turn(USER_ID, SESSION_ID, “Last week I struggled with recursion. I prefer concise explanations.”) print(ans) print(“nRecalled:”, [r[“content”] for r in rel]) print(“Weak topics:”, weak) We orchestrate the full tutor interaction loop, combining extraction, storage, recall, and response generation. We update mastery scores, retrieve relevant memories, and dynamically generate targeted practice. It completes the transformation from a stateless chatbot into a long-term, adaptive tutor. In conclusion, we implemented a tutor agent that remembers, reasons, and adapts across sessions. We showed how structured memory extraction, long-term persistence, and relevance-based recall work together to overcome the “goldfish memory” limitation common in most agents. The resulting system continuously refines its understanding of a user’s weaknesses. It proactively generates targeted practice, demonstrating a practical foundation for building stateful, long-horizon AI agents that improve with sustained interaction. Check out the Full Codes here. Also, feel

Powerful AI now runs on consumer hardware.

The landscape of generative audio is shifting toward efficiency. A new open-source contender, Kani-TTS-2, has been released by the team at nineninesix.ai. This model marks a departure from heavy, compute-expensive TTS systems. Instead, it treats audio as a language, delivering high-fidelity speech synthesis with a remarkably small footprint. Kani-TTS-2 offers a lean, high-performance alternative to closed-source APIs. It is currently available on Hugging Face in both English (EN) and Portuguese (PT) versions. The Architecture: LFM2 and NanoCodec Kani-TTS-2 follows the ‘Audio-as-Language‘ philosophy. The model does not use traditional mel-spectrogram pipelines. Instead, it converts raw audio into discrete tokens using a neural codec. The system relies on a two-stage process: The Language Backbone: The model is built on LiquidAI’s LFM2 (350M) architecture. This backbone generates ‘audio intent’ by predicting the next audio tokens. Because LFM (Liquid Foundation Models) are designed for efficiency, they provide a faster alternative to standard transformers. The Neural Codec: It uses the NVIDIA NanoCodec to turn those tokens into 22kHz waveforms. By using this architecture, the model captures human-like prosody—the rhythm and intonation of speech—without the ‘robotic’ artifacts found in older TTS systems. Efficiency: 10,000 Hours in 6 Hours The training metrics for Kani-TTS-2 are a masterclass in optimization. The English model was trained on 10,000 hours of high-quality speech data. While that scale is impressive, the speed of training is the real story. The research team trained the model in only 6 hours using a cluster of 8 NVIDIA H100 GPUs. This proves that massive datasets no longer require weeks of compute time when paired with efficient architectures like LFM2. Zero-Shot Voice Cloning and Performance The standout feature for developers is zero-shot voice cloning. Unlike traditional models that require fine-tuning for new voices, Kani-TTS-2 uses speaker embeddings. How it works: You provide a short reference audio clip. The result: The model extracts the unique characteristics of that voice and applies them to the generated text instantly. From a deployment perspective, the model is highly accessible: Parameter Count: 400M (0.4B) parameters. Speed: It features a Real-Time Factor (RTF) of 0.2. This means it can generate 10 seconds of speech in roughly 2 seconds. Hardware: It requires only 3GB of VRAM, making it compatible with consumer-grade GPUs like the RTX 3060 or 4050. License: Released under the Apache 2.0 license, allowing for commercial use. Key Takeaways Efficient Architecture: The model uses a 400M parameter backbone based on LiquidAI’s LFM2 (350M). This ‘Audio-as-Language’ approach treats speech as discrete tokens, allowing for faster processing and more human-like intonation compared to traditional architectures. Rapid Training at Scale: Kani-TTS-2-EN was trained on 10,000 hours of high-quality speech data in just 6 hours using 8 NVIDIA H100 GPUs. Instant Zero-Shot Cloning: There is no need for fine-tuning to replicate a specific voice. By providing a short reference audio clip, the model uses speaker embeddings to instantly synthesize text in the target speaker’s voice. High Performance on Edge Hardware: With a Real-Time Factor (RTF) of 0.2, the model can generate 10 seconds of audio in approximately 2 seconds. It requires only 3GB of VRAM, making it fully functional on consumer-grade GPUs like the RTX 3060. Developer-Friendly Licensing: Released under the Apache 2.0 license, Kani-TTS-2 is ready for commercial integration. It offers a local-first, low-latency alternative to expensive closed-source TTS APIs. Check out the Model Weight. 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 Meet ‘Kani-TTS-2’: A 400M Param Open Source Text-to-Speech Model that Runs in 3GB VRAM with Voice Cloning Support appeared first on MarkTechPost.

In this tutorial, we build a self-organizing memory system for an agent that goes beyond storing raw conversation history and instead structures interactions into persistent, meaningful knowledge units. We design the system so that reasoning and memory management are clearly separated, allowing a dedicated component to extract, compress, and organize information. At the same time, the main agent focuses on responding to the user. We use structured storage with SQLite, scene-based grouping, and summary consolidation, and we show how an agent can maintain useful context over long horizons without relying on opaque vector-only retrieval. Copy CodeCopiedUse a different Browser import sqlite3 import json import re from datetime import datetime from typing import List, Dict from getpass import getpass from openai import OpenAI OPENAI_API_KEY = getpass(“Enter your OpenAI API key: “).strip() client = OpenAI(api_key=OPENAI_API_KEY) def llm(prompt, temperature=0.1, max_tokens=500): return client.chat.completions.create( model=”gpt-4o-mini”, messages=[{“role”: “user”, “content”: prompt}], temperature=temperature, max_tokens=max_tokens ).choices[0].message.content.strip() We set up the core runtime by importing all required libraries and securely collecting the API key at execution time. We initialize the language model client and define a single helper function that standardizes all model calls. We ensure that every downstream component relies on this shared interface for consistent generation behavior. Copy CodeCopiedUse a different Browser class MemoryDB: def __init__(self): self.db = sqlite3.connect(“:memory:”) self.db.row_factory = sqlite3.Row self._init_schema() def _init_schema(self): self.db.execute(“”” CREATE TABLE mem_cells ( id INTEGER PRIMARY KEY, scene TEXT, cell_type TEXT, salience REAL, content TEXT, created_at TEXT ) “””) self.db.execute(“”” CREATE TABLE mem_scenes ( scene TEXT PRIMARY KEY, summary TEXT, updated_at TEXT ) “””) self.db.execute(“”” CREATE VIRTUAL TABLE mem_cells_fts USING fts5(content, scene, cell_type) “””) def insert_cell(self, cell): self.db.execute( “INSERT INTO mem_cells VALUES(NULL,?,?,?,?,?)”, ( cell[“scene”], cell[“cell_type”], cell[“salience”], json.dumps(cell[“content”]), datetime.utcnow().isoformat() ) ) self.db.execute( “INSERT INTO mem_cells_fts VALUES(?,?,?)”, ( json.dumps(cell[“content”]), cell[“scene”], cell[“cell_type”] ) ) self.db.commit() We define a structured memory database that persists information across interactions. We create tables for atomic memory units, higher-level scenes, and a full-text search index to enable symbolic retrieval. We also implement the logic to insert new memory entries in a normalized and queryable form. Copy CodeCopiedUse a different Browser def get_scene(self, scene): return self.db.execute( “SELECT * FROM mem_scenes WHERE scene=?”, (scene,) ).fetchone() def upsert_scene(self, scene, summary): self.db.execute(“”” INSERT INTO mem_scenes VALUES(?,?,?) ON CONFLICT(scene) DO UPDATE SET summary=excluded.summary, updated_at=excluded.updated_at “””, (scene, summary, datetime.utcnow().isoformat())) self.db.commit() def retrieve_scene_context(self, query, limit=6): tokens = re.findall(r”[a-zA-Z0-9]+”, query) if not tokens: return [] fts_query = ” OR “.join(tokens) rows = self.db.execute(“”” SELECT scene, content FROM mem_cells_fts WHERE mem_cells_fts MATCH ? LIMIT ? “””, (fts_query, limit)).fetchall() if not rows: rows = self.db.execute(“”” SELECT scene, content FROM mem_cells ORDER BY salience DESC LIMIT ? “””, (limit,)).fetchall() return rows def retrieve_scene_summary(self, scene): row = self.get_scene(scene) return row[“summary”] if row else “” We focus on memory retrieval and scene maintenance logic. We implement safe full-text search by sanitizing user queries and adding a fallback strategy when no lexical matches are found. We also expose helper methods to fetch consolidated scene summaries for long-horizon context building. Copy CodeCopiedUse a different Browser class MemoryManager: def __init__(self, db: MemoryDB): self.db = db def extract_cells(self, user, assistant) -> List[Dict]: prompt = f””” Convert this interaction into structured memory cells. Return JSON array with objects containing: – scene – cell_type (fact, plan, preference, decision, task, risk) – salience (0-1) – content (compressed, factual) User: {user} Assistant: {assistant} “”” raw = llm(prompt) raw = re.sub(r”“`json|“`”, “”, raw) try: cells = json.loads(raw) return cells if isinstance(cells, list) else [] except Exception: return [] def consolidate_scene(self, scene): rows = self.db.db.execute( “SELECT content FROM mem_cells WHERE scene=? ORDER BY salience DESC”, (scene,) ).fetchall() if not rows: return cells = [json.loads(r[“content”]) for r in rows] prompt = f””” Summarize this memory scene in under 100 words. Keep it stable and reusable for future reasoning. Cells: {cells} “”” summary = llm(prompt, temperature=0.05) self.db.upsert_scene(scene, summary) def update(self, user, assistant): cells = self.extract_cells(user, assistant) for cell in cells: self.db.insert_cell(cell) for scene in set(c[“scene”] for c in cells): self.consolidate_scene(scene) We implement the dedicated memory management component responsible for structuring experience. We extract compact memory representations from interactions, store them, and periodically consolidate them into stable scene summaries. We ensure that memory evolves incrementally without interfering with the agent’s response flow. Copy CodeCopiedUse a different Browser class WorkerAgent: def __init__(self, db: MemoryDB, mem_manager: MemoryManager): self.db = db self.mem_manager = mem_manager def answer(self, user_input): recalled = self.db.retrieve_scene_context(user_input) scenes = set(r[“scene”] for r in recalled) summaries = “n”.join( f”[{scene}]n{self.db.retrieve_scene_summary(scene)}” for scene in scenes ) prompt = f””” You are an intelligent agent with long-term memory. Relevant memory: {summaries} User: {user_input} “”” assistant_reply = llm(prompt) self.mem_manager.update(user_input, assistant_reply) return assistant_reply db = MemoryDB() memory_manager = MemoryManager(db) agent = WorkerAgent(db, memory_manager) print(agent.answer(“We are building an agent that remembers projects long term.”)) print(agent.answer(“It should organize conversations into topics automatically.”)) print(agent.answer(“This memory system should support future reasoning.”)) for row in db.db.execute(“SELECT * FROM mem_scenes”): print(dict(row)) We define the worker agent that performs reasoning while remaining memory-aware. We retrieve relevant scenes, assemble contextual summaries, and generate responses grounded in long-term knowledge. We then close the loop by passing the interaction back to the memory manager so the system continuously improves over time. In this tutorial, we demonstrated how an agent can actively curate its own memory and turn past interactions into stable, reusable knowledge rather than ephemeral chat logs. We enabled memory to evolve through consolidation and selective recall, which supports more consistent and grounded reasoning across sessions. This approach provides a practical foundation for building long-lived agentic systems, and it can be naturally extended with mechanisms for forgetting, richer relational memory, or graph-based orchestration as the system grows in complexity. Check out the Full Codes. 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 Build a Self-Organizing Agent Memory System for Long-Term AI Reasoning  appeared first on MarkTechPost.

In this tutorial, we build a complete, production-grade synthetic data pipeline using CTGAN and the SDV ecosystem. We start from raw mixed-type tabular data and progressively move toward constrained generation, conditional sampling, statistical validation, and downstream utility testing. Rather than stopping at sample generation, we focus on understanding how well synthetic data preserves structure, distributions, and predictive signal. This tutorial demonstrates how CTGAN can be used responsibly and rigorously in real-world data science workflows. Copy CodeCopiedUse a different Browser !pip -q install “ctgan” “sdv” “sdmetrics” “scikit-learn” “pandas” “numpy” “matplotlib” import numpy as np import pandas as pd import warnings warnings.filterwarnings(“ignore”) import ctgan, sdv, sdmetrics from ctgan import load_demo, CTGAN from sdv.metadata import SingleTableMetadata from sdv.single_table import CTGANSynthesizer from sdv.cag import Inequality, FixedCombinations from sdv.sampling import Condition from sdmetrics.reports.single_table import DiagnosticReport, QualityReport from sklearn.model_selection import train_test_split from sklearn.metrics import roc_auc_score from sklearn.linear_model import LogisticRegression from sklearn.preprocessing import OneHotEncoder from sklearn.compose import ColumnTransformer from sklearn.pipeline import Pipeline import matplotlib.pyplot as plt print(“Versions:”) print(“ctgan:”, ctgan.__version__) print(“sdv:”, sdv.__version__) print(“sdmetrics:”, sdmetrics.__version__) We set up the environment by installing all required libraries and importing the full dependency stack. We explicitly load CTGAN, SDV, SDMetrics, and downstream ML tooling to ensure compatibility across the pipeline. We also surface library versions to make the experiment reproducible and debuggable. Copy CodeCopiedUse a different Browser real = load_demo().copy() real.columns = [c.strip().replace(” “, “_”) for c in real.columns] target_col = “income” real[target_col] = real[target_col].astype(str) categorical_cols = real.select_dtypes(include=[“object”]).columns.tolist() numerical_cols = [c for c in real.columns if c not in categorical_cols] print(“Rows:”, len(real), “Cols:”, len(real.columns)) print(“Categorical:”, len(categorical_cols), “Numerical:”, len(numerical_cols)) display(real.head()) ctgan_model = CTGAN( epochs=30, batch_size=500, verbose=True ) ctgan_model.fit(real, discrete_columns=categorical_cols) synthetic_ctgan = ctgan_model.sample(5000) print(“Standalone CTGAN sample:”) display(synthetic_ctgan.head()) We load the CTGAN Adult demo dataset and perform minimal normalization on column names and data types. We explicitly identify categorical and numerical columns, which is critical for both CTGAN training and evaluation. We then train a baseline standalone CTGAN model and generate synthetic samples for comparison. Copy CodeCopiedUse a different Browser metadata = SingleTableMetadata() metadata.detect_from_dataframe(data=real) metadata.update_column(column_name=target_col, sdtype=”categorical”) constraints = [] if len(numerical_cols) >= 2: col_lo, col_hi = numerical_cols[0], numerical_cols[1] constraints.append(Inequality(low_column_name=col_lo, high_column_name=col_hi)) print(f”Added Inequality constraint: {col_hi} > {col_lo}”) if len(categorical_cols) >= 2: c1, c2 = categorical_cols[0], categorical_cols[1] constraints.append(FixedCombinations(column_names=[c1, c2])) print(f”Added FixedCombinations constraint on: [{c1}, {c2}]”) synth = CTGANSynthesizer( metadata=metadata, epochs=30, batch_size=500 ) if constraints: synth.add_constraints(constraints) synth.fit(real) synthetic_sdv = synth.sample(num_rows=5000) print(“SDV CTGANSynthesizer sample:”) display(synthetic_sdv.head()) We construct a formal metadata object and attach explicit semantic types to the dataset. We introduce structural constraints using SDV’s constraint graph system, enforcing numeric inequalities and validity of categorical combinations. We then train a CTGAN-based SDV synthesizer that respects these constraints during generation. Copy CodeCopiedUse a different Browser loss_df = synth.get_loss_values() display(loss_df.tail()) x_candidates = [“epoch”, “step”, “steps”, “iteration”, “iter”, “batch”, “update”] xcol = next((c for c in x_candidates if c in loss_df.columns), None) g_candidates = [“generator_loss”, “gen_loss”, “g_loss”] d_candidates = [“discriminator_loss”, “disc_loss”, “d_loss”] gcol = next((c for c in g_candidates if c in loss_df.columns), None) dcol = next((c for c in d_candidates if c in loss_df.columns), None) plt.figure(figsize=(10,4)) if xcol is None: x = np.arange(len(loss_df)) else: x = loss_df[xcol].to_numpy() if gcol is not None: plt.plot(x, loss_df[gcol].to_numpy(), label=gcol) if dcol is not None: plt.plot(x, loss_df[dcol].to_numpy(), label=dcol) plt.xlabel(xcol if xcol is not None else “index”) plt.ylabel(“loss”) plt.legend() plt.title(“CTGAN training losses (SDV wrapper)”) plt.show() cond_col = categorical_cols[0] common_value = real[cond_col].value_counts().index[0] conditions = [Condition({cond_col: common_value}, num_rows=2000)] synthetic_cond = synth.sample_from_conditions( conditions=conditions, max_tries_per_batch=200, batch_size=5000 ) print(“Conditional sampling requested:”, 2000, “got:”, len(synthetic_cond)) print(“Conditional sample distribution (top 5):”) print(synthetic_cond[cond_col].value_counts().head(5)) display(synthetic_cond.head()) We extract and visualize the dynamics of generator and discriminator losses using a version-robust plotting strategy. We perform conditional sampling to generate data under specific attribute constraints and verify that the conditions are satisfied. This demonstrates how CTGAN behaves under guided generation scenarios. Copy CodeCopiedUse a different Browser metadata_dict = metadata.to_dict() diagnostic = DiagnosticReport() diagnostic.generate(real_data=real, synthetic_data=synthetic_sdv, metadata=metadata_dict, verbose=True) print(“Diagnostic score:”, diagnostic.get_score()) quality = QualityReport() quality.generate(real_data=real, synthetic_data=synthetic_sdv, metadata=metadata_dict, verbose=True) print(“Quality score:”, quality.get_score()) def show_report_details(report, title): print(f”n===== {title} details =====”) props = report.get_properties() for p in props: print(f”n— {p} —“) details = report.get_details(property_name=p) try: display(details.head(10)) except Exception: display(details) show_report_details(diagnostic, “DiagnosticReport”) show_report_details(quality, “QualityReport”) train_real, test_real = train_test_split( real, test_size=0.25, random_state=42, stratify=real[target_col] ) def make_pipeline(cat_cols, num_cols): pre = ColumnTransformer( transformers=[ (“cat”, OneHotEncoder(handle_unknown=”ignore”), cat_cols), (“num”, “passthrough”, num_cols), ], remainder=”drop” ) clf = LogisticRegression(max_iter=200) return Pipeline([(“pre”, pre), (“clf”, clf)]) pipe_syn = make_pipeline(categorical_cols, numerical_cols) pipe_syn.fit(synthetic_sdv.drop(columns=[target_col]), synthetic_sdv[target_col]) proba_syn = pipe_syn.predict_proba(test_real.drop(columns=[target_col]))[:, 1] y_true = (test_real[target_col].astype(str).str.contains(“>”)).astype(int) auc_syn = roc_auc_score(y_true, proba_syn) print(“Synthetic-train -> Real-test AUC:”, auc_syn) pipe_real = make_pipeline(categorical_cols, numerical_cols) pipe_real.fit(train_real.drop(columns=[target_col]), train_real[target_col]) proba_real = pipe_real.predict_proba(test_real.drop(columns=[target_col]))[:, 1] auc_real = roc_auc_score(y_true, proba_real) print(“Real-train -> Real-test AUC:”, auc_real) model_path = “ctgan_sdv_synth.pkl” synth.save(model_path) print(“Saved synthesizer to:”, model_path) from sdv.utils import load_synthesizer synth_loaded = load_synthesizer(model_path) synthetic_loaded = synth_loaded.sample(1000) print(“Loaded synthesizer sample:”) display(synthetic_loaded.head()) We evaluate synthetic data using SDMetrics diagnostic and quality reports and a property-level inspection. We validate downstream usefulness by training a classifier on synthetic data and testing it on real data. Finally, we serialize the trained synthesizer and confirm that it can be reloaded and sampled reliably. In conclusion, we demonstrated that synthetic data generation with CTGAN becomes significantly more powerful when paired with metadata, constraints, and rigorous evaluation. By validating both statistical similarity and downstream task performance, we ensured that the synthetic data is not only realistic but also useful. This pipeline serves as a strong foundation for privacy-preserving analytics, data sharing, and simulation workflows. With careful configuration and evaluation, CTGAN can be safely deployed in real-world data science systems. Check out the Full Codes here. 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 [In-Depth Guide] The Complete CTGAN + SDV Pipeline for High-Fidelity Synthetic Data appeared first on MarkTechPost.