Create multi_agent_book_workflow.py

multi_agent_book_workflow.py

@@ -0,0 +1,278 @@
+# File: multi_agent_book_workflow.py
+# Location: /multi_agent_book_workflow.py (in root directory)
+# Description: Core multi-agent book writing orchestration system
+# Usage: Imported by app.py for book generation workflows
+
+import os
+from typing import Dict, List, Any
+import uuid
+import chromadb
+import numpy as np
+
+# Agent and LLM Frameworks
+from crewai import Agent, Task, Crew
+from langchain_openai import ChatOpenAI
+from langchain_anthropic import ChatAnthropic
+from langchain.memory import ConversationBufferMemory
+from langchain.embeddings import OpenAIEmbeddings
+
+class BookWritingOrchestrator:
+    def __init__(self, project_id: str = None):
+        """
+        Initialize the book writing orchestrator with multi-agent setup
+        
+        Args:
+            project_id (str, optional): Unique identifier for the book project
+        """
+        # Project Identification
+        self.project_id = project_id or str(uuid.uuid4())
+        
+        # Vector Store for Context Management
+        self.chroma_client = chromadb.Client()
+        self.context_store = self.chroma_client.create_collection(
+            name=f"book_context_{self.project_id}"
+        )
+        
+        # Embedding Model
+        self.embeddings = OpenAIEmbeddings()
+        
+        # LLM Configurations
+        self.openai_llm = ChatOpenAI(
+            model="gpt-4-turbo",
+            temperature=0.7
+        )
+        self.anthropic_llm = ChatAnthropic(
+            model="claude-3-opus-20240229",
+            temperature=0.7
+        )
+        
+        # Agent Memory
+        self.global_memory = ConversationBufferMemory(
+            memory_key="chat_history",
+            return_messages=True
+        )
+        
+        # Initialize Specialized Agents
+        self.setup_agents()
+    
+    def setup_agents(self):
+        """
+        Create specialized agents for book writing workflow
+        """
+        # Concept Development Agent
+        self.concept_agent = Agent(
+            role='Book Concept Architect',
+            goal='Develop a comprehensive and compelling book concept',
+            backstory='An expert literary strategist who transforms raw ideas into structured book frameworks',
+            verbose=True,
+            llm=self.openai_llm,
+            memory=True
+        )
+        
+        # Research Agent
+        self.research_agent = Agent(
+            role='Literary Research Specialist',
+            goal='Conduct in-depth research to support the book\'s narrative and authenticity',
+            backstory='A meticulous researcher with expertise in gathering and synthesizing complex information',
+            verbose=True,
+            llm=self.anthropic_llm,
+            memory=True
+        )
+        
+        # Content Generation Agent
+        self.writing_agent = Agent(
+            role='Master Storyteller',
+            goal='Create engaging, coherent, and stylistically consistent narrative content',
+            backstory='A versatile writer capable of crafting compelling prose across various genres',
+            verbose=True,
+            llm=self.openai_llm,
+            memory=True
+        )
+        
+        # Editing Agent
+        self.editing_agent = Agent(
+            role='Narrative Refinement Specialist',
+            goal='Ensure narrative consistency, quality, and stylistic excellence',
+            backstory='A seasoned editor with a keen eye for storytelling nuance and structural integrity',
+            verbose=True,
+            llm=self.anthropic_llm,
+            memory=True
+        )
+    
+    def generate_book_concept(self, initial_prompt: str) -> Dict[str, Any]:
+        """
+        Generate a comprehensive book concept using multi-agent collaboration
+        
+        Args:
+            initial_prompt (str): User's initial book idea
+        
+        Returns:
+            Dict containing book concept details
+        """
+        # Concept Development Task
+        concept_task = Task(
+            description=f"""
+            Develop a comprehensive book concept based on this initial idea:
+            {initial_prompt}
+            
+            Provide detailed outputs including:
+            1. Unique Book Title
+            2. Genre and Subgenre
+            3. Target Audience
+            4. Core Narrative Premise
+            5. Potential Chapter Outline
+            6. Distinctive Narrative Approach
+            """,
+            agent=self.concept_agent,
+            expected_output="A detailed JSON-like structure of the book concept"
+        )
+        
+        # Research Validation Task
+        research_task = Task(
+            description="Validate and enrich the book concept with additional research insights",
+            agent=self.research_agent,
+            expected_output="Research-backed annotations and potential depth areas"
+        )
+        
+        # Create Crew for Collaborative Processing
+        book_concept_crew = Crew(
+            agents=[self.concept_agent, self.research_agent],
+            tasks=[concept_task, research_task],
+            verbose=2
+        )
+        
+        # Execute Collaborative Workflow
+        result = book_concept_crew.kickoff()
+        
+        # Store Context in Vector Database
+        self._store_context('book_concept', result)
+        
+        return self._parse_concept(result)
+    
+    def generate_chapter_content(
+        self, 
+        book_concept: Dict[str, Any], 
+        chapter_number: int
+    ) -> str:
+        """
+        Generate content for a specific chapter using multi-agent workflow
+        
+        Args:
+            book_concept (Dict): Comprehensive book concept
+            chapter_number (int): Chapter to generate
+        
+        Returns:
+            str: Generated chapter content
+        """
+        # Retrieve Previous Context
+        previous_context = self._retrieve_context(chapter_number)
+        
+        # Content Generation Task
+        writing_task = Task(
+            description=f"""
+            Write Chapter {chapter_number} for the book
+            
+            Book Concept: {book_concept}
+            
+            Previous Context: {previous_context}
+            
+            Generate a draft that:
+            1. Maintains narrative continuity
+            2. Advances the story
+            3. Matches the established tone and style
+            4. Incorporates research insights
+            """,
+            agent=self.writing_agent,
+            expected_output="A complete chapter draft with narrative coherence"
+        )
+        
+        # Editing and Refinement Task
+        editing_task = Task(
+            description="Review and refine the generated chapter draft",
+            agent=self.editing_agent,
+            expected_output="Polished chapter content with suggested improvements"
+        )
+        
+        # Create Crew for Chapter Generation
+        chapter_crew = Crew(
+            agents=[self.writing_agent, self.editing_agent],
+            tasks=[writing_task, editing_task],
+            verbose=2
+        )
+        
+        # Execute Collaborative Workflow
+        chapter_content = chapter_crew.kickoff()
+        
+        # Store Chapter Context
+        self._store_context(f'chapter_{chapter_number}', chapter_content)
+        
+        return chapter_content
+    
+    def _store_context(self, context_key: str, content: str):
+        """
+        Store context in vector database for retrieval
+        
+        Args:
+            context_key (str): Unique identifier for the context
+            content (str): Content to store
+        """
+        # Generate embedding
+        embedding = self.embeddings.embed_documents([content])[0]
+        
+        # Store in Chroma
+        self.context_store.add(
+            embeddings=[embedding],
+            documents=[content],
+            ids=[f"{self.project_id}_{context_key}"]
+        )
+    
+    def _retrieve_context(
+        self, 
+        chapter_number: int, 
+        top_k: int = 3
+    ) -> List[str]:
+        """
+        Retrieve contextually relevant previous content
+        
+        Args:
+            chapter_number (int): Current chapter number
+            top_k (int): Number of context pieces to retrieve
+        
+        Returns:
+            List of contextually relevant content
+        """
+        # Retrieve previous chapters
+        previous_chapters = [
+            f'chapter_{i}' for i in range(1, chapter_number)
+        ]
+        
+        # Search context
+        retrieved_contexts = []
+        for chapter_key in previous_chapters:
+            search_embedding = self.embeddings.embed_documents([chapter_key])[0]
+            results = self.context_store.query(
+                query_embeddings=[search_embedding],
+                n_results=top_k
+            )
+            
+            retrieved_contexts.extend(results['documents'][0])
+        
+        return retrieved_contexts
+    
+    def _parse_concept(self, concept_result: str) -> Dict[str, Any]:
+        """
+        Parse and structure the book concept
+        
+        Args:
+            concept_result (str): Raw concept generation result
+        
+        Returns:
+            Structured book concept dictionary
+        """
+        # Implement robust parsing logic
+        # This could use another AI call or regex-based extraction
+        return {
+            'title': 'Parsed Book Title',
+            'genre': 'Parsed Genre',
+            'chapters': []  # Parsed chapter outline
+        }