mirror of
https://github.com/MervinPraison/PraisonAI.git
synced 2026-11-08 01:08:39 -10:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b183ac5cbd | ||
|
|
5f57e1d2f1 | ||
|
|
5114165944 | ||
|
|
389442ea04 |
No files matched your search
@@ -58,6 +58,8 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
|
||||
- 🌐 **Native Web Search** (OpenAI, Gemini, Anthropic, xAI, Perplexity)
|
||||
- 📥 **Web Fetch** (Retrieve full content from URLs - Anthropic)
|
||||
- 💾 **Prompt Caching** (Reduce costs & latency - OpenAI, Anthropic, Bedrock, Deepseek)
|
||||
- 🧠 **Claude Memory Tool** (Persistent cross-conversation memory - Anthropic Beta)
|
||||
- 💾 **File-Based Memory** (Zero-dependency persistent memory for all agents)
|
||||
|
||||
## Using Python Code
|
||||
|
||||
@@ -174,6 +176,48 @@ result = agent.rewrite("What about cost?", chat_history=[...])
|
||||
- **CONTEXTUAL**: Resolve references using conversation history
|
||||
- **AUTO**: Automatically detect best strategy
|
||||
|
||||
### 5. Agent Memory (Zero Dependencies)
|
||||
|
||||
Enable persistent memory for agents - works out of the box without any extra packages.
|
||||
|
||||
```python
|
||||
from praisonaiagents import Agent
|
||||
|
||||
# Enable memory with a single parameter
|
||||
agent = Agent(
|
||||
name="Personal Assistant",
|
||||
instructions="You are a helpful assistant that remembers user preferences.",
|
||||
memory=True, # Enables file-based memory (no extra deps!)
|
||||
user_id="user123" # Isolate memory per user
|
||||
)
|
||||
|
||||
# Store memories programmatically
|
||||
agent.store_memory("User prefers dark mode", memory_type="short_term")
|
||||
agent.store_memory("User's name is John", memory_type="long_term", importance=0.9)
|
||||
|
||||
# Memory is automatically injected into conversations
|
||||
result = agent.start("What do you know about me?")
|
||||
# Agent will recall: "Your name is John and you prefer dark mode"
|
||||
|
||||
# Memory persists across sessions - create new agent with same user_id
|
||||
agent2 = Agent(name="Assistant", memory=True, user_id="user123")
|
||||
# agent2 automatically loads all previous memories!
|
||||
```
|
||||
|
||||
**Memory Types:**
|
||||
- **Short-term**: Rolling buffer of recent context (auto-expires)
|
||||
- **Long-term**: Persistent important facts (sorted by importance)
|
||||
- **Entity**: People, places, organizations with attributes
|
||||
- **Episodic**: Date-based interaction history
|
||||
|
||||
**Storage Options:**
|
||||
| Option | Dependencies | Description |
|
||||
|--------|-------------|-------------|
|
||||
| `memory=True` | None | File-based JSON storage (default) |
|
||||
| `memory="file"` | None | Explicit file-based storage |
|
||||
| `memory="sqlite"` | Built-in | SQLite with indexing |
|
||||
| `memory="chromadb"` | chromadb | Vector/semantic search |
|
||||
|
||||
## Using No Code
|
||||
|
||||
### Auto Mode:
|
||||
|
||||
+1
-1
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
|
||||
# Install Python packages (using latest versions)
|
||||
RUN pip install --no-cache-dir \
|
||||
flask \
|
||||
"praisonai>=2.3.2" \
|
||||
"praisonai>=2.3.3" \
|
||||
"praisonai[api]" \
|
||||
gunicorn \
|
||||
markdown
|
||||
|
||||
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
|
||||
# Install Python packages (using latest versions)
|
||||
RUN pip install --no-cache-dir \
|
||||
praisonai_tools \
|
||||
"praisonai>=2.3.2" \
|
||||
"praisonai>=2.3.3" \
|
||||
"praisonai[chat]" \
|
||||
"embedchain[github,youtube]"
|
||||
|
||||
|
||||
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
|
||||
# Install Python packages (using latest versions)
|
||||
RUN pip install --no-cache-dir \
|
||||
praisonai_tools \
|
||||
"praisonai>=2.3.2" \
|
||||
"praisonai>=2.3.3" \
|
||||
"praisonai[ui]" \
|
||||
"praisonai[chat]" \
|
||||
"praisonai[realtime]" \
|
||||
|
||||
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
|
||||
# Install Python packages (using latest versions)
|
||||
RUN pip install --no-cache-dir \
|
||||
praisonai_tools \
|
||||
"praisonai>=2.3.2" \
|
||||
"praisonai>=2.3.3" \
|
||||
"praisonai[ui]" \
|
||||
"praisonai[crewai]"
|
||||
|
||||
|
||||
@@ -433,6 +433,74 @@ result = agent.start("Summarize the key terms")
|
||||
|
||||
---
|
||||
|
||||
## Claude Memory Tool (Beta)
|
||||
|
||||
Enable Claude to store and retrieve information across conversations.
|
||||
|
||||
```python
|
||||
from praisonaiagents import Agent
|
||||
|
||||
# Enable Claude memory (Anthropic models only)
|
||||
agent = Agent(
|
||||
name="Research Assistant",
|
||||
llm="anthropic/claude-sonnet-4-20250514",
|
||||
claude_memory=True
|
||||
)
|
||||
|
||||
# Claude will automatically:
|
||||
# 1. Check /memories directory before tasks
|
||||
# 2. Store progress/learnings in files
|
||||
# 3. Reference memories in future conversations
|
||||
result = agent.start("Research AI trends and remember key findings")
|
||||
```
|
||||
|
||||
**Features:**
|
||||
- File-based persistent memory in `.praison/claude_memory/`
|
||||
- Claude autonomously decides what to store/retrieve
|
||||
- Supports: view, create, str_replace, insert, delete, rename commands
|
||||
- Cross-conversation learning
|
||||
|
||||
**Supported Models:** Claude Sonnet 4, Claude Opus 4, Claude Haiku 4.5
|
||||
|
||||
---
|
||||
|
||||
## Memory (Zero Dependencies)
|
||||
|
||||
Enable persistent memory for agents without any extra packages.
|
||||
|
||||
```python
|
||||
from praisonaiagents import Agent
|
||||
|
||||
# Simple: Enable file-based memory (no extra dependencies!)
|
||||
agent = Agent(
|
||||
name="Assistant",
|
||||
memory=True # Uses FileMemory by default
|
||||
)
|
||||
|
||||
# Store memories
|
||||
agent.store_memory("User prefers dark mode", memory_type="short_term")
|
||||
agent.store_memory("User's name is John", memory_type="long_term", importance=0.9)
|
||||
|
||||
# Get memory context for prompts
|
||||
context = agent.get_memory_context(query="What does the user prefer?")
|
||||
```
|
||||
|
||||
**Memory Types:**
|
||||
- **Short-term**: Rolling buffer of recent context (auto-expires)
|
||||
- **Long-term**: Persistent important facts
|
||||
- **Entity**: People, places, organizations
|
||||
- **Episodic**: Date-based interaction history
|
||||
|
||||
**Storage Providers:**
|
||||
| Provider | Dependencies | Use Case |
|
||||
|----------|-------------|----------|
|
||||
| `memory=True` or `"file"` | None | Default, zero-config |
|
||||
| `"sqlite"` | Built-in | Search, indexing |
|
||||
| `"chromadb"` | chromadb | Semantic/vector search |
|
||||
| `"mem0"` | mem0ai | Graph memory, cloud |
|
||||
|
||||
---
|
||||
|
||||
## Guardrails
|
||||
|
||||
Add safety checks and validation to agent responses.
|
||||
|
||||
@@ -237,7 +237,8 @@ class Agent:
|
||||
api_key: Optional[str] = None,
|
||||
web_search: Optional[Union[bool, Dict[str, Any]]] = None,
|
||||
web_fetch: Optional[Union[bool, Dict[str, Any]]] = None,
|
||||
prompt_caching: Optional[bool] = None
|
||||
prompt_caching: Optional[bool] = None,
|
||||
claude_memory: Optional[Union[bool, Any]] = None
|
||||
):
|
||||
"""Initialize an Agent instance.
|
||||
|
||||
@@ -406,7 +407,8 @@ class Agent:
|
||||
metrics=metrics,
|
||||
web_search=web_search,
|
||||
web_fetch=web_fetch,
|
||||
prompt_caching=prompt_caching
|
||||
prompt_caching=prompt_caching,
|
||||
claude_memory=claude_memory
|
||||
)
|
||||
self._using_custom_llm = True
|
||||
except ImportError as e:
|
||||
@@ -444,6 +446,7 @@ class Agent:
|
||||
llm_params['web_search'] = web_search
|
||||
llm_params['web_fetch'] = web_fetch
|
||||
llm_params['prompt_caching'] = prompt_caching
|
||||
llm_params['claude_memory'] = claude_memory
|
||||
self.llm_instance = LLM(**llm_params)
|
||||
self._using_custom_llm = True
|
||||
|
||||
@@ -480,7 +483,8 @@ class Agent:
|
||||
self.max_iter = max_iter
|
||||
self.max_rpm = max_rpm
|
||||
self.max_execution_time = max_execution_time
|
||||
self.memory = memory
|
||||
self._memory_instance = None
|
||||
self._init_memory(memory, user_id)
|
||||
self.verbose = verbose
|
||||
self.allow_delegation = allow_delegation
|
||||
self.step_callback = step_callback
|
||||
@@ -523,6 +527,7 @@ Your Goal: {self.goal}
|
||||
self.web_search = web_search
|
||||
self.web_fetch = web_fetch
|
||||
self.prompt_caching = prompt_caching
|
||||
self.claude_memory = claude_memory
|
||||
|
||||
# Handle web_search fallback: inject DuckDuckGo tool for unsupported models
|
||||
if web_search and not self._model_supports_web_search():
|
||||
@@ -664,6 +669,149 @@ Your Goal: {self.goal}
|
||||
|
||||
return supports_prompt_caching(model_name)
|
||||
|
||||
def _init_memory(self, memory, user_id: Optional[str] = None):
|
||||
"""
|
||||
Initialize memory based on the memory parameter.
|
||||
|
||||
Args:
|
||||
memory: Can be:
|
||||
- True: Use FileMemory with default settings
|
||||
- False/None: No memory
|
||||
- "file": Use FileMemory
|
||||
- "sqlite": Use existing Memory class with SQLite
|
||||
- dict: Configuration for memory
|
||||
- Memory/FileMemory instance: Use directly
|
||||
user_id: User identifier for memory isolation
|
||||
"""
|
||||
self.memory = memory
|
||||
|
||||
if memory is None or memory is False:
|
||||
self._memory_instance = None
|
||||
return
|
||||
|
||||
# Determine user_id
|
||||
mem_user_id = user_id or getattr(self, 'user_id', None) or "default"
|
||||
|
||||
if memory is True or memory == "file":
|
||||
# Use FileMemory (zero dependencies)
|
||||
from ..memory.file_memory import FileMemory
|
||||
self._memory_instance = FileMemory(
|
||||
user_id=mem_user_id,
|
||||
verbose=1 if getattr(self, 'verbose', False) else 0
|
||||
)
|
||||
elif isinstance(memory, str) and memory in ("sqlite", "chromadb", "mem0", "mongodb"):
|
||||
# Use full Memory class with specific provider
|
||||
try:
|
||||
from ..memory.memory import Memory
|
||||
config = {"provider": memory if memory != "sqlite" else "rag"}
|
||||
self._memory_instance = Memory(config)
|
||||
except ImportError:
|
||||
logging.warning(f"Memory provider '{memory}' requires additional dependencies. Falling back to FileMemory.")
|
||||
from ..memory.file_memory import FileMemory
|
||||
self._memory_instance = FileMemory(user_id=mem_user_id)
|
||||
elif isinstance(memory, dict):
|
||||
# Configuration dict
|
||||
provider = memory.get("provider", "file")
|
||||
if provider == "file":
|
||||
from ..memory.file_memory import FileMemory
|
||||
self._memory_instance = FileMemory(
|
||||
user_id=memory.get("user_id", mem_user_id),
|
||||
config=memory
|
||||
)
|
||||
else:
|
||||
try:
|
||||
from ..memory.memory import Memory
|
||||
self._memory_instance = Memory(memory)
|
||||
except ImportError:
|
||||
logging.warning("Full Memory class requires additional dependencies. Falling back to FileMemory.")
|
||||
from ..memory.file_memory import FileMemory
|
||||
self._memory_instance = FileMemory(user_id=mem_user_id)
|
||||
else:
|
||||
# Assume it's already a memory instance
|
||||
self._memory_instance = memory
|
||||
|
||||
def get_memory_context(self, query: Optional[str] = None) -> str:
|
||||
"""
|
||||
Get memory context for the current conversation.
|
||||
|
||||
Args:
|
||||
query: Optional query to focus the context
|
||||
|
||||
Returns:
|
||||
Formatted memory context string
|
||||
"""
|
||||
if not self._memory_instance:
|
||||
return ""
|
||||
|
||||
if hasattr(self._memory_instance, 'get_context'):
|
||||
return self._memory_instance.get_context(query=query)
|
||||
|
||||
return ""
|
||||
|
||||
def store_memory(self, content: str, memory_type: str = "short_term", **kwargs):
|
||||
"""
|
||||
Store content in memory.
|
||||
|
||||
Args:
|
||||
content: Content to store
|
||||
memory_type: Type of memory (short_term, long_term, entity, episodic)
|
||||
**kwargs: Additional arguments for the memory method
|
||||
"""
|
||||
if not self._memory_instance:
|
||||
return
|
||||
|
||||
if memory_type == "short_term" and hasattr(self._memory_instance, 'add_short_term'):
|
||||
self._memory_instance.add_short_term(content, **kwargs)
|
||||
elif memory_type == "long_term" and hasattr(self._memory_instance, 'add_long_term'):
|
||||
self._memory_instance.add_long_term(content, **kwargs)
|
||||
elif memory_type == "entity" and hasattr(self._memory_instance, 'add_entity'):
|
||||
self._memory_instance.add_entity(content, **kwargs)
|
||||
elif memory_type == "episodic" and hasattr(self._memory_instance, 'add_episodic'):
|
||||
self._memory_instance.add_episodic(content, **kwargs)
|
||||
|
||||
def _display_memory_info(self):
|
||||
"""Display memory information to user in a friendly format."""
|
||||
if not self._memory_instance:
|
||||
return
|
||||
|
||||
# Only display once per chat session
|
||||
if hasattr(self, '_memory_displayed') and self._memory_displayed:
|
||||
return
|
||||
self._memory_displayed = True
|
||||
|
||||
stats = self._memory_instance.get_stats()
|
||||
short_count = stats.get('short_term_count', 0)
|
||||
long_count = stats.get('long_term_count', 0)
|
||||
entity_count = stats.get('entity_count', 0)
|
||||
storage_path = stats.get('storage_path', '')
|
||||
|
||||
total_memories = short_count + long_count + entity_count
|
||||
|
||||
if total_memories > 0:
|
||||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
# Build memory info text
|
||||
info_parts = []
|
||||
if long_count > 0:
|
||||
info_parts.append(f"💾 {long_count} long-term")
|
||||
if short_count > 0:
|
||||
info_parts.append(f"⚡ {short_count} short-term")
|
||||
if entity_count > 0:
|
||||
info_parts.append(f"👤 {entity_count} entities")
|
||||
|
||||
memory_text = Text()
|
||||
memory_text.append("🧠 Memory loaded: ", style="bold cyan")
|
||||
memory_text.append(" | ".join(info_parts))
|
||||
memory_text.append(f"\n📁 Storage: {storage_path}", style="dim")
|
||||
|
||||
self.console.print(Panel(
|
||||
memory_text,
|
||||
title="[bold]Agent Memory[/bold]",
|
||||
border_style="cyan",
|
||||
expand=False
|
||||
))
|
||||
|
||||
@property
|
||||
def llm_model(self):
|
||||
"""Unified property to get the LLM model regardless of configuration type.
|
||||
@@ -910,17 +1058,29 @@ Your Goal: {self.goal}
|
||||
if not self.use_system_prompt:
|
||||
return None
|
||||
|
||||
# Check cache first
|
||||
tools_key = self._get_tools_cache_key(tools)
|
||||
cache_key = f"{self.role}:{self.goal}:{tools_key}"
|
||||
|
||||
if cache_key in self._system_prompt_cache:
|
||||
return self._system_prompt_cache[cache_key]
|
||||
# Check cache first (skip cache if memory is enabled since context is dynamic)
|
||||
if not self._memory_instance:
|
||||
tools_key = self._get_tools_cache_key(tools)
|
||||
cache_key = f"{self.role}:{self.goal}:{tools_key}"
|
||||
|
||||
if cache_key in self._system_prompt_cache:
|
||||
return self._system_prompt_cache[cache_key]
|
||||
else:
|
||||
cache_key = None # Don't cache when memory is enabled
|
||||
|
||||
system_prompt = f"""{self.backstory}\n
|
||||
Your Role: {self.role}\n
|
||||
Your Goal: {self.goal}"""
|
||||
|
||||
# Add memory context if memory is enabled
|
||||
if self._memory_instance:
|
||||
memory_context = self.get_memory_context()
|
||||
if memory_context:
|
||||
system_prompt += f"\n\n## Memory (Information you remember about the user)\n{memory_context}"
|
||||
# Display memory info to user if verbose
|
||||
if self.verbose:
|
||||
self._display_memory_info()
|
||||
|
||||
# Add tool usage instructions if tools are available
|
||||
# Use provided tools or fall back to self.tools
|
||||
tools_to_use = tools if tools is not None else self.tools
|
||||
@@ -950,9 +1110,9 @@ Your Goal: {self.goal}"""
|
||||
if tool_names:
|
||||
system_prompt += f"\n\nYou have access to the following tools: {', '.join(tool_names)}. Use these tools when appropriate to help complete your tasks. Always use tools when they can help provide accurate information or perform actions."
|
||||
|
||||
# Cache the generated system prompt
|
||||
# Cache the generated system prompt (only if cache_key is set, i.e., memory not enabled)
|
||||
# Simple cache size limit to prevent unbounded growth
|
||||
if len(self._system_prompt_cache) < self._max_cache_size:
|
||||
if cache_key and len(self._system_prompt_cache) < self._max_cache_size:
|
||||
self._system_prompt_cache[cache_key] = system_prompt
|
||||
return system_prompt
|
||||
|
||||
|
||||
@@ -242,6 +242,7 @@ class LLM:
|
||||
web_search: Optional[Union[bool, Dict[str, Any]]] = None,
|
||||
web_fetch: Optional[Union[bool, Dict[str, Any]]] = None,
|
||||
prompt_caching: Optional[bool] = None,
|
||||
claude_memory: Optional[Union[bool, Any]] = None,
|
||||
**extra_settings
|
||||
):
|
||||
# Configure logging only once at the class level
|
||||
@@ -279,6 +280,8 @@ class LLM:
|
||||
self.web_search = web_search
|
||||
self.web_fetch = web_fetch
|
||||
self.prompt_caching = prompt_caching
|
||||
self.claude_memory = claude_memory
|
||||
self._claude_memory_tool = None # Lazy initialized
|
||||
self._console = None # Lazy load console when needed
|
||||
self.chat_history = []
|
||||
self.verbose = extra_settings.get('verbose', True)
|
||||
@@ -426,6 +429,70 @@ class LLM:
|
||||
from .model_capabilities import supports_prompt_caching
|
||||
return supports_prompt_caching(self.model)
|
||||
|
||||
def _get_claude_memory_tool(self):
|
||||
"""
|
||||
Get or initialize the Claude Memory Tool instance.
|
||||
|
||||
Returns:
|
||||
ClaudeMemoryTool instance if claude_memory is enabled, None otherwise
|
||||
"""
|
||||
if not self.claude_memory:
|
||||
return None
|
||||
|
||||
if self._claude_memory_tool is None:
|
||||
from ..tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
if isinstance(self.claude_memory, bool):
|
||||
# Create default memory tool
|
||||
self._claude_memory_tool = ClaudeMemoryTool()
|
||||
else:
|
||||
# Use provided memory tool instance
|
||||
self._claude_memory_tool = self.claude_memory
|
||||
|
||||
return self._claude_memory_tool
|
||||
|
||||
def _supports_claude_memory(self) -> bool:
|
||||
"""
|
||||
Check if the current model supports Claude Memory Tool.
|
||||
|
||||
Claude Memory Tool is only supported by Anthropic Claude models.
|
||||
|
||||
Returns:
|
||||
bool: True if the model supports Claude Memory Tool
|
||||
"""
|
||||
if not self.claude_memory:
|
||||
return False
|
||||
|
||||
return self._is_anthropic_model()
|
||||
|
||||
def _is_memory_tool_call(self, function_name: str) -> bool:
|
||||
"""
|
||||
Check if a tool call is for the Claude Memory Tool.
|
||||
|
||||
Args:
|
||||
function_name: Name of the function being called
|
||||
|
||||
Returns:
|
||||
bool: True if this is a memory tool call
|
||||
"""
|
||||
return function_name == "memory" and self._supports_claude_memory()
|
||||
|
||||
def _execute_memory_tool_call(self, arguments: Dict[str, Any]) -> str:
|
||||
"""
|
||||
Execute a Claude Memory Tool call.
|
||||
|
||||
Args:
|
||||
arguments: The arguments from Claude's tool call
|
||||
|
||||
Returns:
|
||||
Result string from the memory operation
|
||||
"""
|
||||
memory_tool = self._get_claude_memory_tool()
|
||||
if not memory_tool:
|
||||
return "Error: Memory tool not initialized"
|
||||
|
||||
return memory_tool.process_tool_call(arguments)
|
||||
|
||||
def _generate_ollama_tool_summary(self, tool_results: List[Any], response_text: str) -> Optional[str]:
|
||||
"""
|
||||
Generate a summary from tool results for Ollama to prevent infinite loops.
|
||||
@@ -3287,6 +3354,28 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
|
||||
|
||||
logging.debug(f"Web fetch enabled with tool: {web_fetch_tool}")
|
||||
|
||||
# Add Claude Memory Tool if enabled and model supports it
|
||||
if self._supports_claude_memory():
|
||||
memory_tool = self._get_claude_memory_tool()
|
||||
if memory_tool:
|
||||
# Add the memory tool definition
|
||||
memory_tool_def = memory_tool.get_tool_definition()
|
||||
|
||||
# Add memory tool to existing tools or create tools list
|
||||
if 'tools' in params and params['tools']:
|
||||
params['tools'].append(memory_tool_def)
|
||||
else:
|
||||
params['tools'] = [memory_tool_def]
|
||||
|
||||
# Add the beta header for Anthropic
|
||||
beta_header = memory_tool.get_beta_header()
|
||||
if 'extra_headers' in params:
|
||||
params['extra_headers']['anthropic-beta'] = beta_header
|
||||
else:
|
||||
params['extra_headers'] = {'anthropic-beta': beta_header}
|
||||
|
||||
logging.debug(f"Claude memory tool enabled with beta header: {beta_header}")
|
||||
|
||||
return params
|
||||
|
||||
def _prepare_response_logging(self, temperature: float, stream: bool, verbose: bool, markdown: bool, **kwargs) -> Optional[Dict[str, Any]]:
|
||||
|
||||
@@ -8,8 +8,19 @@ This module provides memory management capabilities including:
|
||||
- User memory for preferences/history
|
||||
- Quality-based storage decisions
|
||||
- Graph memory support via Mem0
|
||||
|
||||
Memory Providers:
|
||||
- FileMemory: Zero-dependency JSON file-based storage (default)
|
||||
- Memory: Full-featured with SQLite, ChromaDB, Mem0, MongoDB support
|
||||
"""
|
||||
|
||||
from .memory import Memory
|
||||
from .file_memory import FileMemory, create_memory
|
||||
|
||||
__all__ = ["Memory"]
|
||||
# Lazy import for Memory to avoid dependency issues
|
||||
def __getattr__(name):
|
||||
if name == "Memory":
|
||||
from .memory import Memory
|
||||
return Memory
|
||||
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
||||
|
||||
__all__ = ["FileMemory", "Memory", "create_memory"]
|
||||
@@ -0,0 +1,857 @@
|
||||
"""
|
||||
File-based Memory System for PraisonAI Agents.
|
||||
|
||||
Zero-dependency memory implementation using JSON files.
|
||||
Provides short-term, long-term, entity, and episodic memory
|
||||
without requiring any external packages.
|
||||
|
||||
Storage Structure:
|
||||
.praison/memory/{user_id}/
|
||||
├── config.json # Memory configuration
|
||||
├── short_term.json # Rolling buffer (recent context)
|
||||
├── long_term.json # Persistent facts/knowledge
|
||||
├── entities.json # Named entities
|
||||
├── episodic/ # Date-based memories
|
||||
│ └── {date}.json
|
||||
└── summaries.json # LLM-generated summaries
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
import fcntl
|
||||
import hashlib
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Any, Dict, List, Optional
|
||||
from dataclasses import dataclass, asdict, field
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class MemoryItem:
|
||||
"""A single memory item."""
|
||||
id: str
|
||||
content: str
|
||||
metadata: Dict[str, Any] = field(default_factory=dict)
|
||||
created_at: float = field(default_factory=time.time)
|
||||
updated_at: float = field(default_factory=time.time)
|
||||
access_count: int = 0
|
||||
importance: float = 0.5 # 0.0 to 1.0
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
return asdict(self)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> 'MemoryItem':
|
||||
return cls(**data)
|
||||
|
||||
|
||||
@dataclass
|
||||
class EntityItem:
|
||||
"""An entity memory item (person, place, concept)."""
|
||||
id: str
|
||||
name: str
|
||||
entity_type: str # person, place, organization, concept, etc.
|
||||
attributes: Dict[str, Any] = field(default_factory=dict)
|
||||
relationships: List[Dict[str, str]] = field(default_factory=list)
|
||||
created_at: float = field(default_factory=time.time)
|
||||
updated_at: float = field(default_factory=time.time)
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
return asdict(self)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> 'EntityItem':
|
||||
return cls(**data)
|
||||
|
||||
|
||||
class FileMemory:
|
||||
"""
|
||||
Zero-dependency file-based memory system.
|
||||
|
||||
Uses JSON files for persistence with support for:
|
||||
- Short-term memory (rolling buffer)
|
||||
- Long-term memory (persistent facts)
|
||||
- Entity memory (people, places, concepts)
|
||||
- Episodic memory (date-based interactions)
|
||||
- Memory summarization (optional, requires LLM)
|
||||
|
||||
Example:
|
||||
```python
|
||||
memory = FileMemory(user_id="user123")
|
||||
|
||||
# Store short-term memory
|
||||
memory.add_short_term("User prefers dark mode")
|
||||
|
||||
# Store long-term memory
|
||||
memory.add_long_term("User's name is John", importance=0.9)
|
||||
|
||||
# Store entity
|
||||
memory.add_entity("John", "person", {"role": "developer"})
|
||||
|
||||
# Search memories
|
||||
results = memory.search("John")
|
||||
|
||||
# Get context for LLM
|
||||
context = memory.get_context(query="What does John prefer?")
|
||||
```
|
||||
"""
|
||||
|
||||
DEFAULT_CONFIG = {
|
||||
"short_term_limit": 100, # Max items in short-term memory
|
||||
"long_term_limit": 1000, # Max items in long-term memory
|
||||
"episodic_retention_days": 30, # Days to keep episodic memories
|
||||
"enable_summarization": False, # Requires LLM
|
||||
"summarization_threshold": 50, # Summarize after N items
|
||||
"importance_threshold": 0.7, # Min importance for long-term
|
||||
"auto_promote": True, # Auto-promote important short-term to long-term
|
||||
}
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
user_id: str = "default",
|
||||
base_path: Optional[str] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
verbose: int = 0
|
||||
):
|
||||
"""
|
||||
Initialize FileMemory.
|
||||
|
||||
Args:
|
||||
user_id: Unique identifier for the user/session
|
||||
base_path: Base directory for memory storage (default: .praison/memory)
|
||||
config: Configuration overrides
|
||||
verbose: Verbosity level (0=quiet, 1=info, 2+=debug)
|
||||
"""
|
||||
self.user_id = user_id
|
||||
self.verbose = verbose
|
||||
|
||||
# Set up paths
|
||||
if base_path:
|
||||
self.base_path = Path(base_path)
|
||||
else:
|
||||
self.base_path = Path(".praison/memory")
|
||||
|
||||
self.user_path = self.base_path / user_id
|
||||
self.episodic_path = self.user_path / "episodic"
|
||||
|
||||
# Create directories
|
||||
self.user_path.mkdir(parents=True, exist_ok=True)
|
||||
self.episodic_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# File paths
|
||||
self.config_file = self.user_path / "config.json"
|
||||
self.short_term_file = self.user_path / "short_term.json"
|
||||
self.long_term_file = self.user_path / "long_term.json"
|
||||
self.entities_file = self.user_path / "entities.json"
|
||||
self.summaries_file = self.user_path / "summaries.json"
|
||||
|
||||
# Load or create config
|
||||
self.config = self._load_config(config)
|
||||
|
||||
# Initialize memory stores
|
||||
self._short_term: List[MemoryItem] = []
|
||||
self._long_term: List[MemoryItem] = []
|
||||
self._entities: Dict[str, EntityItem] = {}
|
||||
self._summaries: List[Dict[str, Any]] = []
|
||||
|
||||
# Load existing data
|
||||
self._load_all()
|
||||
|
||||
self._log(f"FileMemory initialized for user '{user_id}' at {self.user_path}")
|
||||
|
||||
def _log(self, msg: str, level: int = logging.INFO):
|
||||
"""Log message if verbose."""
|
||||
if self.verbose >= 1:
|
||||
logger.log(level, msg)
|
||||
|
||||
def _generate_id(self, content: str) -> str:
|
||||
"""Generate a unique ID for content."""
|
||||
timestamp = str(time.time())
|
||||
hash_input = f"{content}{timestamp}{self.user_id}"
|
||||
return hashlib.sha256(hash_input.encode()).hexdigest()[:16]
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# File Operations
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def _read_json(self, filepath: Path, default: Any = None) -> Any:
|
||||
"""Read JSON file with file locking."""
|
||||
if not filepath.exists():
|
||||
return default if default is not None else []
|
||||
|
||||
try:
|
||||
with open(filepath, 'r', encoding='utf-8') as f:
|
||||
fcntl.flock(f.fileno(), fcntl.LOCK_SH)
|
||||
try:
|
||||
data = json.load(f)
|
||||
finally:
|
||||
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
|
||||
return data
|
||||
except (json.JSONDecodeError, IOError) as e:
|
||||
self._log(f"Error reading {filepath}: {e}", logging.WARNING)
|
||||
return default if default is not None else []
|
||||
|
||||
def _write_json(self, filepath: Path, data: Any) -> bool:
|
||||
"""Write JSON file with file locking."""
|
||||
try:
|
||||
with open(filepath, 'w', encoding='utf-8') as f:
|
||||
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
|
||||
try:
|
||||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||||
finally:
|
||||
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
|
||||
return True
|
||||
except IOError as e:
|
||||
self._log(f"Error writing {filepath}: {e}", logging.ERROR)
|
||||
return False
|
||||
|
||||
def _load_config(self, overrides: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
"""Load or create configuration."""
|
||||
config = self.DEFAULT_CONFIG.copy()
|
||||
|
||||
# Load existing config
|
||||
if self.config_file.exists():
|
||||
saved_config = self._read_json(self.config_file, {})
|
||||
config.update(saved_config)
|
||||
|
||||
# Apply overrides
|
||||
if overrides:
|
||||
config.update(overrides)
|
||||
|
||||
# Save config
|
||||
self._write_json(self.config_file, config)
|
||||
|
||||
return config
|
||||
|
||||
def _load_all(self):
|
||||
"""Load all memory data from files."""
|
||||
# Load short-term
|
||||
short_data = self._read_json(self.short_term_file, [])
|
||||
self._short_term = [MemoryItem.from_dict(item) for item in short_data]
|
||||
|
||||
# Load long-term
|
||||
long_data = self._read_json(self.long_term_file, [])
|
||||
self._long_term = [MemoryItem.from_dict(item) for item in long_data]
|
||||
|
||||
# Load entities
|
||||
entities_data = self._read_json(self.entities_file, {})
|
||||
self._entities = {k: EntityItem.from_dict(v) for k, v in entities_data.items()}
|
||||
|
||||
# Load summaries
|
||||
self._summaries = self._read_json(self.summaries_file, [])
|
||||
|
||||
self._log(f"Loaded {len(self._short_term)} short-term, {len(self._long_term)} long-term, {len(self._entities)} entities")
|
||||
|
||||
def _save_short_term(self):
|
||||
"""Save short-term memory to file."""
|
||||
data = [item.to_dict() for item in self._short_term]
|
||||
self._write_json(self.short_term_file, data)
|
||||
|
||||
def _save_long_term(self):
|
||||
"""Save long-term memory to file."""
|
||||
data = [item.to_dict() for item in self._long_term]
|
||||
self._write_json(self.long_term_file, data)
|
||||
|
||||
def _save_entities(self):
|
||||
"""Save entities to file."""
|
||||
data = {k: v.to_dict() for k, v in self._entities.items()}
|
||||
self._write_json(self.entities_file, data)
|
||||
|
||||
def _save_summaries(self):
|
||||
"""Save summaries to file."""
|
||||
self._write_json(self.summaries_file, self._summaries)
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Short-Term Memory
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def add_short_term(
|
||||
self,
|
||||
content: str,
|
||||
metadata: Optional[Dict[str, Any]] = None,
|
||||
importance: float = 0.5
|
||||
) -> str:
|
||||
"""
|
||||
Add item to short-term memory.
|
||||
|
||||
Args:
|
||||
content: The memory content
|
||||
metadata: Optional metadata dict
|
||||
importance: Importance score (0.0 to 1.0)
|
||||
|
||||
Returns:
|
||||
The generated memory ID
|
||||
"""
|
||||
item = MemoryItem(
|
||||
id=self._generate_id(content),
|
||||
content=content,
|
||||
metadata=metadata or {},
|
||||
importance=importance
|
||||
)
|
||||
|
||||
self._short_term.append(item)
|
||||
|
||||
# Enforce limit (rolling buffer)
|
||||
limit = self.config["short_term_limit"]
|
||||
if len(self._short_term) > limit:
|
||||
# Before removing, check if should promote to long-term
|
||||
if self.config["auto_promote"]:
|
||||
self._auto_promote_to_long_term()
|
||||
|
||||
# Remove oldest items
|
||||
self._short_term = self._short_term[-limit:]
|
||||
|
||||
self._save_short_term()
|
||||
self._log(f"Added short-term memory: {content[:50]}...")
|
||||
|
||||
return item.id
|
||||
|
||||
def get_short_term(self, limit: Optional[int] = None) -> List[MemoryItem]:
|
||||
"""Get short-term memories, most recent first."""
|
||||
items = list(reversed(self._short_term))
|
||||
if limit:
|
||||
items = items[:limit]
|
||||
return items
|
||||
|
||||
def _auto_promote_to_long_term(self):
|
||||
"""Promote high-importance short-term memories to long-term."""
|
||||
threshold = self.config["importance_threshold"]
|
||||
promoted = []
|
||||
|
||||
for item in self._short_term:
|
||||
if item.importance >= threshold:
|
||||
# Check if not already in long-term
|
||||
existing_ids = {m.id for m in self._long_term}
|
||||
if item.id not in existing_ids:
|
||||
self._long_term.append(item)
|
||||
promoted.append(item.id)
|
||||
|
||||
if promoted:
|
||||
self._save_long_term()
|
||||
self._log(f"Auto-promoted {len(promoted)} items to long-term memory")
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Long-Term Memory
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def add_long_term(
|
||||
self,
|
||||
content: str,
|
||||
metadata: Optional[Dict[str, Any]] = None,
|
||||
importance: float = 0.8
|
||||
) -> str:
|
||||
"""
|
||||
Add item to long-term memory.
|
||||
|
||||
Args:
|
||||
content: The memory content
|
||||
metadata: Optional metadata dict
|
||||
importance: Importance score (0.0 to 1.0)
|
||||
|
||||
Returns:
|
||||
The generated memory ID
|
||||
"""
|
||||
item = MemoryItem(
|
||||
id=self._generate_id(content),
|
||||
content=content,
|
||||
metadata=metadata or {},
|
||||
importance=importance
|
||||
)
|
||||
|
||||
self._long_term.append(item)
|
||||
|
||||
# Enforce limit
|
||||
limit = self.config["long_term_limit"]
|
||||
if len(self._long_term) > limit:
|
||||
# Remove lowest importance items
|
||||
self._long_term.sort(key=lambda x: x.importance, reverse=True)
|
||||
self._long_term = self._long_term[:limit]
|
||||
|
||||
self._save_long_term()
|
||||
self._log(f"Added long-term memory: {content[:50]}...")
|
||||
|
||||
return item.id
|
||||
|
||||
def get_long_term(self, limit: Optional[int] = None) -> List[MemoryItem]:
|
||||
"""Get long-term memories, sorted by importance."""
|
||||
items = sorted(self._long_term, key=lambda x: x.importance, reverse=True)
|
||||
if limit:
|
||||
items = items[:limit]
|
||||
return items
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Entity Memory
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def add_entity(
|
||||
self,
|
||||
name: str,
|
||||
entity_type: str,
|
||||
attributes: Optional[Dict[str, Any]] = None,
|
||||
relationships: Optional[List[Dict[str, str]]] = None
|
||||
) -> str:
|
||||
"""
|
||||
Add or update an entity.
|
||||
|
||||
Args:
|
||||
name: Entity name
|
||||
entity_type: Type (person, place, organization, concept)
|
||||
attributes: Entity attributes
|
||||
relationships: List of relationships [{type, target}]
|
||||
|
||||
Returns:
|
||||
The entity ID
|
||||
"""
|
||||
entity_id = self._generate_id(f"{name}:{entity_type}")
|
||||
|
||||
# Check if entity exists
|
||||
existing = self._find_entity_by_name(name, entity_type)
|
||||
if existing:
|
||||
# Update existing entity
|
||||
existing.attributes.update(attributes or {})
|
||||
if relationships:
|
||||
existing.relationships.extend(relationships)
|
||||
existing.updated_at = time.time()
|
||||
entity_id = existing.id
|
||||
else:
|
||||
# Create new entity
|
||||
entity = EntityItem(
|
||||
id=entity_id,
|
||||
name=name,
|
||||
entity_type=entity_type,
|
||||
attributes=attributes or {},
|
||||
relationships=relationships or []
|
||||
)
|
||||
self._entities[entity_id] = entity
|
||||
|
||||
self._save_entities()
|
||||
self._log(f"Added/updated entity: {name} ({entity_type})")
|
||||
|
||||
return entity_id
|
||||
|
||||
def get_entity(self, name: str, entity_type: Optional[str] = None) -> Optional[EntityItem]:
|
||||
"""Get entity by name and optionally type."""
|
||||
return self._find_entity_by_name(name, entity_type)
|
||||
|
||||
def get_all_entities(self, entity_type: Optional[str] = None) -> List[EntityItem]:
|
||||
"""Get all entities, optionally filtered by type."""
|
||||
entities = list(self._entities.values())
|
||||
if entity_type:
|
||||
entities = [e for e in entities if e.entity_type == entity_type]
|
||||
return entities
|
||||
|
||||
def _find_entity_by_name(self, name: str, entity_type: Optional[str] = None) -> Optional[EntityItem]:
|
||||
"""Find entity by name."""
|
||||
name_lower = name.lower()
|
||||
for entity in self._entities.values():
|
||||
if entity.name.lower() == name_lower:
|
||||
if entity_type is None or entity.entity_type == entity_type:
|
||||
return entity
|
||||
return None
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Episodic Memory
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def add_episodic(
|
||||
self,
|
||||
content: str,
|
||||
metadata: Optional[Dict[str, Any]] = None,
|
||||
date: Optional[str] = None
|
||||
) -> str:
|
||||
"""
|
||||
Add episodic memory (date-based).
|
||||
|
||||
Args:
|
||||
content: The memory content
|
||||
metadata: Optional metadata
|
||||
date: Date string (YYYY-MM-DD), defaults to today
|
||||
|
||||
Returns:
|
||||
The memory ID
|
||||
"""
|
||||
if date is None:
|
||||
date = datetime.now().strftime("%Y-%m-%d")
|
||||
|
||||
episodic_file = self.episodic_path / f"{date}.json"
|
||||
|
||||
# Load existing episodic memories for this date
|
||||
episodes = self._read_json(episodic_file, [])
|
||||
|
||||
# Create new item
|
||||
item = MemoryItem(
|
||||
id=self._generate_id(content),
|
||||
content=content,
|
||||
metadata=metadata or {}
|
||||
)
|
||||
|
||||
episodes.append(item.to_dict())
|
||||
self._write_json(episodic_file, episodes)
|
||||
|
||||
self._log(f"Added episodic memory for {date}: {content[:50]}...")
|
||||
|
||||
return item.id
|
||||
|
||||
def get_episodic(
|
||||
self,
|
||||
date: Optional[str] = None,
|
||||
days_back: int = 7
|
||||
) -> List[MemoryItem]:
|
||||
"""
|
||||
Get episodic memories.
|
||||
|
||||
Args:
|
||||
date: Specific date (YYYY-MM-DD), or None for recent
|
||||
days_back: Number of days to look back if date is None
|
||||
|
||||
Returns:
|
||||
List of episodic memories
|
||||
"""
|
||||
if date:
|
||||
episodic_file = self.episodic_path / f"{date}.json"
|
||||
data = self._read_json(episodic_file, [])
|
||||
return [MemoryItem.from_dict(item) for item in data]
|
||||
|
||||
# Get memories from last N days
|
||||
all_episodes = []
|
||||
for i in range(days_back):
|
||||
check_date = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d")
|
||||
episodic_file = self.episodic_path / f"{check_date}.json"
|
||||
if episodic_file.exists():
|
||||
data = self._read_json(episodic_file, [])
|
||||
all_episodes.extend([MemoryItem.from_dict(item) for item in data])
|
||||
|
||||
return all_episodes
|
||||
|
||||
def cleanup_episodic(self):
|
||||
"""Remove old episodic memories based on retention policy."""
|
||||
retention_days = self.config["episodic_retention_days"]
|
||||
cutoff_date = datetime.now() - timedelta(days=retention_days)
|
||||
|
||||
removed = 0
|
||||
for episodic_file in self.episodic_path.glob("*.json"):
|
||||
try:
|
||||
file_date = datetime.strptime(episodic_file.stem, "%Y-%m-%d")
|
||||
if file_date < cutoff_date:
|
||||
episodic_file.unlink()
|
||||
removed += 1
|
||||
except ValueError:
|
||||
continue
|
||||
|
||||
if removed:
|
||||
self._log(f"Cleaned up {removed} old episodic memory files")
|
||||
|
||||
return removed
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Search & Retrieval
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def search(
|
||||
self,
|
||||
query: str,
|
||||
memory_types: Optional[List[str]] = None,
|
||||
limit: int = 10
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Search across all memory types.
|
||||
|
||||
Args:
|
||||
query: Search query
|
||||
memory_types: List of types to search (short_term, long_term, entity, episodic)
|
||||
limit: Maximum results
|
||||
|
||||
Returns:
|
||||
List of matching memories with scores
|
||||
"""
|
||||
if memory_types is None:
|
||||
memory_types = ["short_term", "long_term", "entity", "episodic"]
|
||||
|
||||
results = []
|
||||
query_lower = query.lower()
|
||||
query_words = set(query_lower.split())
|
||||
|
||||
def score_match(text: str) -> float:
|
||||
"""Simple relevance scoring."""
|
||||
text_lower = text.lower()
|
||||
|
||||
# Exact match
|
||||
if query_lower in text_lower:
|
||||
return 1.0
|
||||
|
||||
# Word overlap
|
||||
text_words = set(text_lower.split())
|
||||
overlap = len(query_words & text_words)
|
||||
if overlap > 0:
|
||||
return overlap / len(query_words) * 0.8
|
||||
|
||||
return 0.0
|
||||
|
||||
# Search short-term
|
||||
if "short_term" in memory_types:
|
||||
for item in self._short_term:
|
||||
score = score_match(item.content)
|
||||
if score > 0:
|
||||
results.append({
|
||||
"type": "short_term",
|
||||
"item": item.to_dict(),
|
||||
"score": score * item.importance
|
||||
})
|
||||
|
||||
# Search long-term
|
||||
if "long_term" in memory_types:
|
||||
for item in self._long_term:
|
||||
score = score_match(item.content)
|
||||
if score > 0:
|
||||
results.append({
|
||||
"type": "long_term",
|
||||
"item": item.to_dict(),
|
||||
"score": score * item.importance
|
||||
})
|
||||
|
||||
# Search entities
|
||||
if "entity" in memory_types:
|
||||
for entity in self._entities.values():
|
||||
score = max(
|
||||
score_match(entity.name),
|
||||
score_match(str(entity.attributes))
|
||||
)
|
||||
if score > 0:
|
||||
results.append({
|
||||
"type": "entity",
|
||||
"item": entity.to_dict(),
|
||||
"score": score
|
||||
})
|
||||
|
||||
# Search episodic (recent only)
|
||||
if "episodic" in memory_types:
|
||||
episodes = self.get_episodic(days_back=7)
|
||||
for item in episodes:
|
||||
score = score_match(item.content)
|
||||
if score > 0:
|
||||
results.append({
|
||||
"type": "episodic",
|
||||
"item": item.to_dict(),
|
||||
"score": score
|
||||
})
|
||||
|
||||
# Sort by score and limit
|
||||
results.sort(key=lambda x: x["score"], reverse=True)
|
||||
return results[:limit]
|
||||
|
||||
def get_context(
|
||||
self,
|
||||
query: Optional[str] = None,
|
||||
include_short_term: bool = True,
|
||||
include_long_term: bool = True,
|
||||
include_entities: bool = True,
|
||||
include_episodic: bool = False,
|
||||
max_tokens: int = 2000
|
||||
) -> str:
|
||||
"""
|
||||
Build context string for LLM from memories.
|
||||
|
||||
Args:
|
||||
query: Optional query to focus context
|
||||
include_short_term: Include short-term memories
|
||||
include_long_term: Include long-term memories
|
||||
include_entities: Include entity information
|
||||
include_episodic: Include recent episodic memories
|
||||
max_tokens: Approximate max tokens (chars / 4)
|
||||
|
||||
Returns:
|
||||
Formatted context string
|
||||
"""
|
||||
context_parts = []
|
||||
char_limit = max_tokens * 4 # Rough estimate
|
||||
|
||||
# If query provided, use search
|
||||
if query:
|
||||
memory_types = []
|
||||
if include_short_term:
|
||||
memory_types.append("short_term")
|
||||
if include_long_term:
|
||||
memory_types.append("long_term")
|
||||
if include_entities:
|
||||
memory_types.append("entity")
|
||||
if include_episodic:
|
||||
memory_types.append("episodic")
|
||||
|
||||
results = self.search(query, memory_types, limit=20)
|
||||
|
||||
if results:
|
||||
context_parts.append("## Relevant Memories")
|
||||
for r in results:
|
||||
if r["type"] == "entity":
|
||||
item = r["item"]
|
||||
context_parts.append(f"- {item['name']} ({item['entity_type']}): {item['attributes']}")
|
||||
else:
|
||||
context_parts.append(f"- {r['item']['content']}")
|
||||
else:
|
||||
# No query - include recent/important memories
|
||||
if include_long_term:
|
||||
long_term = self.get_long_term(limit=10)
|
||||
if long_term:
|
||||
context_parts.append("## Important Facts")
|
||||
for item in long_term:
|
||||
context_parts.append(f"- {item.content}")
|
||||
|
||||
if include_entities:
|
||||
entities = self.get_all_entities()
|
||||
if entities:
|
||||
context_parts.append("## Known Entities")
|
||||
for entity in entities[:10]:
|
||||
attrs = ", ".join(f"{k}={v}" for k, v in entity.attributes.items())
|
||||
context_parts.append(f"- {entity.name} ({entity.entity_type}): {attrs}")
|
||||
|
||||
if include_short_term:
|
||||
short_term = self.get_short_term(limit=10)
|
||||
if short_term:
|
||||
context_parts.append("## Recent Context")
|
||||
for item in short_term:
|
||||
context_parts.append(f"- {item.content}")
|
||||
|
||||
if include_episodic:
|
||||
episodes = self.get_episodic(days_back=3)
|
||||
if episodes:
|
||||
context_parts.append("## Recent Interactions")
|
||||
for item in episodes[:10]:
|
||||
context_parts.append(f"- {item.content}")
|
||||
|
||||
# Join and truncate
|
||||
context = "\n".join(context_parts)
|
||||
if len(context) > char_limit:
|
||||
context = context[:char_limit] + "\n... (truncated)"
|
||||
|
||||
return context
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Summarization
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def add_summary(
|
||||
self,
|
||||
summary: str,
|
||||
source_type: str,
|
||||
source_count: int,
|
||||
metadata: Optional[Dict[str, Any]] = None
|
||||
):
|
||||
"""
|
||||
Add a summary (typically LLM-generated).
|
||||
|
||||
Args:
|
||||
summary: The summary text
|
||||
source_type: What was summarized (short_term, episodic, etc.)
|
||||
source_count: Number of items summarized
|
||||
metadata: Optional metadata
|
||||
"""
|
||||
self._summaries.append({
|
||||
"id": self._generate_id(summary),
|
||||
"summary": summary,
|
||||
"source_type": source_type,
|
||||
"source_count": source_count,
|
||||
"metadata": metadata or {},
|
||||
"created_at": time.time()
|
||||
})
|
||||
|
||||
self._save_summaries()
|
||||
self._log(f"Added summary for {source_count} {source_type} items")
|
||||
|
||||
def get_summaries(self, limit: int = 5) -> List[Dict[str, Any]]:
|
||||
"""Get recent summaries."""
|
||||
return list(reversed(self._summaries))[:limit]
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Utility Methods
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def clear_short_term(self):
|
||||
"""Clear all short-term memory."""
|
||||
self._short_term = []
|
||||
self._save_short_term()
|
||||
self._log("Cleared short-term memory")
|
||||
|
||||
def clear_all(self):
|
||||
"""Clear all memory for this user."""
|
||||
self._short_term = []
|
||||
self._long_term = []
|
||||
self._entities = {}
|
||||
self._summaries = []
|
||||
|
||||
self._save_short_term()
|
||||
self._save_long_term()
|
||||
self._save_entities()
|
||||
self._save_summaries()
|
||||
|
||||
# Clear episodic
|
||||
for f in self.episodic_path.glob("*.json"):
|
||||
f.unlink()
|
||||
|
||||
self._log("Cleared all memory")
|
||||
|
||||
def get_stats(self) -> Dict[str, Any]:
|
||||
"""Get memory statistics."""
|
||||
episodic_count = sum(1 for _ in self.episodic_path.glob("*.json"))
|
||||
|
||||
return {
|
||||
"user_id": self.user_id,
|
||||
"short_term_count": len(self._short_term),
|
||||
"long_term_count": len(self._long_term),
|
||||
"entity_count": len(self._entities),
|
||||
"episodic_days": episodic_count,
|
||||
"summary_count": len(self._summaries),
|
||||
"storage_path": str(self.user_path)
|
||||
}
|
||||
|
||||
def export(self) -> Dict[str, Any]:
|
||||
"""Export all memory data."""
|
||||
return {
|
||||
"user_id": self.user_id,
|
||||
"config": self.config,
|
||||
"short_term": [item.to_dict() for item in self._short_term],
|
||||
"long_term": [item.to_dict() for item in self._long_term],
|
||||
"entities": {k: v.to_dict() for k, v in self._entities.items()},
|
||||
"summaries": self._summaries,
|
||||
"exported_at": time.time()
|
||||
}
|
||||
|
||||
def import_data(self, data: Dict[str, Any]):
|
||||
"""Import memory data from export."""
|
||||
if "short_term" in data:
|
||||
self._short_term = [MemoryItem.from_dict(item) for item in data["short_term"]]
|
||||
self._save_short_term()
|
||||
|
||||
if "long_term" in data:
|
||||
self._long_term = [MemoryItem.from_dict(item) for item in data["long_term"]]
|
||||
self._save_long_term()
|
||||
|
||||
if "entities" in data:
|
||||
self._entities = {k: EntityItem.from_dict(v) for k, v in data["entities"].items()}
|
||||
self._save_entities()
|
||||
|
||||
if "summaries" in data:
|
||||
self._summaries = data["summaries"]
|
||||
self._save_summaries()
|
||||
|
||||
self._log("Imported memory data")
|
||||
|
||||
|
||||
# Convenience function for simple usage
|
||||
def create_memory(
|
||||
user_id: str = "default",
|
||||
**kwargs
|
||||
) -> FileMemory:
|
||||
"""
|
||||
Create a FileMemory instance.
|
||||
|
||||
Args:
|
||||
user_id: User identifier
|
||||
**kwargs: Additional configuration
|
||||
|
||||
Returns:
|
||||
FileMemory instance
|
||||
"""
|
||||
return FileMemory(user_id=user_id, **kwargs)
|
||||
@@ -173,6 +173,9 @@ TOOL_MAPPINGS = {
|
||||
'cot_save': ('.train.data.generatecot', 'GenerateCOT'), # Save single QA to file
|
||||
'cot_upload_to_huggingface': ('.train.data.generatecot', 'GenerateCOT'), # Upload dataset to HuggingFace
|
||||
'cot_tools': ('.train.data.generatecot', 'GenerateCOT'), # Full toolkit access
|
||||
|
||||
# Claude Memory Tool (Anthropic Beta)
|
||||
'ClaudeMemoryTool': ('.claude_memory_tool', 'ClaudeMemoryTool'),
|
||||
}
|
||||
|
||||
_instances = {} # Cache for class instances
|
||||
@@ -184,6 +187,11 @@ def __getattr__(name: str) -> Any:
|
||||
|
||||
module_path, class_name = TOOL_MAPPINGS[name]
|
||||
|
||||
# Return class itself (not instance) for ClaudeMemoryTool
|
||||
if name == 'ClaudeMemoryTool':
|
||||
module = import_module(module_path, __package__)
|
||||
return getattr(module, class_name)
|
||||
|
||||
if class_name is None:
|
||||
# Direct function import
|
||||
module = import_module(module_path, __package__)
|
||||
|
||||
@@ -0,0 +1,529 @@
|
||||
"""
|
||||
Claude Memory Tool - File-based persistent memory for Claude models.
|
||||
|
||||
This module implements Anthropic's Memory Tool API (beta) which enables Claude
|
||||
to store and retrieve information across conversations through a memory file directory.
|
||||
|
||||
The memory tool operates client-side - you control where and how the data is stored.
|
||||
|
||||
Beta Header: context-management-2025-06-27
|
||||
Tool Type: memory_20250818
|
||||
|
||||
Supported Models:
|
||||
- Claude Sonnet 4.5 (claude-sonnet-4-5-20250929)
|
||||
- Claude Sonnet 4 (claude-sonnet-4-20250514)
|
||||
- Claude Haiku 4.5 (claude-haiku-4-5-20251001)
|
||||
- Claude Opus 4.5 (claude-opus-4-5-20251101)
|
||||
- Claude Opus 4.1 (claude-opus-4-1-20250805)
|
||||
- Claude Opus 4 (claude-opus-4-20250514)
|
||||
|
||||
Reference: https://docs.anthropic.com/en/docs/agents-and-tools/tool-use/memory-tool
|
||||
"""
|
||||
|
||||
import shutil
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Memory system prompt that Claude uses to manage memory
|
||||
MEMORY_SYSTEM_PROMPT = """
|
||||
IMPORTANT: ALWAYS VIEW YOUR MEMORY DIRECTORY BEFORE DOING ANYTHING ELSE.
|
||||
|
||||
MEMORY PROTOCOL:
|
||||
1. Use the `view` command of your `memory` tool to check for earlier progress.
|
||||
2. ... (work on the task) ...
|
||||
- As you make progress, record status / progress / thoughts etc in your memory.
|
||||
|
||||
ASSUME INTERRUPTION: Your context window might be reset at any moment, so you risk losing any progress that is not recorded in your memory directory.
|
||||
|
||||
When editing your memory folder, always try to keep its content up-to-date, coherent and organized. You can rename or delete files that are no longer relevant. Do not create new files unless necessary.
|
||||
"""
|
||||
|
||||
|
||||
class ClaudeMemoryTool:
|
||||
"""
|
||||
File-based memory tool for Claude models.
|
||||
|
||||
Implements Anthropic's Memory Tool API (beta). Claude autonomously decides
|
||||
what to store and retrieve from the memory directory.
|
||||
|
||||
Storage: {base_path}/memories/
|
||||
|
||||
Commands:
|
||||
- view: Show directory contents or file contents with optional line ranges
|
||||
- create: Create or overwrite a file
|
||||
- str_replace: Replace text in a file
|
||||
- insert: Insert text at a specific line
|
||||
- delete: Delete a file or directory
|
||||
- rename: Rename or move a file/directory
|
||||
|
||||
Example:
|
||||
```python
|
||||
from praisonaiagents import Agent
|
||||
from praisonaiagents.tools import ClaudeMemoryTool
|
||||
|
||||
# Use default storage
|
||||
agent = Agent(
|
||||
name="Assistant",
|
||||
llm="anthropic/claude-sonnet-4-20250514",
|
||||
claude_memory=True
|
||||
)
|
||||
|
||||
# Or with custom storage path
|
||||
memory = ClaudeMemoryTool(base_path="./my_project/memory")
|
||||
agent = Agent(
|
||||
name="Assistant",
|
||||
llm="anthropic/claude-sonnet-4-20250514",
|
||||
claude_memory=memory
|
||||
)
|
||||
```
|
||||
|
||||
Security:
|
||||
- All paths are validated to prevent directory traversal attacks
|
||||
- Paths must start with /memories
|
||||
- Symlinks are not followed outside the memory directory
|
||||
"""
|
||||
|
||||
# Beta header required for memory tool
|
||||
BETA_HEADER = "context-management-2025-06-27"
|
||||
|
||||
# Tool definition for API requests
|
||||
TOOL_DEFINITION = {
|
||||
"type": "memory_20250818",
|
||||
"name": "memory"
|
||||
}
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_path: Optional[str] = None,
|
||||
user_id: Optional[str] = None,
|
||||
max_file_size: int = 1024 * 1024, # 1MB default
|
||||
max_files: int = 100
|
||||
):
|
||||
"""
|
||||
Initialize the Claude Memory Tool.
|
||||
|
||||
Args:
|
||||
base_path: Base directory for memory storage.
|
||||
Defaults to .praison/claude_memory/{user_id}
|
||||
user_id: User identifier for user-specific memory.
|
||||
Defaults to "default"
|
||||
max_file_size: Maximum size per file in bytes (default 1MB)
|
||||
max_files: Maximum number of files allowed (default 100)
|
||||
"""
|
||||
self.user_id = user_id or "default"
|
||||
|
||||
if base_path:
|
||||
self.base_path = Path(base_path)
|
||||
else:
|
||||
self.base_path = Path(f".praison/claude_memory/{self.user_id}")
|
||||
|
||||
self.memory_root = self.base_path / "memories"
|
||||
self.max_file_size = max_file_size
|
||||
self.max_files = max_files
|
||||
|
||||
# Create memory directory
|
||||
self.memory_root.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
logger.debug(f"ClaudeMemoryTool initialized at {self.memory_root}")
|
||||
|
||||
def _validate_path(self, path: str) -> Path:
|
||||
"""
|
||||
Validate and resolve memory paths securely.
|
||||
|
||||
Args:
|
||||
path: Path string starting with /memories
|
||||
|
||||
Returns:
|
||||
Resolved Path object within memory directory
|
||||
|
||||
Raises:
|
||||
ValueError: If path is invalid or attempts directory traversal
|
||||
"""
|
||||
if not path:
|
||||
raise ValueError("Path cannot be empty")
|
||||
|
||||
if not path.startswith("/memories"):
|
||||
raise ValueError(f"Path must start with /memories, got: {path}")
|
||||
|
||||
# Remove /memories prefix and leading slashes
|
||||
relative_path = path[len("/memories"):].lstrip("/")
|
||||
|
||||
# Resolve the full path
|
||||
if relative_path:
|
||||
full_path = self.memory_root / relative_path
|
||||
else:
|
||||
full_path = self.memory_root
|
||||
|
||||
# Security check: ensure resolved path is within memory_root
|
||||
try:
|
||||
resolved = full_path.resolve()
|
||||
resolved.relative_to(self.memory_root.resolve())
|
||||
except ValueError as e:
|
||||
raise ValueError(f"Path {path} would escape /memories directory") from e
|
||||
|
||||
# Check for suspicious patterns
|
||||
if ".." in path or "~" in path:
|
||||
raise ValueError(f"Invalid path pattern in: {path}")
|
||||
|
||||
return full_path
|
||||
|
||||
def execute(self, command: str, **kwargs) -> str:
|
||||
"""
|
||||
Execute a memory command.
|
||||
|
||||
Args:
|
||||
command: Command name (view, create, str_replace, insert, delete, rename)
|
||||
**kwargs: Command-specific arguments
|
||||
|
||||
Returns:
|
||||
Result string from the command
|
||||
|
||||
Raises:
|
||||
ValueError: If command is unknown
|
||||
"""
|
||||
commands = {
|
||||
"view": self.view,
|
||||
"create": self.create,
|
||||
"str_replace": self.str_replace,
|
||||
"insert": self.insert,
|
||||
"delete": self.delete,
|
||||
"rename": self.rename
|
||||
}
|
||||
|
||||
if command not in commands:
|
||||
raise ValueError(f"Unknown command: {command}. Valid commands: {list(commands.keys())}")
|
||||
|
||||
return commands[command](**kwargs)
|
||||
|
||||
def view(self, path: str, view_range: Optional[List[int]] = None) -> str:
|
||||
"""
|
||||
Show directory contents or file contents with optional line ranges.
|
||||
|
||||
Args:
|
||||
path: Path to view (e.g., "/memories" or "/memories/notes.txt")
|
||||
view_range: Optional [start_line, end_line] for file viewing.
|
||||
Use -1 for end_line to read to end of file.
|
||||
|
||||
Returns:
|
||||
Directory listing or file contents with line numbers
|
||||
|
||||
Raises:
|
||||
RuntimeError: If path cannot be read
|
||||
"""
|
||||
full_path = self._validate_path(path)
|
||||
|
||||
if full_path.is_dir():
|
||||
# List directory contents
|
||||
items: List[str] = []
|
||||
try:
|
||||
for item in sorted(full_path.iterdir()):
|
||||
# Skip hidden files
|
||||
if item.name.startswith("."):
|
||||
continue
|
||||
# Add trailing slash for directories
|
||||
items.append(f"{item.name}/" if item.is_dir() else item.name)
|
||||
|
||||
result = f"Directory: {path}\n"
|
||||
if items:
|
||||
result += "\n".join([f"- {item}" for item in items])
|
||||
else:
|
||||
result += "(empty directory)"
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Cannot read directory {path}: {e}") from e
|
||||
|
||||
elif full_path.is_file():
|
||||
# Read file contents
|
||||
try:
|
||||
content = full_path.read_text(encoding="utf-8")
|
||||
lines = content.splitlines()
|
||||
|
||||
# Apply view range if specified
|
||||
if view_range:
|
||||
start_line = max(1, view_range[0]) - 1 # Convert to 0-indexed
|
||||
end_line = len(lines) if view_range[1] == -1 else view_range[1]
|
||||
lines = lines[start_line:end_line]
|
||||
start_num = start_line + 1
|
||||
else:
|
||||
start_num = 1
|
||||
|
||||
# Add line numbers
|
||||
numbered_lines = [
|
||||
f"{i + start_num:4d}: {line}"
|
||||
for i, line in enumerate(lines)
|
||||
]
|
||||
return "\n".join(numbered_lines) if numbered_lines else "(empty file)"
|
||||
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Cannot read file {path}: {e}") from e
|
||||
|
||||
else:
|
||||
raise RuntimeError(f"Path not found: {path}")
|
||||
|
||||
def create(self, path: str, file_text: str) -> str:
|
||||
"""
|
||||
Create or overwrite a file.
|
||||
|
||||
Args:
|
||||
path: Path for the new file (e.g., "/memories/notes.txt")
|
||||
file_text: Content to write to the file
|
||||
|
||||
Returns:
|
||||
Success message
|
||||
|
||||
Raises:
|
||||
ValueError: If file would exceed size limit or max files reached
|
||||
"""
|
||||
full_path = self._validate_path(path)
|
||||
|
||||
# Check file size limit
|
||||
if len(file_text.encode('utf-8')) > self.max_file_size:
|
||||
raise ValueError(
|
||||
f"File content exceeds maximum size of {self.max_file_size} bytes"
|
||||
)
|
||||
|
||||
# Check max files limit (only for new files)
|
||||
if not full_path.exists():
|
||||
current_files = sum(1 for _ in self.memory_root.rglob("*") if _.is_file())
|
||||
if current_files >= self.max_files:
|
||||
raise ValueError(
|
||||
f"Maximum number of files ({self.max_files}) reached. "
|
||||
"Delete some files before creating new ones."
|
||||
)
|
||||
|
||||
# Create parent directories if needed
|
||||
full_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Write file
|
||||
full_path.write_text(file_text, encoding="utf-8")
|
||||
|
||||
return f"File created successfully at {path}"
|
||||
|
||||
def str_replace(self, path: str, old_str: str, new_str: str) -> str:
|
||||
"""
|
||||
Replace text in a file.
|
||||
|
||||
Args:
|
||||
path: Path to the file
|
||||
old_str: Text to find (must be unique in file)
|
||||
new_str: Replacement text
|
||||
|
||||
Returns:
|
||||
Success message
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: If file doesn't exist
|
||||
ValueError: If old_str not found or appears multiple times
|
||||
"""
|
||||
full_path = self._validate_path(path)
|
||||
|
||||
if not full_path.is_file():
|
||||
raise FileNotFoundError(f"File not found: {path}")
|
||||
|
||||
content = full_path.read_text(encoding="utf-8")
|
||||
|
||||
# Count occurrences
|
||||
count = content.count(old_str)
|
||||
|
||||
if count == 0:
|
||||
raise ValueError(f"Text not found in {path}")
|
||||
elif count > 1:
|
||||
raise ValueError(
|
||||
f"Text appears {count} times in {path}. Must be unique for replacement."
|
||||
)
|
||||
|
||||
# Replace and write
|
||||
new_content = content.replace(old_str, new_str)
|
||||
|
||||
# Check size limit
|
||||
if len(new_content.encode('utf-8')) > self.max_file_size:
|
||||
raise ValueError(
|
||||
f"Resulting file would exceed maximum size of {self.max_file_size} bytes"
|
||||
)
|
||||
|
||||
full_path.write_text(new_content, encoding="utf-8")
|
||||
|
||||
return f"File {path} has been edited"
|
||||
|
||||
def insert(self, path: str, insert_line: int, insert_text: str) -> str:
|
||||
"""
|
||||
Insert text at a specific line.
|
||||
|
||||
Args:
|
||||
path: Path to the file
|
||||
insert_line: Line number to insert at (0-indexed)
|
||||
insert_text: Text to insert
|
||||
|
||||
Returns:
|
||||
Success message
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: If file doesn't exist
|
||||
ValueError: If line number is invalid
|
||||
"""
|
||||
full_path = self._validate_path(path)
|
||||
|
||||
if not full_path.is_file():
|
||||
raise FileNotFoundError(f"File not found: {path}")
|
||||
|
||||
lines = full_path.read_text(encoding="utf-8").splitlines()
|
||||
|
||||
# Validate line number
|
||||
if insert_line < 0 or insert_line > len(lines):
|
||||
raise ValueError(
|
||||
f"Invalid insert_line {insert_line}. Must be 0-{len(lines)}"
|
||||
)
|
||||
|
||||
# Insert the text (strip trailing newline to avoid double newlines)
|
||||
lines.insert(insert_line, insert_text.rstrip("\n"))
|
||||
|
||||
# Reconstruct content
|
||||
new_content = "\n".join(lines) + "\n"
|
||||
|
||||
# Check size limit
|
||||
if len(new_content.encode('utf-8')) > self.max_file_size:
|
||||
raise ValueError(
|
||||
f"Resulting file would exceed maximum size of {self.max_file_size} bytes"
|
||||
)
|
||||
|
||||
full_path.write_text(new_content, encoding="utf-8")
|
||||
|
||||
return f"Text inserted at line {insert_line} in {path}"
|
||||
|
||||
def delete(self, path: str) -> str:
|
||||
"""
|
||||
Delete a file or directory.
|
||||
|
||||
Args:
|
||||
path: Path to delete
|
||||
|
||||
Returns:
|
||||
Success message
|
||||
|
||||
Raises:
|
||||
ValueError: If trying to delete /memories root
|
||||
FileNotFoundError: If path doesn't exist
|
||||
"""
|
||||
# Prevent deleting the root memories directory
|
||||
if path == "/memories" or path == "/memories/":
|
||||
raise ValueError("Cannot delete the /memories directory itself")
|
||||
|
||||
full_path = self._validate_path(path)
|
||||
|
||||
if full_path.is_file():
|
||||
full_path.unlink()
|
||||
return f"File deleted: {path}"
|
||||
elif full_path.is_dir():
|
||||
shutil.rmtree(full_path)
|
||||
return f"Directory deleted: {path}"
|
||||
else:
|
||||
raise FileNotFoundError(f"Path not found: {path}")
|
||||
|
||||
def rename(self, old_path: str, new_path: str) -> str:
|
||||
"""
|
||||
Rename or move a file/directory.
|
||||
|
||||
Args:
|
||||
old_path: Current path
|
||||
new_path: New path
|
||||
|
||||
Returns:
|
||||
Success message
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: If source doesn't exist
|
||||
ValueError: If destination already exists
|
||||
"""
|
||||
old_full_path = self._validate_path(old_path)
|
||||
new_full_path = self._validate_path(new_path)
|
||||
|
||||
if not old_full_path.exists():
|
||||
raise FileNotFoundError(f"Source path not found: {old_path}")
|
||||
|
||||
if new_full_path.exists():
|
||||
raise ValueError(f"Destination already exists: {new_path}")
|
||||
|
||||
# Create parent directories if needed
|
||||
new_full_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Rename/move
|
||||
old_full_path.rename(new_full_path)
|
||||
|
||||
return f"Renamed {old_path} to {new_path}"
|
||||
|
||||
def clear_all(self) -> str:
|
||||
"""
|
||||
Clear all memory files.
|
||||
|
||||
Returns:
|
||||
Success message
|
||||
"""
|
||||
if self.memory_root.exists():
|
||||
shutil.rmtree(self.memory_root)
|
||||
self.memory_root.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
return "All memory cleared"
|
||||
|
||||
def get_tool_definition(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Get the tool definition for API requests.
|
||||
|
||||
Returns:
|
||||
Tool definition dict for Anthropic API
|
||||
"""
|
||||
return self.TOOL_DEFINITION.copy()
|
||||
|
||||
def get_beta_header(self) -> str:
|
||||
"""
|
||||
Get the beta header value for API requests.
|
||||
|
||||
Returns:
|
||||
Beta header string
|
||||
"""
|
||||
return self.BETA_HEADER
|
||||
|
||||
def get_system_prompt(self) -> str:
|
||||
"""
|
||||
Get the memory system prompt to append to agent instructions.
|
||||
|
||||
Returns:
|
||||
Memory protocol system prompt
|
||||
"""
|
||||
return MEMORY_SYSTEM_PROMPT
|
||||
|
||||
def process_tool_call(self, tool_input: Dict[str, Any]) -> str:
|
||||
"""
|
||||
Process a memory tool call from Claude.
|
||||
|
||||
Args:
|
||||
tool_input: The input dict from Claude's tool_use block
|
||||
Contains 'command' and command-specific parameters
|
||||
|
||||
Returns:
|
||||
Result string to return to Claude
|
||||
|
||||
Example:
|
||||
```python
|
||||
# Claude sends:
|
||||
# {"command": "view", "path": "/memories"}
|
||||
result = memory_tool.process_tool_call(tool_input)
|
||||
```
|
||||
"""
|
||||
command = tool_input.get("command")
|
||||
if not command:
|
||||
return "Error: No command specified in tool input"
|
||||
|
||||
try:
|
||||
# Extract command-specific parameters
|
||||
params = {k: v for k, v in tool_input.items() if k != "command"}
|
||||
return self.execute(command, **params)
|
||||
except Exception as e:
|
||||
return f"Error: {str(e)}"
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"ClaudeMemoryTool(base_path='{self.base_path}', user_id='{self.user_id}')"
|
||||
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "praisonaiagents"
|
||||
version = "0.0.175"
|
||||
version = "0.0.178"
|
||||
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
|
||||
requires-python = ">=3.10"
|
||||
authors = [
|
||||
|
||||
@@ -0,0 +1,460 @@
|
||||
"""
|
||||
Tests for Claude Memory Tool.
|
||||
|
||||
Claude Memory Tool enables Claude to store and retrieve information across
|
||||
conversations through a memory file directory.
|
||||
|
||||
Reference: https://docs.anthropic.com/en/docs/agents-and-tools/tool-use/memory-tool
|
||||
"""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class TestClaudeMemoryToolBasics:
|
||||
"""Tests for ClaudeMemoryTool basic functionality."""
|
||||
|
||||
def test_import(self):
|
||||
"""Test that ClaudeMemoryTool can be imported."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
assert ClaudeMemoryTool is not None
|
||||
|
||||
def test_init_default(self):
|
||||
"""Test default initialization."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
assert tool.memory_root.exists()
|
||||
assert tool.user_id == "default"
|
||||
|
||||
def test_init_with_user_id(self):
|
||||
"""Test initialization with custom user_id."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir, user_id="test_user")
|
||||
assert tool.user_id == "test_user"
|
||||
|
||||
def test_get_tool_definition(self):
|
||||
"""Test tool definition for API."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
definition = tool.get_tool_definition()
|
||||
|
||||
assert definition["type"] == "memory_20250818"
|
||||
assert definition["name"] == "memory"
|
||||
|
||||
def test_get_beta_header(self):
|
||||
"""Test beta header value."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
header = tool.get_beta_header()
|
||||
|
||||
assert header == "context-management-2025-06-27"
|
||||
|
||||
|
||||
class TestClaudeMemoryToolPathValidation:
|
||||
"""Tests for path validation and security."""
|
||||
|
||||
def test_valid_path(self):
|
||||
"""Test valid path validation."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
path = tool._validate_path("/memories")
|
||||
assert path == tool.memory_root
|
||||
|
||||
def test_valid_subpath(self):
|
||||
"""Test valid subpath validation."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
path = tool._validate_path("/memories/notes.txt")
|
||||
assert path == tool.memory_root / "notes.txt"
|
||||
|
||||
def test_invalid_path_no_memories_prefix(self):
|
||||
"""Test that paths without /memories prefix are rejected."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
with pytest.raises(ValueError, match="must start with /memories"):
|
||||
tool._validate_path("/etc/passwd")
|
||||
|
||||
def test_directory_traversal_blocked(self):
|
||||
"""Test that directory traversal is blocked."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
tool._validate_path("/memories/../../../etc/passwd")
|
||||
|
||||
|
||||
class TestClaudeMemoryToolCommands:
|
||||
"""Tests for memory tool commands."""
|
||||
|
||||
def test_view_empty_directory(self):
|
||||
"""Test viewing empty directory."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
result = tool.view(path="/memories")
|
||||
|
||||
assert "Directory: /memories" in result
|
||||
assert "(empty directory)" in result
|
||||
|
||||
def test_create_and_view_file(self):
|
||||
"""Test creating and viewing a file."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
# Create file
|
||||
result = tool.create(path="/memories/notes.txt", file_text="Hello, World!")
|
||||
assert "created successfully" in result
|
||||
|
||||
# View file
|
||||
result = tool.view(path="/memories/notes.txt")
|
||||
assert "Hello, World!" in result
|
||||
|
||||
def test_view_directory_with_files(self):
|
||||
"""Test viewing directory with files."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
# Create files
|
||||
tool.create(path="/memories/file1.txt", file_text="Content 1")
|
||||
tool.create(path="/memories/file2.txt", file_text="Content 2")
|
||||
|
||||
# View directory
|
||||
result = tool.view(path="/memories")
|
||||
assert "file1.txt" in result
|
||||
assert "file2.txt" in result
|
||||
|
||||
def test_str_replace(self):
|
||||
"""Test string replacement in file."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
# Create file
|
||||
tool.create(path="/memories/notes.txt", file_text="Hello, World!")
|
||||
|
||||
# Replace text
|
||||
result = tool.str_replace(
|
||||
path="/memories/notes.txt",
|
||||
old_str="World",
|
||||
new_str="Claude"
|
||||
)
|
||||
assert "edited" in result
|
||||
|
||||
# Verify replacement
|
||||
result = tool.view(path="/memories/notes.txt")
|
||||
assert "Hello, Claude!" in result
|
||||
|
||||
def test_str_replace_not_found(self):
|
||||
"""Test string replacement when text not found."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
tool.create(path="/memories/notes.txt", file_text="Hello, World!")
|
||||
|
||||
with pytest.raises(ValueError, match="not found"):
|
||||
tool.str_replace(
|
||||
path="/memories/notes.txt",
|
||||
old_str="Nonexistent",
|
||||
new_str="Replacement"
|
||||
)
|
||||
|
||||
def test_insert(self):
|
||||
"""Test inserting text at line."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
# Create file with multiple lines
|
||||
tool.create(path="/memories/notes.txt", file_text="Line 1\nLine 2\nLine 3")
|
||||
|
||||
# Insert at line 1
|
||||
result = tool.insert(
|
||||
path="/memories/notes.txt",
|
||||
insert_line=1,
|
||||
insert_text="Inserted Line"
|
||||
)
|
||||
assert "inserted" in result
|
||||
|
||||
# Verify insertion
|
||||
result = tool.view(path="/memories/notes.txt")
|
||||
assert "Inserted Line" in result
|
||||
|
||||
def test_delete_file(self):
|
||||
"""Test deleting a file."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
# Create and delete file
|
||||
tool.create(path="/memories/notes.txt", file_text="Content")
|
||||
result = tool.delete(path="/memories/notes.txt")
|
||||
|
||||
assert "deleted" in result
|
||||
|
||||
# Verify deletion
|
||||
with pytest.raises(RuntimeError):
|
||||
tool.view(path="/memories/notes.txt")
|
||||
|
||||
def test_delete_root_blocked(self):
|
||||
"""Test that deleting /memories root is blocked."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
with pytest.raises(ValueError, match="Cannot delete"):
|
||||
tool.delete(path="/memories")
|
||||
|
||||
def test_rename(self):
|
||||
"""Test renaming a file."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
# Create file
|
||||
tool.create(path="/memories/old.txt", file_text="Content")
|
||||
|
||||
# Rename
|
||||
result = tool.rename(
|
||||
old_path="/memories/old.txt",
|
||||
new_path="/memories/new.txt"
|
||||
)
|
||||
assert "Renamed" in result
|
||||
|
||||
# Verify rename
|
||||
with pytest.raises(RuntimeError):
|
||||
tool.view(path="/memories/old.txt")
|
||||
|
||||
result = tool.view(path="/memories/new.txt")
|
||||
assert "Content" in result
|
||||
|
||||
def test_clear_all(self):
|
||||
"""Test clearing all memory."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
# Create files
|
||||
tool.create(path="/memories/file1.txt", file_text="Content 1")
|
||||
tool.create(path="/memories/file2.txt", file_text="Content 2")
|
||||
|
||||
# Clear all
|
||||
result = tool.clear_all()
|
||||
assert "cleared" in result
|
||||
|
||||
# Verify empty
|
||||
result = tool.view(path="/memories")
|
||||
assert "(empty directory)" in result
|
||||
|
||||
|
||||
class TestClaudeMemoryToolProcessToolCall:
|
||||
"""Tests for process_tool_call method."""
|
||||
|
||||
def test_process_view_command(self):
|
||||
"""Test processing view command."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
result = tool.process_tool_call({
|
||||
"command": "view",
|
||||
"path": "/memories"
|
||||
})
|
||||
|
||||
assert "Directory: /memories" in result
|
||||
|
||||
def test_process_create_command(self):
|
||||
"""Test processing create command."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
result = tool.process_tool_call({
|
||||
"command": "create",
|
||||
"path": "/memories/test.txt",
|
||||
"file_text": "Test content"
|
||||
})
|
||||
|
||||
assert "created" in result
|
||||
|
||||
def test_process_unknown_command(self):
|
||||
"""Test processing unknown command."""
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tool = ClaudeMemoryTool(base_path=tmpdir)
|
||||
|
||||
result = tool.process_tool_call({
|
||||
"command": "unknown_command",
|
||||
"path": "/memories"
|
||||
})
|
||||
|
||||
assert "Error" in result
|
||||
|
||||
|
||||
class TestLLMClaudeMemoryIntegration:
|
||||
"""Tests for LLM class integration with Claude Memory Tool."""
|
||||
|
||||
def test_llm_init_with_claude_memory_true(self):
|
||||
"""Test LLM initialization with claude_memory=True."""
|
||||
from praisonaiagents.llm.llm import LLM
|
||||
|
||||
llm = LLM(model="anthropic/claude-sonnet-4-20250514", claude_memory=True)
|
||||
assert llm.claude_memory == True
|
||||
|
||||
def test_llm_init_with_claude_memory_false(self):
|
||||
"""Test LLM initialization with claude_memory=False."""
|
||||
from praisonaiagents.llm.llm import LLM
|
||||
|
||||
llm = LLM(model="anthropic/claude-sonnet-4-20250514", claude_memory=False)
|
||||
assert llm.claude_memory == False
|
||||
|
||||
def test_llm_init_without_claude_memory(self):
|
||||
"""Test LLM initialization without claude_memory parameter."""
|
||||
from praisonaiagents.llm.llm import LLM
|
||||
|
||||
llm = LLM(model="gpt-4o-mini")
|
||||
assert llm.claude_memory is None
|
||||
|
||||
def test_llm_supports_claude_memory_anthropic(self):
|
||||
"""Test _supports_claude_memory for Anthropic model."""
|
||||
from praisonaiagents.llm.llm import LLM
|
||||
|
||||
llm = LLM(model="anthropic/claude-sonnet-4-20250514", claude_memory=True)
|
||||
assert llm._supports_claude_memory() == True
|
||||
|
||||
def test_llm_supports_claude_memory_non_anthropic(self):
|
||||
"""Test _supports_claude_memory for non-Anthropic model."""
|
||||
from praisonaiagents.llm.llm import LLM
|
||||
|
||||
llm = LLM(model="gpt-4o-mini", claude_memory=True)
|
||||
assert llm._supports_claude_memory() == False
|
||||
|
||||
def test_llm_get_claude_memory_tool(self):
|
||||
"""Test _get_claude_memory_tool method."""
|
||||
from praisonaiagents.llm.llm import LLM
|
||||
from praisonaiagents.tools.claude_memory_tool import ClaudeMemoryTool
|
||||
|
||||
llm = LLM(model="anthropic/claude-sonnet-4-20250514", claude_memory=True)
|
||||
tool = llm._get_claude_memory_tool()
|
||||
|
||||
assert isinstance(tool, ClaudeMemoryTool)
|
||||
|
||||
def test_llm_is_memory_tool_call(self):
|
||||
"""Test _is_memory_tool_call method."""
|
||||
from praisonaiagents.llm.llm import LLM
|
||||
|
||||
llm = LLM(model="anthropic/claude-sonnet-4-20250514", claude_memory=True)
|
||||
|
||||
assert llm._is_memory_tool_call("memory") == True
|
||||
assert llm._is_memory_tool_call("other_tool") == False
|
||||
|
||||
|
||||
class TestAgentClaudeMemoryIntegration:
|
||||
"""Tests for Agent class integration with Claude Memory Tool."""
|
||||
|
||||
def test_agent_init_with_claude_memory_true(self):
|
||||
"""Test Agent initialization with claude_memory=True."""
|
||||
from praisonaiagents import Agent
|
||||
|
||||
agent = Agent(
|
||||
name="Test Agent",
|
||||
llm="anthropic/claude-sonnet-4-20250514",
|
||||
claude_memory=True
|
||||
)
|
||||
assert agent.claude_memory == True
|
||||
|
||||
def test_agent_init_with_claude_memory_false(self):
|
||||
"""Test Agent initialization with claude_memory=False."""
|
||||
from praisonaiagents import Agent
|
||||
|
||||
agent = Agent(
|
||||
name="Test Agent",
|
||||
llm="anthropic/claude-sonnet-4-20250514",
|
||||
claude_memory=False
|
||||
)
|
||||
assert agent.claude_memory == False
|
||||
|
||||
def test_agent_init_without_claude_memory(self):
|
||||
"""Test Agent initialization without claude_memory parameter."""
|
||||
from praisonaiagents import Agent
|
||||
|
||||
agent = Agent(name="Test Agent")
|
||||
assert agent.claude_memory is None
|
||||
|
||||
def test_agent_passes_claude_memory_to_llm(self):
|
||||
"""Test that Agent passes claude_memory to LLM instance."""
|
||||
from praisonaiagents import Agent
|
||||
|
||||
agent = Agent(
|
||||
name="Test Agent",
|
||||
llm="anthropic/claude-sonnet-4-20250514",
|
||||
claude_memory=True
|
||||
)
|
||||
|
||||
if hasattr(agent, 'llm_instance') and agent.llm_instance:
|
||||
assert agent.llm_instance.claude_memory == True
|
||||
|
||||
|
||||
class TestClaudeMemoryToolIntegration:
|
||||
"""Integration tests for Claude Memory Tool (requires API key)."""
|
||||
|
||||
@pytest.mark.skip(reason="Requires ANTHROPIC_API_KEY environment variable")
|
||||
def test_claude_memory_with_anthropic(self):
|
||||
"""Test Claude memory with actual Anthropic API."""
|
||||
import os
|
||||
from praisonaiagents import Agent
|
||||
|
||||
# Skip if no API key
|
||||
if not os.getenv("ANTHROPIC_API_KEY"):
|
||||
pytest.skip("ANTHROPIC_API_KEY not set")
|
||||
|
||||
agent = Agent(
|
||||
name="Memory Test Agent",
|
||||
instructions="You are a helpful assistant with memory capabilities.",
|
||||
llm="anthropic/claude-sonnet-4-20250514",
|
||||
claude_memory=True,
|
||||
verbose=False
|
||||
)
|
||||
|
||||
# First call - should check memory
|
||||
result1 = agent.start("Remember that my favorite color is blue.")
|
||||
assert result1 is not None
|
||||
|
||||
# Second call - should recall from memory
|
||||
result2 = agent.start("What is my favorite color?")
|
||||
assert result2 is not None
|
||||
@@ -0,0 +1,653 @@
|
||||
"""
|
||||
Tests for FileMemory - Zero-dependency file-based memory system.
|
||||
"""
|
||||
|
||||
import tempfile
|
||||
import pytest
|
||||
|
||||
|
||||
class TestFileMemoryBasics:
|
||||
"""Tests for FileMemory basic functionality."""
|
||||
|
||||
def test_import(self):
|
||||
"""Test that FileMemory can be imported."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
assert FileMemory is not None
|
||||
|
||||
def test_init_default(self):
|
||||
"""Test default initialization."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
assert memory.user_id == "default"
|
||||
assert memory.user_path.exists()
|
||||
|
||||
def test_init_with_user_id(self):
|
||||
"""Test initialization with custom user_id."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir, user_id="test_user")
|
||||
assert memory.user_id == "test_user"
|
||||
assert "test_user" in str(memory.user_path)
|
||||
|
||||
def test_config_persistence(self):
|
||||
"""Test that config is saved and loaded."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
# Create with custom config
|
||||
memory1 = FileMemory(
|
||||
base_path=tmpdir,
|
||||
config={"short_term_limit": 50}
|
||||
)
|
||||
assert memory1.config["short_term_limit"] == 50
|
||||
|
||||
# Create new instance - should load saved config
|
||||
memory2 = FileMemory(base_path=tmpdir)
|
||||
assert memory2.config["short_term_limit"] == 50
|
||||
|
||||
|
||||
class TestShortTermMemory:
|
||||
"""Tests for short-term memory functionality."""
|
||||
|
||||
def test_add_short_term(self):
|
||||
"""Test adding short-term memory."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
mem_id = memory.add_short_term("User prefers dark mode")
|
||||
assert mem_id is not None
|
||||
assert len(memory._short_term) == 1
|
||||
|
||||
def test_get_short_term(self):
|
||||
"""Test retrieving short-term memories."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_short_term("Memory 1")
|
||||
memory.add_short_term("Memory 2")
|
||||
memory.add_short_term("Memory 3")
|
||||
|
||||
# Get all
|
||||
items = memory.get_short_term()
|
||||
assert len(items) == 3
|
||||
# Most recent first
|
||||
assert items[0].content == "Memory 3"
|
||||
|
||||
# Get limited
|
||||
items = memory.get_short_term(limit=2)
|
||||
assert len(items) == 2
|
||||
|
||||
def test_short_term_rolling_buffer(self):
|
||||
"""Test that short-term memory enforces limit."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(
|
||||
base_path=tmpdir,
|
||||
config={"short_term_limit": 5, "auto_promote": False}
|
||||
)
|
||||
|
||||
# Add more than limit
|
||||
for i in range(10):
|
||||
memory.add_short_term(f"Memory {i}")
|
||||
|
||||
# Should only have last 5
|
||||
assert len(memory._short_term) == 5
|
||||
items = memory.get_short_term()
|
||||
assert items[0].content == "Memory 9"
|
||||
|
||||
def test_short_term_persistence(self):
|
||||
"""Test that short-term memory persists to file."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
# Add memory
|
||||
memory1 = FileMemory(base_path=tmpdir)
|
||||
memory1.add_short_term("Persistent memory")
|
||||
|
||||
# Create new instance - should load from file
|
||||
memory2 = FileMemory(base_path=tmpdir)
|
||||
items = memory2.get_short_term()
|
||||
assert len(items) == 1
|
||||
assert items[0].content == "Persistent memory"
|
||||
|
||||
|
||||
class TestLongTermMemory:
|
||||
"""Tests for long-term memory functionality."""
|
||||
|
||||
def test_add_long_term(self):
|
||||
"""Test adding long-term memory."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
mem_id = memory.add_long_term("User's name is John", importance=0.9)
|
||||
assert mem_id is not None
|
||||
assert len(memory._long_term) == 1
|
||||
|
||||
def test_get_long_term_sorted_by_importance(self):
|
||||
"""Test that long-term memories are sorted by importance."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_long_term("Low importance", importance=0.3)
|
||||
memory.add_long_term("High importance", importance=0.9)
|
||||
memory.add_long_term("Medium importance", importance=0.6)
|
||||
|
||||
items = memory.get_long_term()
|
||||
assert items[0].content == "High importance"
|
||||
assert items[1].content == "Medium importance"
|
||||
assert items[2].content == "Low importance"
|
||||
|
||||
def test_auto_promote_to_long_term(self):
|
||||
"""Test auto-promotion of high-importance short-term to long-term."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(
|
||||
base_path=tmpdir,
|
||||
config={
|
||||
"short_term_limit": 3,
|
||||
"auto_promote": True,
|
||||
"importance_threshold": 0.7
|
||||
}
|
||||
)
|
||||
|
||||
# Add high importance items
|
||||
memory.add_short_term("Important 1", importance=0.8)
|
||||
memory.add_short_term("Important 2", importance=0.9)
|
||||
memory.add_short_term("Low importance", importance=0.3)
|
||||
|
||||
# Trigger promotion by exceeding limit
|
||||
memory.add_short_term("New item", importance=0.5)
|
||||
|
||||
# High importance items should be in long-term
|
||||
long_term = memory.get_long_term()
|
||||
assert len(long_term) >= 2
|
||||
|
||||
|
||||
class TestEntityMemory:
|
||||
"""Tests for entity memory functionality."""
|
||||
|
||||
def test_add_entity(self):
|
||||
"""Test adding an entity."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
entity_id = memory.add_entity(
|
||||
name="John",
|
||||
entity_type="person",
|
||||
attributes={"role": "developer", "company": "Acme"}
|
||||
)
|
||||
|
||||
assert entity_id is not None
|
||||
assert len(memory._entities) == 1
|
||||
|
||||
def test_get_entity(self):
|
||||
"""Test retrieving an entity."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_entity("John", "person", {"role": "developer"})
|
||||
|
||||
entity = memory.get_entity("John")
|
||||
assert entity is not None
|
||||
assert entity.name == "John"
|
||||
assert entity.entity_type == "person"
|
||||
assert entity.attributes["role"] == "developer"
|
||||
|
||||
def test_update_entity(self):
|
||||
"""Test updating an existing entity."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_entity("John", "person", {"role": "developer"})
|
||||
memory.add_entity("John", "person", {"company": "Acme"})
|
||||
|
||||
entity = memory.get_entity("John")
|
||||
assert entity.attributes["role"] == "developer"
|
||||
assert entity.attributes["company"] == "Acme"
|
||||
|
||||
def test_get_all_entities(self):
|
||||
"""Test getting all entities."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_entity("John", "person", {})
|
||||
memory.add_entity("Acme", "organization", {})
|
||||
memory.add_entity("New York", "place", {})
|
||||
|
||||
# All entities
|
||||
all_entities = memory.get_all_entities()
|
||||
assert len(all_entities) == 3
|
||||
|
||||
# Filter by type
|
||||
people = memory.get_all_entities(entity_type="person")
|
||||
assert len(people) == 1
|
||||
assert people[0].name == "John"
|
||||
|
||||
|
||||
class TestEpisodicMemory:
|
||||
"""Tests for episodic memory functionality."""
|
||||
|
||||
def test_add_episodic(self):
|
||||
"""Test adding episodic memory."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
mem_id = memory.add_episodic("Had a meeting about project X")
|
||||
assert mem_id is not None
|
||||
|
||||
def test_get_episodic_by_date(self):
|
||||
"""Test retrieving episodic memories by date."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
from datetime import datetime
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
today = datetime.now().strftime("%Y-%m-%d")
|
||||
memory.add_episodic("Today's event", date=today)
|
||||
|
||||
episodes = memory.get_episodic(date=today)
|
||||
assert len(episodes) == 1
|
||||
assert episodes[0].content == "Today's event"
|
||||
|
||||
def test_get_recent_episodic(self):
|
||||
"""Test retrieving recent episodic memories."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_episodic("Event 1")
|
||||
memory.add_episodic("Event 2")
|
||||
|
||||
episodes = memory.get_episodic(days_back=7)
|
||||
assert len(episodes) == 2
|
||||
|
||||
|
||||
class TestSearch:
|
||||
"""Tests for search functionality."""
|
||||
|
||||
def test_search_short_term(self):
|
||||
"""Test searching short-term memory."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_short_term("User prefers dark mode")
|
||||
memory.add_short_term("User likes Python programming")
|
||||
memory.add_short_term("Weather is sunny today")
|
||||
|
||||
results = memory.search("user", memory_types=["short_term"])
|
||||
assert len(results) == 2
|
||||
|
||||
def test_search_all_types(self):
|
||||
"""Test searching across all memory types."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_short_term("John mentioned Python")
|
||||
memory.add_long_term("John is a developer")
|
||||
memory.add_entity("John", "person", {"skill": "Python"})
|
||||
|
||||
results = memory.search("John")
|
||||
assert len(results) >= 3
|
||||
|
||||
def test_search_relevance_scoring(self):
|
||||
"""Test that search results are scored by relevance."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_short_term("Python is great", importance=0.5)
|
||||
memory.add_short_term("I love Python programming", importance=0.8)
|
||||
|
||||
results = memory.search("Python")
|
||||
assert len(results) == 2
|
||||
# Higher importance should score higher
|
||||
assert results[0]["score"] >= results[1]["score"]
|
||||
|
||||
|
||||
class TestContext:
|
||||
"""Tests for context building."""
|
||||
|
||||
def test_get_context_empty(self):
|
||||
"""Test getting context with no memories."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
context = memory.get_context()
|
||||
assert context == ""
|
||||
|
||||
def test_get_context_with_memories(self):
|
||||
"""Test getting context with memories."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_long_term("User's name is John", importance=0.9)
|
||||
memory.add_short_term("User asked about Python")
|
||||
memory.add_entity("John", "person", {"role": "developer"})
|
||||
|
||||
context = memory.get_context()
|
||||
assert "John" in context
|
||||
assert "Important Facts" in context or "Known Entities" in context
|
||||
|
||||
def test_get_context_with_query(self):
|
||||
"""Test getting focused context with query."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_long_term("User's name is John")
|
||||
memory.add_long_term("User likes Python")
|
||||
memory.add_long_term("Weather is sunny")
|
||||
|
||||
context = memory.get_context(query="John")
|
||||
assert "John" in context
|
||||
|
||||
|
||||
class TestUtilities:
|
||||
"""Tests for utility methods."""
|
||||
|
||||
def test_clear_short_term(self):
|
||||
"""Test clearing short-term memory."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_short_term("Memory 1")
|
||||
memory.add_short_term("Memory 2")
|
||||
|
||||
memory.clear_short_term()
|
||||
|
||||
assert len(memory._short_term) == 0
|
||||
|
||||
def test_clear_all(self):
|
||||
"""Test clearing all memory."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir)
|
||||
|
||||
memory.add_short_term("Short term")
|
||||
memory.add_long_term("Long term")
|
||||
memory.add_entity("John", "person", {})
|
||||
memory.add_episodic("Episode")
|
||||
|
||||
memory.clear_all()
|
||||
|
||||
assert len(memory._short_term) == 0
|
||||
assert len(memory._long_term) == 0
|
||||
assert len(memory._entities) == 0
|
||||
|
||||
def test_get_stats(self):
|
||||
"""Test getting memory statistics."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = FileMemory(base_path=tmpdir, user_id="test_user")
|
||||
|
||||
memory.add_short_term("Short term")
|
||||
memory.add_long_term("Long term")
|
||||
memory.add_entity("John", "person", {})
|
||||
|
||||
stats = memory.get_stats()
|
||||
|
||||
assert stats["user_id"] == "test_user"
|
||||
assert stats["short_term_count"] == 1
|
||||
assert stats["long_term_count"] == 1
|
||||
assert stats["entity_count"] == 1
|
||||
|
||||
def test_export_import(self):
|
||||
"""Test exporting and importing memory data."""
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
# Create and populate memory
|
||||
memory1 = FileMemory(base_path=tmpdir, user_id="user1")
|
||||
memory1.add_short_term("Short term memory")
|
||||
memory1.add_long_term("Long term memory")
|
||||
memory1.add_entity("John", "person", {"role": "dev"})
|
||||
|
||||
# Export
|
||||
exported = memory1.export()
|
||||
|
||||
# Import to new memory
|
||||
memory2 = FileMemory(base_path=tmpdir, user_id="user2")
|
||||
memory2.import_data(exported)
|
||||
|
||||
assert len(memory2._short_term) == 1
|
||||
assert len(memory2._long_term) == 1
|
||||
assert len(memory2._entities) == 1
|
||||
|
||||
|
||||
class TestAgentIntegration:
|
||||
"""Tests for Agent class integration with FileMemory."""
|
||||
|
||||
def test_agent_memory_true(self):
|
||||
"""Test Agent with memory=True uses FileMemory."""
|
||||
from praisonaiagents import Agent
|
||||
|
||||
agent = Agent(
|
||||
name="Test Agent",
|
||||
memory=True
|
||||
)
|
||||
|
||||
assert agent.memory is True
|
||||
assert agent._memory_instance is not None
|
||||
|
||||
def test_agent_memory_false(self):
|
||||
"""Test Agent with memory=False has no memory."""
|
||||
from praisonaiagents import Agent
|
||||
|
||||
agent = Agent(
|
||||
name="Test Agent",
|
||||
memory=False
|
||||
)
|
||||
|
||||
assert agent._memory_instance is None
|
||||
|
||||
def test_agent_memory_file(self):
|
||||
"""Test Agent with memory='file' uses FileMemory."""
|
||||
from praisonaiagents import Agent
|
||||
from praisonaiagents.memory.file_memory import FileMemory
|
||||
|
||||
agent = Agent(
|
||||
name="Test Agent",
|
||||
memory="file"
|
||||
)
|
||||
|
||||
assert isinstance(agent._memory_instance, FileMemory)
|
||||
|
||||
def test_agent_store_memory(self):
|
||||
"""Test Agent store_memory method."""
|
||||
from praisonaiagents import Agent
|
||||
|
||||
agent = Agent(
|
||||
name="Test Agent",
|
||||
memory=True
|
||||
)
|
||||
|
||||
agent.store_memory("User prefers dark mode", memory_type="short_term")
|
||||
|
||||
items = agent._memory_instance.get_short_term()
|
||||
assert len(items) == 1
|
||||
assert items[0].content == "User prefers dark mode"
|
||||
|
||||
def test_agent_get_memory_context(self):
|
||||
"""Test Agent get_memory_context method."""
|
||||
from praisonaiagents import Agent
|
||||
|
||||
agent = Agent(
|
||||
name="Test Agent",
|
||||
memory=True
|
||||
)
|
||||
|
||||
agent.store_memory("User's name is John", memory_type="long_term", importance=0.9)
|
||||
|
||||
context = agent.get_memory_context()
|
||||
assert "John" in context
|
||||
|
||||
|
||||
class TestCreateMemoryFunction:
|
||||
"""Tests for create_memory convenience function."""
|
||||
|
||||
def test_create_memory(self):
|
||||
"""Test create_memory function."""
|
||||
from praisonaiagents.memory import create_memory
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
memory = create_memory(user_id="test", base_path=tmpdir)
|
||||
|
||||
assert memory.user_id == "test"
|
||||
memory.add_short_term("Test memory")
|
||||
assert len(memory._short_term) == 1
|
||||
|
||||
|
||||
class TestMemoryIntegrationWithOpenAI:
|
||||
"""Integration tests for memory with OpenAI API (requires OPENAI_API_KEY)."""
|
||||
|
||||
def test_memory_injection_into_system_prompt(self):
|
||||
"""Test that memory context is injected into system prompt and agent recalls it."""
|
||||
import os
|
||||
import shutil
|
||||
|
||||
# Skip if no API key
|
||||
if not os.getenv("OPENAI_API_KEY"):
|
||||
pytest.skip("OPENAI_API_KEY not set")
|
||||
|
||||
from praisonaiagents import Agent
|
||||
|
||||
test_user_id = "integration_test_user"
|
||||
|
||||
try:
|
||||
# Create agent with memory enabled
|
||||
agent = Agent(
|
||||
name="Memory Test Agent",
|
||||
role="Assistant",
|
||||
goal="Help users and remember their information",
|
||||
backstory="You are a helpful assistant with memory capabilities.",
|
||||
# Uses default model (gpt-4o-mini via OPENAI_MODEL_NAME or fallback)
|
||||
memory=True,
|
||||
user_id=test_user_id,
|
||||
verbose=False
|
||||
)
|
||||
|
||||
# Store memories BEFORE the chat
|
||||
agent.store_memory("User's favorite color is blue", memory_type="short_term")
|
||||
agent.store_memory("User's name is Alice", memory_type="long_term", importance=0.95)
|
||||
agent.store_memory("User works as a software engineer", memory_type="long_term", importance=0.9)
|
||||
|
||||
# Add entity
|
||||
agent._memory_instance.add_entity("Alice", "person", {
|
||||
"occupation": "software engineer",
|
||||
"favorite_color": "blue"
|
||||
})
|
||||
|
||||
# Ask agent about the user - should recall from memory
|
||||
result = agent.start("What is my name and what do I do for work? Answer in one sentence.")
|
||||
|
||||
# Verify the response contains the remembered information
|
||||
assert result is not None
|
||||
result_lower = result.lower()
|
||||
assert "alice" in result_lower, f"Expected 'alice' in response: {result}"
|
||||
assert "software" in result_lower or "engineer" in result_lower, f"Expected occupation in response: {result}"
|
||||
|
||||
print(f"✅ Memory injection test passed! Response: {result}")
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
memory_path = f".praison/memory/{test_user_id}"
|
||||
if os.path.exists(memory_path):
|
||||
shutil.rmtree(memory_path)
|
||||
|
||||
def test_memory_persistence_across_agent_instances(self):
|
||||
"""Test that memory persists when creating new agent instances."""
|
||||
import os
|
||||
import shutil
|
||||
|
||||
# Skip if no API key
|
||||
if not os.getenv("OPENAI_API_KEY"):
|
||||
pytest.skip("OPENAI_API_KEY not set")
|
||||
|
||||
from praisonaiagents import Agent
|
||||
|
||||
test_user_id = "persistence_test_user"
|
||||
|
||||
try:
|
||||
# First agent - store memories
|
||||
agent1 = Agent(
|
||||
name="Agent 1",
|
||||
# Uses default model (gpt-4o-mini via OPENAI_MODEL_NAME or fallback)
|
||||
memory=True,
|
||||
user_id=test_user_id,
|
||||
verbose=False
|
||||
)
|
||||
|
||||
agent1.store_memory("User prefers Python over JavaScript", memory_type="long_term", importance=0.9)
|
||||
agent1.store_memory("User is learning machine learning", memory_type="short_term")
|
||||
|
||||
# Verify memories stored
|
||||
stats1 = agent1._memory_instance.get_stats()
|
||||
assert stats1["long_term_count"] == 1
|
||||
assert stats1["short_term_count"] == 1
|
||||
|
||||
# Second agent - same user_id, should load memories
|
||||
agent2 = Agent(
|
||||
name="Agent 2",
|
||||
# Uses default model (gpt-4o-mini via OPENAI_MODEL_NAME or fallback)
|
||||
memory=True,
|
||||
user_id=test_user_id,
|
||||
verbose=False
|
||||
)
|
||||
|
||||
# Verify memories loaded
|
||||
stats2 = agent2._memory_instance.get_stats()
|
||||
assert stats2["long_term_count"] == 1, f"Expected 1 long-term memory, got {stats2}"
|
||||
assert stats2["short_term_count"] == 1, f"Expected 1 short-term memory, got {stats2}"
|
||||
|
||||
# Ask about preferences - should recall from loaded memory
|
||||
result = agent2.start("What programming language do I prefer? Answer in one word.")
|
||||
|
||||
assert result is not None
|
||||
assert "python" in result.lower(), f"Expected 'python' in response: {result}"
|
||||
|
||||
print(f"✅ Memory persistence test passed! Response: {result}")
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
memory_path = f".praison/memory/{test_user_id}"
|
||||
if os.path.exists(memory_path):
|
||||
shutil.rmtree(memory_path)
|
||||
Generated
+1
-1
@@ -2422,7 +2422,7 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "praisonaiagents"
|
||||
version = "0.0.175"
|
||||
version = "0.0.178"
|
||||
source = { editable = "." }
|
||||
dependencies = [
|
||||
{ name = "aiohttp" },
|
||||
|
||||
@@ -58,6 +58,8 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
|
||||
- 🌐 **Native Web Search** (OpenAI, Gemini, Anthropic, xAI, Perplexity)
|
||||
- 📥 **Web Fetch** (Retrieve full content from URLs - Anthropic)
|
||||
- 💾 **Prompt Caching** (Reduce costs & latency - OpenAI, Anthropic, Bedrock, Deepseek)
|
||||
- 🧠 **Claude Memory Tool** (Persistent cross-conversation memory - Anthropic Beta)
|
||||
- 💾 **File-Based Memory** (Zero-dependency persistent memory for all agents)
|
||||
|
||||
## Using Python Code
|
||||
|
||||
@@ -174,6 +176,48 @@ result = agent.rewrite("What about cost?", chat_history=[...])
|
||||
- **CONTEXTUAL**: Resolve references using conversation history
|
||||
- **AUTO**: Automatically detect best strategy
|
||||
|
||||
### 5. Agent Memory (Zero Dependencies)
|
||||
|
||||
Enable persistent memory for agents - works out of the box without any extra packages.
|
||||
|
||||
```python
|
||||
from praisonaiagents import Agent
|
||||
|
||||
# Enable memory with a single parameter
|
||||
agent = Agent(
|
||||
name="Personal Assistant",
|
||||
instructions="You are a helpful assistant that remembers user preferences.",
|
||||
memory=True, # Enables file-based memory (no extra deps!)
|
||||
user_id="user123" # Isolate memory per user
|
||||
)
|
||||
|
||||
# Store memories programmatically
|
||||
agent.store_memory("User prefers dark mode", memory_type="short_term")
|
||||
agent.store_memory("User's name is John", memory_type="long_term", importance=0.9)
|
||||
|
||||
# Memory is automatically injected into conversations
|
||||
result = agent.start("What do you know about me?")
|
||||
# Agent will recall: "Your name is John and you prefer dark mode"
|
||||
|
||||
# Memory persists across sessions - create new agent with same user_id
|
||||
agent2 = Agent(name="Assistant", memory=True, user_id="user123")
|
||||
# agent2 automatically loads all previous memories!
|
||||
```
|
||||
|
||||
**Memory Types:**
|
||||
- **Short-term**: Rolling buffer of recent context (auto-expires)
|
||||
- **Long-term**: Persistent important facts (sorted by importance)
|
||||
- **Entity**: People, places, organizations with attributes
|
||||
- **Episodic**: Date-based interaction history
|
||||
|
||||
**Storage Options:**
|
||||
| Option | Dependencies | Description |
|
||||
|--------|-------------|-------------|
|
||||
| `memory=True` | None | File-based JSON storage (default) |
|
||||
| `memory="file"` | None | Explicit file-based storage |
|
||||
| `memory="sqlite"` | Built-in | SQLite with indexing |
|
||||
| `memory="chromadb"` | chromadb | Vector/semantic search |
|
||||
|
||||
## Using No Code
|
||||
|
||||
### Auto Mode:
|
||||
|
||||
@@ -3,8 +3,8 @@ class Praisonai < Formula
|
||||
|
||||
desc "AI tools for various AI applications"
|
||||
homepage "https://github.com/MervinPraison/PraisonAI"
|
||||
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.2.tar.gz"
|
||||
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.2.tar.gz | shasum -a 256`.split.first
|
||||
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.3.tar.gz"
|
||||
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.3.tar.gz | shasum -a 256`.split.first
|
||||
license "MIT"
|
||||
|
||||
depends_on "[email protected]"
|
||||
|
||||
@@ -57,7 +57,7 @@ class CloudDeployer:
|
||||
file.write("FROM python:3.11-slim\n")
|
||||
file.write("WORKDIR /app\n")
|
||||
file.write("COPY . .\n")
|
||||
file.write("RUN pip install flask praisonai==2.3.2 gunicorn markdown\n")
|
||||
file.write("RUN pip install flask praisonai==2.3.3 gunicorn markdown\n")
|
||||
file.write("EXPOSE 8080\n")
|
||||
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
|
||||
|
||||
|
||||
@@ -1 +1 @@
|
||||
__version__ = "2.3.2"
|
||||
__version__ = "2.3.3"
|
||||
@@ -12,7 +12,7 @@ dependencies = [
|
||||
"rich>=13.7",
|
||||
"markdown>=3.5",
|
||||
"pyparsing>=3.0.0",
|
||||
"praisonaiagents>=0.0.175",
|
||||
"praisonaiagents>=0.0.178",
|
||||
"python-dotenv>=0.19.0",
|
||||
"instructor>=1.3.3",
|
||||
"PyYAML>=6.0",
|
||||
|
||||
Generated
+4
-4
@@ -4197,7 +4197,7 @@ requires-dist = [
|
||||
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.22" },
|
||||
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.0.22" },
|
||||
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.22" },
|
||||
{ name = "praisonaiagents", specifier = ">=0.0.175" },
|
||||
{ name = "praisonaiagents", specifier = ">=0.0.178" },
|
||||
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
|
||||
{ name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
|
||||
{ name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
|
||||
@@ -4252,7 +4252,7 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "praisonaiagents"
|
||||
version = "0.0.175"
|
||||
version = "0.0.178"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "aiohttp" },
|
||||
@@ -4261,9 +4261,9 @@ dependencies = [
|
||||
{ name = "pydantic" },
|
||||
{ name = "rich" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/a7/3a/a3f78a53c5ffdab876b59c7f7ea86ed815d641dff45465c70b6fd8ab1a8f/praisonaiagents-0.0.175.tar.gz", hash = "sha256:5f2bdeae4988a34947aa231c7a3de76bf50d60d2dfd43e214769d188886d6362", size = 332558, upload-time = "2025-12-15T12:43:06.491Z" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/db/40/9c7355184b107ab09db83525b6ac8633b27c2cd12104902c7f95cdef6e61/praisonaiagents-0.0.178.tar.gz", hash = "sha256:9427be0e26c08f9a6c5e356e52991ee43a531a8088bd45ecb0a27f4e4b602eaa", size = 352905, upload-time = "2025-12-15T14:52:03.258Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/ee/c9/93cf31b2c1011ee205efc75080a4d7ab9ed68e53e033d51a861112545428/praisonaiagents-0.0.175-py3-none-any.whl", hash = "sha256:1bdb672efefd562c47be481008551894e57083e52005227650f97999f88e7ff6", size = 339912, upload-time = "2025-12-15T12:43:05.064Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/5b/4f/46b7030a57852127a0f185b0b2a9340df277d3396dff7ad86f2f8e580122/praisonaiagents-0.0.178-py3-none-any.whl", hash = "sha256:91a032a0da8574662269b7266c41667ae905434388f1e9bc74265f18df5596a7", size = 354385, upload-time = "2025-12-15T14:52:01.915Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
Reference in new issue
Block a user