Procedural Memory: Agent Skill Libraries for Reusable Functions
Source: mortalapps.com- Procedural memory enables AI agents to acquire, store, and retrieve reusable executable skills, akin to human muscle memory or learned procedures.
- This solves the problem of agents repeatedly 're-learning' or hallucinating solutions for common tasks by providing a validated, callable library of functions.
- In production, it enhances agent reliability, reduces token consumption by minimizing redundant reasoning, and improves response consistency.
- Agents can dynamically write new Python functions, validate their schemas, embed them into a vector database, and retrieve them semantically based on user intent for execution.
- Key considerations include robust function schema design, secure code execution environments, and efficient semantic retrieval strategies.
Why This Matters
AI agents often struggle with consistency and efficiency when performing repetitive or complex tasks, frequently re-deriving solutions or failing to execute them correctly. This challenge is addressed by implementing agent procedural memory skill libraries. This approach was invented to provide agents with a persistent, retrievable knowledge base of 'how-to' information, moving beyond declarative facts (episodic memory) or short-term context (working memory). By allowing agents to write, validate, and store reusable Python functions, it creates a robust mechanism for skill acquisition and application. Ignoring procedural memory leads to increased token costs due to redundant LLM calls, higher rates of hallucination in tool use, and brittle agent behavior that struggles with task generalization. In production, this translates to unreliable systems, higher operational expenses, and a poor user experience. Procedural memory is critical when an agent needs to perform complex, multi-step operations consistently, or when new capabilities must be dynamically integrated. It is preferred over simple prompt-based tool descriptions when the number of tools grows large, or when tools themselves need to be dynamically generated and managed, offering a scalable and maintainable alternative for complex agent systems.
Core Concepts
Procedural memory in AI agents refers to the system's ability to acquire, store, and retrieve executable skills or procedures. Unlike episodic memory (past events) or working memory (current context), procedural memory focuses on 'how to do' tasks.
- Skill Library: A collection of executable functions or code snippets that an agent can invoke. These skills are typically stored in a structured manner, often in a vector database, allowing for semantic search and retrieval.
- Function Schema: A formal, machine-readable description of a function's parameters, return types, and purpose. This schema enables an LLM to understand how to correctly call a function, including required arguments and their types. OpenAPI specifications or Pydantic models are common formats.
- Semantic Retrieval: The process of finding relevant skills from the library based on the semantic similarity between a user's query or agent's intent and the embedded representations of the stored skills. Vector embeddings are used to capture the meaning of both queries and skills.
- Tool Calling: The mechanism by which an LLM, after reasoning, decides to invoke an external function or API. Procedural memory skills are integrated into this tool-calling framework, providing the LLM with a dynamic set of available actions.
- Code Validation: Before a newly generated or modified skill is added to the library, it undergoes validation. This includes syntax checks, type checks against its schema, and potentially security scans or sandbox execution to ensure correctness and safety.
- Execution Environment: A secure and isolated environment (e.g., a sandbox, container, or dedicated interpreter) where retrieved skills are executed. This mitigates risks associated with running arbitrary agent-generated code.
How It Works
The lifecycle of an agent's procedural memory system involves skill definition, storage, retrieval, and execution.
1. Skill Definition and Schema Generation
An agent, or a human developer, defines a new skill as a Python function. This function includes a docstring describing its purpose and parameters. A critical step is generating a machine-readable schema (e.g., JSON Schema) from this function. This schema details the function's name, description, and the types and descriptions of its arguments. For robust systems, Pydantic models are often used to define input parameters, which simplifies schema generation and runtime validation.
2. Skill Validation and Embedding
Before storage, the defined skill undergoes validation. This includes static analysis for syntax errors, type checking against its generated schema, and potentially dynamic checks by executing it in a sandboxed environment with mock inputs. If validation passes, the function's description and schema are embedded into a vector using an embedding model. This vector captures the semantic meaning of the skill, enabling future retrieval based on intent.
3. Skill Storage
The validated skill's code, its generated schema, and its semantic embedding are stored in a vector database. The vector database indexes the embeddings for efficient similarity search. The raw function code and schema are typically stored as metadata associated with the embedding, or in a separate persistent store (e.g., a relational database or object storage) with a reference in the vector database.
4. Runtime Retrieval
When an agent receives a task or query, it first generates an embedding of the user's intent or its current sub-task. This query embedding is then used to perform a similarity search against the skill embeddings in the vector database. The top-k most semantically relevant skills (their schemas and descriptions) are retrieved. If the retrieval fails to yield relevant skills, the agent may attempt to rephrase its query, perform a broader search, or escalate to a human.
5. LLM Integration and Tool Calling
The retrieved skill schemas and descriptions are provided to the LLM as part of its context window, alongside the user's original query. The LLM, using its reasoning capabilities, determines if any of the provided skills are appropriate for the current task. If it decides to use a skill, it generates a tool call, specifying the skill's name and the arguments based on its schema. If the LLM generates an invalid tool call (e.g., missing required arguments, incorrect types), the system captures this error, logs it, and can prompt the LLM to correct its call or mark the skill as problematic.
6. Skill Execution
The generated tool call is intercepted by the agent's runtime. The arguments are validated against the skill's schema. If valid, the corresponding function code is retrieved and executed within a secure, isolated environment (e.g., a Docker container or a dedicated Python sandbox). The output of the skill execution is then returned to the LLM for further processing or to the user. Execution failures (e.g., runtime errors, timeouts) are caught, logged, and reported back to the LLM, allowing it to attempt recovery, select an alternative skill, or inform the user of the failure.
Architecture
The procedural memory system architecture consists of several interconnected components designed for skill management and execution.
Execution begins with an Agent Orchestrator receiving a user request. This orchestrator, often an LLM-powered agent, initiates a Query Embedding Service to vectorize the user's intent. The resulting query embedding is sent to the Vector Database, which acts as the central repository for procedural skills. The Vector Database stores skill embeddings, along with associated metadata like function code, JSON schema, and descriptions.
Upon receiving the query embedding, the Vector Database performs a semantic similarity search, returning the top-k most relevant skill schemas and descriptions to the Agent Orchestrator. The Orchestrator then incorporates these retrieved skill definitions into the LLM's context window. The LLM, acting as a Tool Calling Engine, generates a tool call based on the user's request and the available skills.
This tool call is routed to a Skill Execution Manager. The manager first retrieves the full function code for the requested skill from the Vector Database (or a linked persistent storage). It then performs input validation against the skill's schema. Validated calls are dispatched to a Sandboxed Execution Environment, which could be a containerized service or a dedicated Python interpreter, ensuring isolation and security. The execution environment runs the skill and returns its output to the Skill Execution Manager, which then relays it back to the Agent Orchestrator for further processing or direct user response. Error handling and logging are integrated across all components, particularly within the Skill Execution Manager and Sandboxed Execution Environment.
Designing Robust Function Schemas
Effective procedural memory relies on precise function schemas. These schemas enable LLMs to understand the purpose, inputs, and outputs of a skill without needing to parse the actual code. While simple docstring parsing can generate basic schemas, using a library like Pydantic for defining function inputs and outputs offers significant advantages. Pydantic models provide strong type hints, default values, and validation rules, which translate directly into robust JSON schemas. This ensures that the LLM receives clear, unambiguous instructions and that runtime arguments conform to expectations. For example, a Pydantic-defined function can automatically generate a JSON schema that includes title, description, type, properties, and required fields, making it highly interpretable by LLMs and facilitating strict input validation before execution.
Skill Validation Pipeline
Before a new skill is committed to the procedural memory, a multi-stage validation pipeline is crucial. Initial validation involves static analysis: checking for syntax errors (e.g., using Python's ast module or pyflakes), and ensuring the function signature matches its declared schema. Dynamic validation is more complex but vital for production systems. This involves executing the skill in a sandboxed environment with synthetic or mock inputs. The sandbox ensures that the skill does not have unintended side effects or access unauthorized resources. Monitoring its execution for unexpected behavior, resource consumption, or security vulnerabilities (e.g., arbitrary file access, network calls) is paramount. Failed validations should trigger alerts and prevent the skill from being stored, providing feedback to the agent or human developer.
Semantic Retrieval Strategies
The efficiency and relevance of skill retrieval directly impact agent performance. While basic vector similarity search is a starting point, advanced strategies improve results. Hybrid retrieval, combining semantic search (vector similarity) with keyword search (e.g., BM25), can capture both conceptual relevance and exact term matches. Re-ranking retrieved skills using a smaller, more powerful cross-encoder model can further refine the relevance of the initial top-k results. Contextual re-ranking, where the LLM itself evaluates the retrieved skill descriptions against the current task context, can also be employed. For very large skill libraries, hierarchical indexing or multi-stage retrieval (e.g., first retrieving categories, then skills within categories) can reduce search latency and improve precision.
Secure Code Execution Environments
Executing agent-generated or dynamically retrieved code introduces significant security risks. A robust procedural memory system must employ secure execution environments. Docker containers or lightweight virtual machines (e.g., gVisor, Firecracker) provide strong isolation, limiting a skill's access to system resources, network, and file system. For Python-specific sandboxing, libraries like restrictedpython or custom exec environments with carefully controlled globals and locals can be used, though these are generally less secure than containerization. Each skill execution should ideally occur in a fresh, ephemeral sandbox instance to prevent state leakage or malicious persistence. Resource limits (CPU, memory, execution time) must be enforced to prevent denial-of-service attacks or runaway processes. Logging all execution attempts, inputs, outputs, and any errors is critical for auditing and debugging.
Versioning and Lifecycle Management
Skills in a procedural memory system are not static. They evolve, require updates, or become deprecated. Implementing a versioning strategy for skills is essential. Each skill should have a unique identifier and a version number. Updates create new versions, allowing agents to specify which version of a skill to use or to automatically use the latest stable version. A lifecycle management process should define stages like draft, validated, active, deprecated, and archived. This enables controlled rollout of new skills and graceful deprecation of old ones. Monitoring skill usage and performance metrics helps identify underperforming or problematic skills, informing decisions about updates or deprecation. A rollback mechanism should also be in place, allowing quick reversion to a previous stable version if a new skill introduces regressions.
Code Example
import json
import os
import logging
import inspect
from typing import Callable, Dict, Any, Type
from pydantic import BaseModel, Field
import chromadb
from chromadb.utils import embedding_functions
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# --- Configuration ---
# Use a persistent ChromaDB client for demonstration. In production, use a dedicated service.
CHROMA_PATH = os.environ.get("CHROMA_PATH", "./chroma_db")
COLLECTION_NAME = os.environ.get("SKILL_COLLECTION_NAME", "agent_skills")
# OpenAI API Key for embeddings (replace with your preferred embedding provider)
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
if not OPENAI_API_KEY:
logging.error("OPENAI_API_KEY environment variable not set. Embeddings will fail.")
# Fallback to a dummy embedding function if key is missing for local testing
# In production, this should raise an error or use a local model.
class DummyEmbeddingFunction:
def __call__(self, texts):
return [[0.0] * 1536 for _ in texts] # Return dummy embeddings
embedding_function = DummyEmbeddingFunction()
else:
embedding_function = embedding_functions.OpenAIEmbeddingFunction(api_key=OPENAI_API_KEY)
# --- Skill Definition and Schema Generation ---
class SkillInput(BaseModel):
a: float = Field(description="The first number for the operation.")
b: float = Field(description="The second number for the operation.")
class SkillOutput(BaseModel):
result: float = Field(description="The result of the arithmetic operation.")
def add_numbers(a: float, b: float) -> float:
"""Adds two numbers together and returns their sum."""
return a + b
def subtract_numbers(a: float, b: float) -> float:
"""Subtracts the second number from the first and returns the difference."""
return a - b
class Skill:
def __init__(self, func: Callable, input_model: Type[BaseModel], output_model: Type[BaseModel]):
self.func = func
self.name = func.__name__
self.description = func.__doc__.strip() if func.__doc__ else "No description provided."
self.input_model = input_model
self.output_model = output_model
self.schema = self._generate_schema()
def _generate_schema(self) -> Dict[str, Any]:
# Generate JSON schema for tool calling
schema = {
"name": self.name,
"description": self.description,
"parameters": self.input_model.model_json_schema()
}
return schema
def execute(self, **kwargs) -> Any:
# Basic sandboxing: only allow specific built-ins and prevent file/network access
# For production, use a more robust containerized sandbox (e.g., Docker, E2B)
safe_globals = {
'__builtins__': {
'abs': abs, 'all': all, 'any': any, 'bool': bool, 'dict': dict, 'enumerate': enumerate,
'float': float, 'int': int, 'len': len, 'list': list, 'map': map, 'max': max, 'min': min,
'pow': pow, 'range': range, 'round': round, 'set': set, 'str': str, 'sum': sum, 'tuple': tuple,
'type': type, 'zip': zip
},
'add_numbers': add_numbers, # Expose the function itself
'subtract_numbers': subtract_numbers
}
# Validate inputs using Pydantic model
try:
validated_inputs = self.input_model(**kwargs)
except Exception as e:
logging.error(f"Input validation failed for skill {self.name}: {e}")
raise ValueError(f"Invalid inputs for skill {self.name}: {e}")
# Execute the function in a restricted environment
try:
# Use inspect to get function arguments and pass only relevant ones
func_args = inspect.signature(self.func).parameters
call_kwargs = {k: getattr(validated_inputs, k) for k in func_args if k in validated_inputs.model_fields}
result = self.func(**call_kwargs)
return self.output_model(result=result).model_dump() # Validate and return output
except Exception as e:
logging.error(f"Error executing skill {self.name}: {e}")
raise RuntimeError(f"Skill execution failed: {e}")
# --- Vector Database Management ---
class SkillLibrary:
def __init__(self, path: str, collection_name: str, embedding_function: Any):
self.client = chromadb.PersistentClient(path=path))
self.collection = self.client.get_or_create_collection(
name=collection_name,
embedding_function=embedding_function
)
logging.info(f"ChromaDB client initialized at {path}, collection: {collection_name}")
def add_skill(self, skill: Skill):
# Create a unique ID for the skill
skill_id = skill.name
# Embed the description for retrieval
embedding_text = f"Name: {skill.name}
Description: {skill.description}
Schema: {json.dumps(skill.schema)}"
self.collection.add(
documents=[embedding_text],
metadatas=[{"name": skill.name, "schema": json.dumps(skill.schema), "code": inspect.getsource(skill.func)}],
ids=[skill_id]
)
logging.info(f"Skill '{skill.name}' added to library.")
def retrieve_skills(self, query: str, n_results: int = 3) -> list[Dict[str, Any]]:
results = self.collection.query(
query_texts=[query],
n_results=n_results,
include=['metadatas']
)
retrieved_skills = []
if results and results['metadatas']:
for metadata in results['metadatas'][0]:
retrieved_skills.append({
"name": metadata['name'],
"schema": json.loads(metadata['schema']),
"code": metadata['code'] # In a real system, you'd load the function dynamically
})
logging.info(f"Retrieved {len(retrieved_skills)} skills for query: '{query}'.")
return retrieved_skills
def get_skill_by_name(self, skill_name: str) -> Skill:
results = self.collection.get(ids=[skill_name], include=['metadatas'])
if results and results['metadatas']:
metadata = results['metadatas'][0]
# Reconstruct the function from code (DANGEROUS in production without sandboxing)
# For this example, we'll assume the original functions are available in scope
# In a real system, you'd dynamically load and sandbox the code.
func_map = {"add_numbers": add_numbers, "subtract_numbers": subtract_numbers}
func = func_map.get(skill_name)
if not func:
raise ValueError(f"Function '{skill_name}' not found in local scope.")
# For simplicity, assuming input/output models are globally accessible or derivable
# In a real system, these would be stored/retrieved with the skill.
return Skill(func, SkillInput, SkillOutput)
raise ValueError(f"Skill '{skill_name}' not found in library.")
# --- Main Execution Flow ---
if __name__ == "__main__":
# 1. Initialize Skill Library
skill_library = SkillLibrary(CHROMA_PATH, COLLECTION_NAME, embedding_function)
# 2. Define and Add Skills
add_skill = Skill(add_numbers, SkillInput, SkillOutput)
subtract_skill = Skill(subtract_numbers, SkillInput, SkillOutput)
skill_library.add_skill(add_skill)
skill_library.add_skill(subtract_skill)
# 3. Simulate Agent Query and Skill Retrieval
user_query = "I need to sum two numbers, 5 and 3."
logging.info(f"
Agent receives query: '{user_query}'")
retrieved_schemas = skill_library.retrieve_skills(user_query, n_results=1)
if retrieved_schemas:
chosen_skill_info = retrieved_schemas[0]
logging.info(f"Agent chose skill: {chosen_skill_info['name']}")
logging.info(f"Skill schema: {json.dumps(chosen_skill_info['schema'], indent=2)}")
# 4. Simulate LLM Tool Call Generation
# In a real scenario, an LLM would generate this based on the schema and query.
# For this example, we'll hardcode a valid call.
tool_call_args = {"a": 5.0, "b": 3.0}
# 5. Execute the Skill
try:
# Retrieve the actual Skill object to execute
skill_to_execute = skill_library.get_skill_by_name(chosen_skill_info['name'])
execution_result = skill_to_execute.execute(**tool_call_args)
logging.info(f"Skill execution successful. Result: {execution_result}")
except (ValueError, RuntimeError) as e:
logging.error(f"Skill execution failed: {e}")
else:
logging.warning("No relevant skills retrieved for the query.")
# Example of a different query
user_query_2 = "What is 10 minus 4?"
logging.info(f"
Agent receives query: '{user_query_2}'")
retrieved_schemas_2 = skill_library.retrieve_skills(user_query_2, n_results=1)
if retrieved_schemas_2:
chosen_skill_info_2 = retrieved_schemas_2[0]
logging.info(f"Agent chose skill: {chosen_skill_info_2['name']}")
logging.info(f"Skill schema: {json.dumps(chosen_skill_info_2['schema'], indent=2)}")
tool_call_args_2 = {"a": 10.0, "b": 4.0}
try:
skill_to_execute_2 = skill_library.get_skill_by_name(chosen_skill_info_2['name'])
execution_result_2 = skill_to_execute_2.execute(**tool_call_args_2)
logging.info(f"Skill execution successful. Result: {execution_result_2}")
except (ValueError, RuntimeError) as e:
logging.error(f"Skill execution failed: {e}")
else:
logging.warning("No relevant skills retrieved for the second query.")
# Example of invalid input
user_query_3 = "Add 7 and 'hello'"
logging.info(f"
Agent receives query: '{user_query_3}'")
retrieved_schemas_3 = skill_library.retrieve_skills(user_query_3, n_results=1)
if retrieved_schemas_3:
chosen_skill_info_3 = retrieved_schemas_3[0]
logging.info(f"Agent chose skill: {chosen_skill_info_3['name']}")
logging.info(f"Skill schema: {json.dumps(chosen_skill_info_3['schema'], indent=2)}")
tool_call_args_3 = {"a": 7.0, "b": "hello"}
try:
skill_to_execute_3 = skill_library.get_skill_by_name(chosen_skill_info_3['name'])
execution_result_3 = skill_to_execute_3.execute(**tool_call_args_3)
logging.info(f"Skill execution successful. Result: {execution_result_3}")
except (ValueError, RuntimeError) as e:
logging.error(f"Skill execution failed as expected: {e}")
else:
logging.warning("No relevant skills retrieved for the third query.")
The script will output log messages indicating the initialization of the skill library, the addition of 'add_numbers' and 'subtract_numbers' skills, and then simulate two successful retrieval and execution flows for adding and subtracting numbers. It will also demonstrate an input validation failure for the third query, as 'hello' is not a valid float for the 'b' parameter of the 'add_numbers' skill.
Example successful output for the first query:
... - INFO - Agent receives query: 'I need to sum two numbers, 5 and 3.'
... - INFO - Retrieved 1 skills for query: 'I need to sum two numbers, 5 and 3.'.
... - INFO - Agent chose skill: add_numbers
... - INFO - Skill schema: {"name": "add_numbers", "description": "Adds two numbers together and returns their sum.", "parameters": {"properties": {"a": {"description": "The first number for the operation.", "title": "A", "type": "number"}, "b": {"description": "The second number for the operation.", "title": "B", "type": "number"}}, "required": ["a", "b"], "title": "SkillInput", "type": "object"}}
... - INFO - Skill execution successful. Result: {'result': 8.0}
Example error output for the third query:
... - INFO - Agent receives query: 'Add 7 and 'hello''
... - INFO - Retrieved 1 skills for query: 'Add 7 and 'hello''.
... - INFO - Agent chose skill: add_numbers
... - INFO - Skill schema: {...}
... - ERROR - Input validation failed for skill add_numbers: 1 validation error for SkillInput
b
Input should be a valid 'float' [type=float_type, input_value='hello', input_type=str]
... - ERROR - Skill execution failed as expected: Invalid inputs for skill add_numbers: 1 validation error for SkillInput
b
Input should be a valid 'float' [type=float_type, input_value='hello', input_type=str]