Automatic Memory Extraction Guide (v0.7)¶

extract_from_conversation(
    project_id="my-project",
    conversation=[
        {
            "role": "user",
            "content": "Should we use Redis or Memcached for caching?"
        },
        {
            "role": "assistant",
            "content": "I recommend Redis because it supports data persistence, has richer data structures, and provides better tooling. Redis also allows you to use it as both cache and message queue."
        },
        {
            "role": "user",
            "content": "Great, let's go with Redis then."
        }
    ],
    auto_save=True
)

from src.codebase_rag.services.memory import memory_extractor

result = await memory_extractor.extract_from_conversation(
    project_id="my-project",
    conversation=[
        {"role": "user", "content": "Should we use Redis or Memcached?"},
        {"role": "assistant", "content": "I recommend Redis..."}
    ],
    auto_save=True
)

print(f"Auto-saved: {result['auto_saved_count']} memories")
print(f"Suggestions: {len(result['suggestions'])} memories")

conversation = [
    {"role": "user", "content": "How should we handle user authentication?"},
    {"role": "assistant", "content": "For your use case, I'd recommend JWT tokens with refresh token rotation..."},
    {"role": "user", "content": "What about session storage?"},
    {"role": "assistant", "content": "For JWTs, you don't need server-side sessions. Store tokens in httpOnly cookies..."},
    {"role": "user", "content": "Should we use Redis for token blacklisting?"},
    {"role": "assistant", "content": "Yes, Redis is perfect for token blacklisting with TTL support..."}
]

result = await memory_extractor.extract_from_conversation(
    project_id="web-app",
    conversation=conversation,
    auto_save=False  # Review before saving
)

# Review suggestions
for suggestion in result['suggestions']:
    print(f"Type: {suggestion['type']}")
    print(f"Title: {suggestion['title']}")
    print(f"Confidence: {suggestion['confidence']}")
    print(f"Importance: {suggestion['importance']}")
    print()

# Manually save high-value suggestions
for suggestion in result['suggestions']:
    if suggestion['confidence'] >= 0.8:
        await memory_store.add_memory(
            project_id="web-app",
            **suggestion
        )

# Automatically save memories with confidence >= 0.7
result = await memory_extractor.extract_from_conversation(
    project_id="my-project",
    conversation=conversation,
    auto_save=True
)

# Only high-confidence memories are saved
print(f"Auto-saved: {result['auto_saved_count']}")

# Get suggestions, review before saving
result = await memory_extractor.extract_from_conversation(
    project_id="my-project",
    conversation=conversation,
    auto_save=False
)

# Review each suggestion
for suggestion in result['suggestions']:
    print(f"Review: {suggestion['title']}")
    print(f"Confidence: {suggestion['confidence']}")

    # Manually save selected ones
    if user_approves(suggestion):
        await memory_store.add_memory(
            project_id="my-project",
            memory_type=suggestion['type'],
            title=suggestion['title'],
            content=suggestion['content'],
            reason=suggestion['reason'],
            tags=suggestion['tags'],
            importance=suggestion['importance']
        )

extract_from_git_commit(
    project_id="my-project",
    commit_sha="abc123def456",
    commit_message="feat: add JWT authentication\n\nImplemented JWT-based authentication for API endpoints. Tokens expire after 24 hours with refresh token support.",
    changed_files=[
        "src/auth/jwt.py",
        "src/middleware/auth.py",
        "tests/test_auth.py"
    ],
    auto_save=True
)

from src.codebase_rag.services.memory import memory_extractor

result = await memory_extractor.extract_from_git_commit(
    project_id="my-project",
    commit_sha="abc123def456",
    commit_message="feat: add JWT authentication\n\nImplemented JWT for API",
    changed_files=["src/auth/jwt.py"],
    auto_save=True
)

print(f"Commit type: {result['commit_type']}")
print(f"Extracted: {result['auto_saved_count']} memories")

import subprocess
from pathlib import Path

async def analyze_git_history(
    project_id: str,
    repo_path: str,
    max_commits: int = 50
):
    """Analyze recent git commits and extract memories"""

    # Get recent commits
    result = subprocess.run(
        ["git", "log", f"-{max_commits}", "--pretty=format:%H|%s|%b"],
        cwd=repo_path,
        capture_output=True,
        text=True
    )

    extracted_count = 0

    for line in result.stdout.split('\n'):
        if not line.strip():
            continue

        parts = line.split('|', 2)
        commit_sha = parts[0]
        subject = parts[1]
        body = parts[2] if len(parts) > 2 else ""
        commit_message = f"{subject}\n{body}".strip()

        # Get changed files
        files_result = subprocess.run(
            ["git", "diff-tree", "--no-commit-id", "--name-only", "-r", commit_sha],
            cwd=repo_path,
            capture_output=True,
            text=True
        )
        changed_files = files_result.stdout.strip().split('\n')

        # Extract memories
        result = await memory_extractor.extract_from_git_commit(
            project_id=project_id,
            commit_sha=commit_sha,
            commit_message=commit_message,
            changed_files=changed_files,
            auto_save=True
        )

        if result.get('success'):
            extracted_count += result.get('auto_saved_count', 0)

    print(f"Analyzed {max_commits} commits, extracted {extracted_count} memories")
    return extracted_count

# Usage
await analyze_git_history("my-project", "/path/to/repo", max_commits=100)

extract_from_code_comments(
    project_id="my-project",
    file_path="/path/to/project/src/service.py"
)

from src.codebase_rag.services.memory import memory_extractor

result = await memory_extractor.extract_from_code_comments(
    project_id="my-project",
    file_path="/path/to/project/src/service.py"
)

print(f"Total comments: {result['total_comments']}")
print(f"Extracted: {result['total_extracted']} memories")

class UserService:
    def __init__(self):
        # DECISION: Using Redis for session storage instead of database
        # Reason: Need sub-millisecond latency for session lookups
        self.redis_client = RedisClient()

        # NOTE: Connection pool must have minimum 10 connections
        # Lower values cause connection timeout under load
        self.pool_size = 10

    def authenticate(self, token: str):
        # TODO: Add refresh token rotation for better security
        # This will require database changes and client updates
        pass

    def get_user(self, user_id: int):
        # FIXME: Cache invalidation doesn't work for user updates
        # Need to implement pub/sub pattern for cache invalidation
        return self._fetch_from_db(user_id)

from pathlib import Path

async def extract_from_all_files(
    project_id: str,
    repo_path: str,
    file_patterns: list = ["*.py", "*.js", "*.ts"]
):
    """Extract comments from all matching files"""

    repo = Path(repo_path)
    total_extracted = 0

    for pattern in file_patterns:
        for file_path in repo.rglob(pattern):
            try:
                result = await memory_extractor.extract_from_code_comments(
                    project_id=project_id,
                    file_path=str(file_path)
                )

                if result.get('success'):
                    count = result.get('total_extracted', 0)
                    total_extracted += count
                    print(f"{file_path.name}: {count} memories")

            except Exception as e:
                print(f"Error processing {file_path}: {e}")

    print(f"Total extracted: {total_extracted} memories")
    return total_extracted

# Usage
await extract_from_all_files(
    "my-project",
    "/path/to/repo",
    file_patterns=["*.py", "*.js", "*.ts", "*.java"]
)

suggest_memory_from_query(
    project_id="my-project",
    query="How does the authentication system work?",
    answer="The system uses JWT tokens with refresh token rotation. Access tokens expire after 15 minutes, refresh tokens after 7 days. Tokens are stored in httpOnly cookies to prevent XSS attacks."
)

from src.codebase_rag.services.memory import memory_extractor

result = await memory_extractor.suggest_memory_from_query(
    project_id="my-project",
    query="How does the authentication system work?",
    answer="The system uses JWT tokens with refresh token rotation..."
)

if result['should_save']:
    suggested = result['suggested_memory']
    print(f"Suggested: {suggested['title']}")
    print(f"Type: {suggested['type']}")
    print(f"Importance: {suggested['importance']}")

    # Manually save if approved
    await memory_store.add_memory(
        project_id="my-project",
        **suggested
    )

from src.codebase_rag.services.knowledge import knowledge_service
from src.codebase_rag.services.memory import memory_extractor

async def query_with_memory_suggestion(
    project_id: str,
    query: str
):
    """Query knowledge base and suggest saving as memory if important"""

    # Query knowledge base
    result = await knowledge_service.query_knowledge(query)
    answer = result.get('answer', '')

    # Suggest memory
    suggestion = await memory_extractor.suggest_memory_from_query(
        project_id=project_id,
        query=query,
        answer=answer
    )

    # Auto-save if highly important
    if suggestion.get('should_save'):
        suggested = suggestion['suggested_memory']

        if suggested['importance'] >= 0.8:
            # Auto-save critical information
            await memory_store.add_memory(
                project_id=project_id,
                **suggested,
                metadata={'source': 'auto_query'}
            )
            print(f"Auto-saved memory: {suggested['title']}")

    return {
        'answer': answer,
        'memory_suggested': suggestion.get('should_save', False)
    }

batch_extract_from_repository(
    project_id="my-project",
    repo_path="/path/to/repository",
    max_commits=50,
    file_patterns=["*.py", "*.js", "*.ts"]
)

from src.codebase_rag.services.memory import memory_extractor

result = await memory_extractor.batch_extract_from_repository(
    project_id="my-project",
    repo_path="/path/to/repository",
    max_commits=50,
    file_patterns=["*.py", "*.js", "*.ts"]
)

print(f"Total extracted: {result['total_extracted']}")
print(f"From commits: {result['by_source']['git_commits']}")
print(f"From comments: {result['by_source']['code_comments']}")
print(f"From docs: {result['by_source']['documentation']}")

# Default configuration
await memory_extractor.batch_extract_from_repository(
    project_id="my-project",
    repo_path="/path/to/repo",
    max_commits=50,  # Last 50 commits
    file_patterns=["*.py", "*.js", "*.ts", "*.java", "*.go", "*.rs"]
)

# Focused on recent commits
await memory_extractor.batch_extract_from_repository(
    project_id="my-project",
    repo_path="/path/to/repo",
    max_commits=20,  # Just last 20
    file_patterns=None  # Skip code comments
)

# Deep codebase analysis
await memory_extractor.batch_extract_from_repository(
    project_id="my-project",
    repo_path="/path/to/repo",
    max_commits=100,  # More commits
    file_patterns=["*.py", "*.js", "*.ts", "*.java", "*.go", "*.rs", "*.cpp", "*.c"]
)

# Quick bootstrap
await memory_extractor.batch_extract_from_repository(
    project_id="my-project",
    repo_path="/path/to/repo",
    max_commits=10,  # Limited commits
    file_patterns=["*.py"]  # Single language
)

# Comprehensive analysis (run overnight)
await memory_extractor.batch_extract_from_repository(
    project_id="my-project",
    repo_path="/path/to/repo",
    max_commits=100,
    file_patterns=["*.py", "*.js", "*.ts", "*.java", "*.go"]
)

async def post_session_extraction(
    project_id: str,
    conversation: list,
    repo_path: str
):
    """Extract memories after coding session"""

    # 1. Extract from conversation
    conv_result = await memory_extractor.extract_from_conversation(
        project_id=project_id,
        conversation=conversation,
        auto_save=True
    )

    # 2. Get recent commit (if any)
    import subprocess
    try:
        commit_sha = subprocess.check_output(
            ["git", "rev-parse", "HEAD"],
            cwd=repo_path,
            text=True
        ).strip()

        commit_msg = subprocess.check_output(
            ["git", "log", "-1", "--pretty=%B"],
            cwd=repo_path,
            text=True
        ).strip()

        changed_files = subprocess.check_output(
            ["git", "diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD"],
            cwd=repo_path,
            text=True
        ).strip().split('\n')

        # Extract from commit
        commit_result = await memory_extractor.extract_from_git_commit(
            project_id=project_id,
            commit_sha=commit_sha,
            commit_message=commit_msg,
            changed_files=changed_files,
            auto_save=True
        )

    except Exception as e:
        print(f"No recent commit: {e}")

    return {
        'conversation_memories': conv_result['auto_saved_count'],
        'commit_memories': commit_result.get('auto_saved_count', 0)
    }

# .git/hooks/post-commit
#!/usr/bin/env python3

import asyncio
import subprocess
import sys
sys.path.insert(0, '/path/to/project')

from src.codebase_rag.services.memory import memory_extractor

async def main():
    # Get commit details
    commit_sha = subprocess.check_output(
        ["git", "rev-parse", "HEAD"],
        text=True
    ).strip()

    commit_msg = subprocess.check_output(
        ["git", "log", "-1", "--pretty=%B"],
        text=True
    ).strip()

    changed_files = subprocess.check_output(
        ["git", "diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD"],
        text=True
    ).strip().split('\n')

    # Extract memories
    result = await memory_extractor.extract_from_git_commit(
        project_id="my-project",
        commit_sha=commit_sha,
        commit_message=commit_msg,
        changed_files=changed_files,
        auto_save=True
    )

    if result.get('auto_saved_count', 0) > 0:
        print(f"✅ Extracted {result['auto_saved_count']} memories from commit")

if __name__ == "__main__":
    asyncio.run(main())

import schedule
import asyncio

async def daily_repository_scan():
    """Daily scan of repository for new knowledge"""

    result = await memory_extractor.batch_extract_from_repository(
        project_id="my-project",
        repo_path="/path/to/repo",
        max_commits=10,  # Last day's commits
        file_patterns=["*.py", "*.js"]
    )

    print(f"Daily scan: {result['total_extracted']} new memories")

    # Send notification
    if result['total_extracted'] > 5:
        send_slack_notification(
            f"⚠️ {result['total_extracted']} new memories extracted from codebase"
        )

# Schedule daily at 2 AM
schedule.every().day.at("02:00").do(lambda: asyncio.run(daily_repository_scan()))

from src.codebase_rag.services.memory import memory_extractor

# Lower threshold (more auto-saves)
memory_extractor.confidence_threshold = 0.6

# Higher threshold (fewer auto-saves, higher quality)
memory_extractor.confidence_threshold = 0.8

from src.codebase_rag.services.memory import MemoryExtractor

# Custom limits
MemoryExtractor.MAX_COMMITS_TO_PROCESS = 30
MemoryExtractor.MAX_FILES_TO_SAMPLE = 50
MemoryExtractor.MAX_ITEMS_PER_TYPE = 5

# Critical production project: Review before saving
auto_save=False

# Personal project: Auto-save high confidence
auto_save=True

# Run comprehensive analysis during off-hours
# Avoid running during active development

# Good: 2 AM daily scan
schedule.every().day.at("02:00").do(run_batch_extraction)

# Bad: Every 10 minutes during work hours

result = await memory_extractor.extract_from_conversation(
    project_id="my-project",
    conversation=conversation,
    auto_save=False
)

# Review confidence distribution
high_confidence = sum(1 for s in result['suggestions'] if s['confidence'] >= 0.8)
medium_confidence = sum(1 for s in result['suggestions'] if 0.6 <= s['confidence'] < 0.8)
low_confidence = sum(1 for s in result['suggestions'] if s['confidence'] < 0.6)

print(f"High: {high_confidence}, Medium: {medium_confidence}, Low: {low_confidence}")

# For your codebase, adjust marker importance in memory_extractor.py
# Example: Make TODOs more important in your project

# Override classification
def custom_classify_comment(text: str) -> dict:
    if "TODO:" in text.upper():
        # Higher importance for TODOs in your project
        return {
            "type": "plan",
            "importance": 0.7  # Instead of default 0.4
        }
    # ... other markers

async def review_auto_extracted_memories(project_id: str):
    """Review and clean auto-extracted memories"""

    # Find all auto-extracted memories
    all_memories = await memory_store.search_memories(
        project_id=project_id,
        limit=100
    )

    auto_extracted = [
        m for m in all_memories['memories']
        if m.get('metadata', {}).get('source') in ['conversation', 'git_commit', 'code_comment']
    ]

    # Review low-importance ones
    for memory in auto_extracted:
        if memory['importance'] < 0.4:
            print(f"Review: {memory['title']} (importance: {memory['importance']})")
            # Manually decide: keep, delete, or update importance

# 1. Check LLM is configured
from llama_index.core import Settings
print(f"LLM configured: {Settings.llm is not None}")

# 2. Verify source content is substantial
# Short conversations may not yield memories

# 3. Check confidence threshold
memory_extractor.confidence_threshold = 0.5  # Lower threshold

# 4. Disable auto_save to see all suggestions
result = await memory_extractor.extract_from_conversation(
    project_id="my-project",
    conversation=conversation,
    auto_save=False
)
print(f"Suggestions: {len(result['suggestions'])}")

# 1. Increase confidence threshold
memory_extractor.confidence_threshold = 0.8

# 2. Use manual review
auto_save=False

# 3. Provide better source content
# More detailed conversations yield better extractions

# 1. Reduce max_commits
max_commits=10  # Instead of 50

# 2. Limit file patterns
file_patterns=["*.py"]  # Just Python

# 3. Use sampling
# MemoryExtractor already samples (MAX_FILES_TO_SAMPLE = 30)