CrewAI Integration

pip install crewai memorystack

from crewai import Agent, Task, Crew, Tool
from langchain.chat_models import ChatOpenAI
from memorystack import MemoryStackClient
import os

# Initialize Memorystack (only needs API key)
memory = MemoryStackClient(
    api_key=os.environ["MEMORYSTACK_API_KEY"]
)

# Custom tools to access memories
def search_memories(query: str):
    """Search memories for relevant information"""
    results = memory.search(query, limit=10)
    return "\n".join([
        f"- {m['content']}"
        for m in results.get('results', [])
    ])

def store_memory(content: str, agent_id: str):
    """Store information in memory"""
    return memory.add(content, metadata={"agent_id": agent_id})

memory_search_tool = Tool(
    name="search_memories",
    description="Search memories for context and preferences",
    func=search_memories
)

memory_store_tool = Tool(
    name="store_memory",
    description="Store important information in memory",
    func=lambda x: store_memory(x, "researcher")
)

# Create agents with memory access
researcher = Agent(
    role='Researcher',
    goal='Research topics based on user preferences',
    backstory='Expert researcher who remembers user interests',
    tools=[memory_search_tool, memory_store_tool],
    llm=ChatOpenAI(model="gpt-4"),
    verbose=True
)

writer = Agent(
    role='Writer',
    goal='Write content tailored to user preferences',
    backstory='Creative writer who personalizes content',
    tools=[memory_search_tool],
    llm=ChatOpenAI(model="gpt-4"),
    verbose=True
)

# Create tasks
research_task = Task(
    description='Research AI trends considering user preferences from memory',
    agent=researcher,
    expected_output='Research summary'
)

write_task = Task(
    description='Write an article based on research and user preferences',
    agent=writer,
    expected_output='Article content'
)

# Create crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    verbose=True
)

# Run crew
result = crew.kickoff()

# Save crew output to memory
memory.add(
    f"Crew completed task: {result}",
    metadata={"type": "crew_output"}
)

print(result)

class LearningCrew:
    def __init__(self, api_key: str, user_id: str):
        # Initialize client (only needs API key)
        self.memory = MemoryStackClient(api_key=api_key)
        self.user_id = user_id
        self.setup_crew()
    
    def get_learnings(self):
        """Get past learnings from memory"""
        results = self.memory.search(
            "learning feedback experience",
            user_id=self.user_id,
            limit=20
        )
        return "\n".join([m['content'] for m in results.get('results', [])])
    
    def setup_crew(self):
        # Get past learnings
        learnings = self.get_learnings()
        
        # Create learning-aware agent
        self.agent = Agent(
            role='Learning Assistant',
            goal='Help user while learning from feedback',
            backstory=f'I learn from experience. Past learnings:\n{learnings}',
            llm=ChatOpenAI(model="gpt-4"),
            verbose=True
        )
        
        self.task = Task(
            description='Complete user request and learn from outcome',
            agent=self.agent,
            expected_output='Task result'
        )
        
        self.crew = Crew(
            agents=[self.agent],
            tasks=[self.task],
            verbose=True
        )
    
    def execute(self, request: str):
        """Execute task and save learnings"""
        # Run crew
        result = self.crew.kickoff()
        
        # Save interaction as conversation
        self.memory.add([
            {"role": "user", "content": request},
            {"role": "assistant", "content": str(result)}
        ], user_id=self.user_id)
        
        return result
    
    def add_feedback(self, feedback: str):
        """Learn from user feedback"""
        self.memory.add(
            f"Feedback: {feedback}",
            user_id=self.user_id,
            metadata={"type": "learning"}
        )
        
        # Refresh crew with new learnings
        self.setup_crew()

# Usage
crew = LearningCrew(
    api_key=os.environ["MEMORYSTACK_API_KEY"],
    user_id="user_123"
)

# Execute task
result = crew.execute("Write a blog post about AI")
print(result)

# Provide feedback
crew.add_feedback("Great! But make it more technical next time")

# Next execution will use the feedback
result2 = crew.execute("Write another blog post")
print(result2)