GROQ with CREWAI Approach

agents.yaml

content_writer:
  role: >
    Educational Content Writer
  goal: >
    Create engaging, informative content that thoroughly explains the assigned topic
    and provides valuable insights to the reader
  backstory: >
    You are a talented educational writer with expertise in creating clear, engaging
    content. You have a gift for explaining complex concepts in accessible language
    and organizing information in a way that helps readers build their understanding.
  llm:
    model: groq/llama-3.3-70b-versatile

content_reviewer:
  role: >
    Content Review Specialist
  goal: >
    Ensure content quality, accuracy, and consistency while maintaining
    appropriate tone and complexity for the target audience
  backstory: >
    You are an experienced editor with a keen eye for detail and a deep
    understanding of educational content standards. You excel at improving
    clarity and ensuring content meets its educational objectives.
  llm:
    model: groq/llama-3.3-70b-versatile

main.py

#!/usr/bin/env python
import json
from typing import List, Dict
from pydantic import BaseModel, Field
from crewai import LLM
from crewai.flow.flow import Flow, listen, start
from researcher_content.crews.content_crew.content_crew import ContentCrew
from dotenv import load_dotenv
import os

load_dotenv()

# Create a default LLM configuration
def get_default_llm():
    return LLM(
        model="groq/llama-3.3-70b-versatile"
    )

# Define our models for structured data
class Section(BaseModel):
    title: str = Field(description="Title of the section")
    description: str = Field(description="Brief description of what the section should cover")

class GuideOutline(BaseModel):
    title: str = Field(description="Title of the guide")
    introduction: str = Field(description="Introduction to the topic")
    target_audience: str = Field(description="Description of the target audience")
    sections: List[Section] = Field(description="List of sections in the guide")
    conclusion: str = Field(description="Conclusion or summary of the guide")

# Define our flow state
class GuideCreatorState(BaseModel):
    topic: str = ""
    audience_level: str = ""
    guide_outline: GuideOutline = None
    sections_content: Dict[str, str] = Field(default_factory=dict)

class GuideCreatorFlow(Flow[GuideCreatorState]):
    def __init__(self):
        super().__init__()
        self.llm = get_default_llm()

    """Flow for creating a comprehensive guide on any topic"""

    @start()
    def get_user_input(self):
        """Get input from the user about the guide topic and audience"""
        print("\n=== Create Your Comprehensive Guide ===\n")

        # Get user input
        self.state.topic = input("What topic would you like to create a guide for? ")

        # Get audience level with validation
        while True:
            audience = input("Who is your target audience? (beginner/intermediate/advanced) ").lower()
            if audience in ["beginner", "intermediate", "advanced"]:
                self.state.audience_level = audience
                break
            print("Please enter 'beginner', 'intermediate', or 'advanced'")

        print(f"\nCreating a guide on {self.state.topic} for {self.state.audience_level} audience...\n")
        return self.state

    @listen(get_user_input)
    def create_guide_outline(self, state):
        """Create a structured outline for the guide using a direct LLM call"""
        print("Creating guide outline...")

        # Use the configured LLM
        llm = self.llm

        # Create the messages for the outline
        messages = [
            {"role": "system", "content": """You are a helpful assistant that ALWAYS responds in valid JSON format.
             Your response should be a single JSON object containing the guide outline structure.
             The JSON structure must include: title, introduction, target_audience, sections (array of objects with title and description), and conclusion."""},
            {"role": "user", "content": f"""
            Create a detailed outline for a comprehensive guide on "{state.topic}" for {state.audience_level} level learners.
            
            Respond with a JSON object that has this exact structure:
            {{
                "title": "string",
                "introduction": "string",
                "target_audience": "string",
                "sections": [
                    {{
                        "title": "string",
                        "description": "string"
                    }}
                ],
                "conclusion": "string"
            }}

            Include 4-6 main sections that cover the most important aspects of the topic.
            Make sure the response is valid JSON that can be parsed."""}
        ]

        try:
            # Make the LLM call with JSON response format
            response = llm.call(messages=messages)
            
            # Try to parse the JSON response
            try:
                outline_dict = json.loads(response)
            except json.JSONDecodeError:
                # If parsing fails, try to extract JSON from the response
                # Sometimes the model might include additional text
                import re
                json_match = re.search(r'\{.*\}', response, re.DOTALL)
                if json_match:
                    outline_dict = json.loads(json_match.group(0))
                else:
                    raise Exception("Could not extract valid JSON from the response")

            self.state.guide_outline = GuideOutline(**outline_dict)

            # Save the outline to a file
            with open("output/guide_outline.json", "w") as f:
                json.dump(outline_dict, f, indent=2)

            print(f"Guide outline created with {len(self.state.guide_outline.sections)} sections")
            return self.state.guide_outline

        except Exception as e:
            print(f"Error creating outline: {str(e)}")
            # Create a basic outline as fallback
            fallback_outline = {
                "title": f"Guide to {state.topic}",
                "introduction": f"An introduction to {state.topic}",
                "target_audience": f"{state.audience_level} level learners",
                "sections": [
                    {
                        "title": "Getting Started",
                        "description": "Basic introduction to the concepts"
                    },
                    {
                        "title": "Core Concepts",
                        "description": "Essential principles and ideas"
                    },
                    {
                        "title": "Practical Applications",
                        "description": "How to apply the knowledge"
                    },
                    {
                        "title": "Advanced Topics",
                        "description": "Deeper dive into complex aspects"
                    }
                ],
                "conclusion": f"Summary of key points about {state.topic}"
            }
            self.state.guide_outline = GuideOutline(**fallback_outline)
            
            # Save the fallback outline
            with open("output/guide_outline.json", "w") as f:
                json.dump(fallback_outline, f, indent=2)
                
            print("Created basic outline structure due to error")
            return self.state.guide_outline

    @listen(create_guide_outline)
    def write_and_compile_guide(self, outline):
        """Write all sections and compile the guide"""
        print("Writing guide sections and compiling...")
        completed_sections = []
        
        try:
            from crewai import Agent, Task, Crew, Process

            # Create agents once
            writer = Agent(
                role='Content Writer',
                goal='Write comprehensive and engaging content',
                backstory='Expert content writer with deep knowledge in various topics',
                llm=self.llm,
                verbose=True
            )

            reviewer = Agent(
                role='Content Reviewer',
                goal='Review and improve content quality',
                backstory='Experienced editor with attention to detail',
                llm=self.llm,
                verbose=True
            )

            # Process sections one by one
            for section in outline.sections:
                section_key = section.title
                print(f"Processing section: {section_key}")

                # Build context from previous sections
                previous_sections_text = ""
                if completed_sections:
                    previous_sections_text = "# Previously Written Sections\n\n"
                    for prev_title in completed_sections:
                        content = self.state.sections_content.get(prev_title, "")
                        previous_sections_text += f"## {prev_title}\n\n{content}\n\n"
                else:
                    previous_sections_text = "No previous sections written yet."

                try:
                    # Create tasks for this section
                    write_task = Task(
                        description=f"""Write a comprehensive section about '{section_key}' for a {self.state.audience_level} level audience.
                            Section Description: {section.description}
                            Previous Sections: {previous_sections_text}
                            
                            Write a detailed, well-structured section that covers all important aspects.
                            Use clear examples and explanations appropriate for the {self.state.audience_level} level.
                            Ensure smooth transitions between topics.""",
                        expected_output="A well-written section in Markdown format",
                        agent=writer
                    )

                    review_task = Task(
                        description=f"""Review and improve the content for section '{section_key}'.
                            Ensure:
                            1. Content matches the {self.state.audience_level} level
                            2. Information is accurate and well-organized
                            3. Writing is clear and engaging
                            4. Section flows well with previous content
                            
                            Make necessary improvements while maintaining the original message.""",
                        expected_output="An improved version of the section in Markdown format",
                        agent=reviewer
                    )

                    # Create crew for this section
                    crew = Crew(
                        agents=[writer, reviewer],
                        tasks=[write_task, review_task],
                        process=Process.sequential,
                        verbose=True
                    )

                    # Execute the tasks
                    result = crew.kickoff()
                    
                    # Store the result
                    self.state.sections_content[section_key] = str(result)
                    completed_sections.append(section_key)
                    print(f"Section completed: {section_key}")

                except Exception as e:
                    print(f"Error processing section {section_key}: {str(e)}")
                    self.state.sections_content[section_key] = f"Error generating content: {str(e)}"

            # Compile the final guide
            guide_content = [f"# {outline.title}\n\n"]
            guide_content.append(f"## Introduction\n\n{outline.introduction}\n\n")
            guide_content.append(f"## Target Audience\n\n{outline.target_audience}\n\n")

            # Add each section in order
            for section in outline.sections:
                section_content = self.state.sections_content.get(section.title, "")
                guide_content.append(f"## {section.title}\n\n{section_content}\n\n")

            # Add conclusion
            guide_content.append(f"## Conclusion\n\n{outline.conclusion}\n\n")

            # Create output directory if it doesn't exist
            os.makedirs("output", exist_ok=True)

            # Join all content and save the guide
            final_content = "".join(guide_content)
            with open("output/complete_guide.md", "w", encoding='utf-8') as f:
                f.write(final_content)

            print("\nComplete guide compiled and saved to output/complete_guide.md")
            return "Guide creation completed successfully"

        except Exception as e:
            print(f"Error in write_and_compile_guide: {str(e)}")
            raise

def kickoff():
    """Run the guide creator flow"""
    GuideCreatorFlow().kickoff()
    print("\n=== Flow Complete ===")
    print("Your comprehensive guide is ready in the output directory.")
    print("Open output/complete_guide.md to view it.")

def plot():
    """Generate a visualization of the flow"""
    flow = GuideCreatorFlow()
    flow.plot("guide_creator_flow")
    print("Flow visualization saved to guide_creator_flow.html")

if __name__ == "__main__":
    kickoff()
    plot()