CrewAI

pip install crewai supermemory python-dotenv

# .env
SUPERMEMORY_API_KEY=your-supermemory-api-key
OPENAI_API_KEY=your-openai-api-key

import os
from crewai import Agent, Task, Crew, Process
from supermemory import Supermemory
from dotenv import load_dotenv

load_dotenv()

memory = Supermemory()

def build_context(user_id: str, query: str) -> str:
    """Fetch user profile and relevant memories."""
    result = memory.profile(container_tag=user_id, q=query)

    static = result.profile.static or []
    dynamic = result.profile.dynamic or []
    memories = result.search_results.results if result.search_results else []

    return f"""
User Profile:
{chr(10).join(static) if static else 'No profile data.'}

Current Context:
{chr(10).join(dynamic) if dynamic else 'No recent activity.'}

Relevant History:
{chr(10).join([m.memory or m.chunk for m in memories[:5]]) if memories else 'None.'}
"""

def create_agent_with_memory(user_id: str, role: str, goal: str, query: str) -> Agent:
    """Create an agent with user context baked into its backstory."""
    context = build_context(user_id, query)

    return Agent(
        role=role,
        goal=goal,
        backstory=f"""You have access to the following information about the user:
{context}

Use this context to personalize your work.""",
        verbose=True
    )

result = memory.profile(
    container_tag="user_abc",
    q="project planning"  # Optional: also returns relevant memories
)

print(result.profile.static)   # ["Prefers Agile methodology", "Senior engineer"]
print(result.profile.dynamic)  # ["Working on Q2 roadmap", "Focused on API design"]

def store_crew_result(user_id: str, task_description: str, result: str):
    """Save crew output as a memory."""
    memory.add(
        content=f"Task: {task_description}\nResult: {result}",
        container_tag=user_id,
        metadata={"type": "crew_execution"}
    )

results = memory.search.memories(
    q="previous project recommendations",
    container_tag="user_abc",
    search_mode="hybrid",
    limit=10
)

for r in results.results:
    print(r.memory or r.chunk)

import os
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
from supermemory import Supermemory
from dotenv import load_dotenv

load_dotenv()

class ResearchCrew:
    def __init__(self):
        self.memory = Supermemory()
        self.search_tool = SerperDevTool()

    def get_user_context(self, user_id: str, topic: str) -> dict:
        """Retrieve user profile and related research history."""
        result = self.memory.profile(
            container_tag=user_id,
            q=topic,
            threshold=0.5
        )

        return {
            "expertise": result.profile.static or [],
            "focus": result.profile.dynamic or [],
            "history": [m.memory for m in (result.search_results.results or [])[:3]]
        }

    def create_researcher(self, context: dict) -> Agent:
        """Build a researcher agent with user context."""
        expertise_note = ""
        if context["expertise"]:
            expertise_note = f"The user has this background: {', '.join(context['expertise'])}. Adjust technical depth accordingly."

        history_note = ""
        if context["history"]:
            history_note = f"Previous research on related topics: {'; '.join(context['history'])}"

        return Agent(
            role="Research Analyst",
            goal="Conduct research tailored to the user's expertise level",
            backstory=f"""You research topics and synthesize findings into clear summaries.
{expertise_note}
{history_note}""",
            tools=[self.search_tool],
            verbose=True
        )

    def create_writer(self, context: dict) -> Agent:
        """Build a writer agent that matches user preferences."""
        style_note = "Write in a clear, technical style."
        for fact in context.get("expertise", []):
            if "non-technical" in fact.lower():
                style_note = "Write in plain language, avoiding jargon."
                break

        return Agent(
            role="Content Writer",
            goal="Transform research into readable content",
            backstory=f"""You write clear, engaging content. {style_note}""",
            verbose=True
        )

    def research(self, user_id: str, topic: str) -> str:
        """Run the research crew and store results."""
        context = self.get_user_context(user_id, topic)

        researcher = self.create_researcher(context)
        writer = self.create_writer(context)

        research_task = Task(
            description=f"Research the following topic: {topic}",
            expected_output="Detailed findings with sources",
            agent=researcher
        )

        writing_task = Task(
            description="Write a summary based on the research findings",
            expected_output="A clear, structured summary",
            agent=writer
        )

        crew = Crew(
            agents=[researcher, writer],
            tasks=[research_task, writing_task],
            process=Process.sequential,
            verbose=True
        )

        result = crew.kickoff()

        # Store for future sessions
        self.memory.add(
            content=f"Research on '{topic}': {str(result)[:500]}",
            container_tag=user_id,
            metadata={"type": "research", "topic": topic}
        )

        return str(result)


if __name__ == "__main__":
    crew = ResearchCrew()

    # Teach preferences
    crew.memory.add(
        content="User prefers concise summaries with bullet points",
        container_tag="researcher_1"
    )

    # Run research
    result = crew.research("researcher_1", "latest developments in AI agents")
    print(result)

def create_collaborative_context(user_ids: list[str], topic: str) -> str:
    """Aggregate context from multiple users."""
    combined = []

    for user_id in user_ids:
        result = memory.profile(container_tag=user_id, q=topic)
        if result.profile.static:
            combined.append(f"{user_id}: {', '.join(result.profile.static[:3])}")

    return "\n".join(combined) if combined else "No shared context available."

def store_if_successful(user_id: str, task: str, result: str, success: bool):
    """Only store successful task completions."""
    if not success:
        return

    memory.add(
        content=f"Completed: {task}\nOutcome: {result}",
        container_tag=user_id,
        metadata={"type": "success", "task": task}
    )

# Store with metadata
memory.add(
    content="Research findings on distributed systems",
    container_tag="user_123",
    metadata={
        "project": "infrastructure-review",
        "agents": ["researcher", "writer"],
        "confidence": "high"
    }
)

# Search with filters
results = memory.search.memories(
    q="distributed systems",
    container_tag="user_123",
    filters={
        "AND": [
            {"key": "project", "value": "infrastructure-review"},
            {"key": "confidence", "value": "high"}
        ]
    }
)