""" LangGraph Agent for Vibe Reader Implements the agentic workflow for book recommendation based on visual vibes """ import os import json from typing import TypedDict, List, Dict, Any, Literal, Annotated from operator import add from openai import OpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langgraph.graph import StateGraph, END from langgraph.types import interrupt from dotenv import load_dotenv # ============================================================================ # CONFIGURATION # ============================================================================ NEBIUS_API_KEY = os.getenv("NEBIUS_API_KEY") NEBIUS_BASE_URL = "https://api.tokenfactory.nebius.com/v1/" VLM_MODEL = "google/gemma-3-27b-it-fast" REASONING_MODEL = "Qwen/Qwen3-30B-A3B-Thinking-2507" FAST_MODEL = "moonshotai/Kimi-K2-Instruct" # Non-thinking model for simple tasks MODAL_VECTOR_STORE_URL = os.getenv("MODAL_VECTOR_STORE_URL", "https://placeholder-modal-url.modal.run/search") GOOGLE_BOOKS_MCP_URL = os.getenv("GOOGLE_BOOKS_MCP_URL", "https://mcp-1st-birthday-google-books-mcp.hf.space") NUM_BOOKS_TO_RETRIEVE = 7 # Target number of books with valid descriptions NUM_BOOKS_TO_FETCH = 13 # Fetch extra to account for books without descriptions NUM_FINAL_BOOKS = 3 # ============================================================================ # STATE DEFINITION # ============================================================================ class AgentState(TypedDict): """State maintained throughout the agent workflow""" # User inputs images: List[str] # List of image URLs or base64 encoded images # Conversation history (no reducer - we manage the list directly) messages: List[Dict[str, str]] # Vibe components (from JSON extraction) aesthetic_genre_keywords: List[str] # Genre/aesthetic keywords mood_atmosphere: List[str] # Mood descriptors core_themes: List[str] # Core themes tropes: List[str] # Story tropes feels_like: str # User-facing "feels like" description (what gets refined) vibe_refinement_count: int # Number of refinement iterations # Book retrieval retrieved_books: List[Dict[str, str]] # List of {title, author} dicts books_with_metadata: List[Dict[str, Any]] # Enriched with Google Books data # Narrowing process q1_question: str # First narrowing question (stored for resume) q2_question: str # Second narrowing question (stored for resume) user_preferences: Dict[str, Any] # Accumulated user preferences from Q&A (question + answer pairs) final_books: List[Dict[str, Any]] # Final 3 books # Final outputs soundtrack_url: str # ElevenLabs generated soundtrack # Debug/reasoning (no reducer - we manage the list directly) reasoning: List[str] # ============================================================================ # HELPER FUNCTIONS # ============================================================================ def create_openai_client() -> OpenAI: """Create OpenAI client configured for Nebius""" return OpenAI(api_key=NEBIUS_API_KEY, base_url=NEBIUS_BASE_URL) def call_llm(messages: List[Dict[str, Any]], temperature: float = 0.7, model: str = REASONING_MODEL, include_reasoning: bool = False, max_tokens: int = 2500): """Generic LLM call for reasoning and decision-making using Nebius API Args: messages: Conversation messages temperature: Sampling temperature model: Model to use include_reasoning: If True, returns tuple of (content, reasoning_text) max_tokens: Maximum tokens for response (default 1000) Returns: str or tuple: Response content, or (content, reasoning) if include_reasoning=True """ client = create_openai_client() # Uses Nebius response = client.chat.completions.create( model=model, messages=messages, temperature=temperature, max_tokens=max_tokens ) message = response.choices[0].message content = message.content or "" if include_reasoning: # Nebius API returns reasoning in a separate field for Thinking models reasoning = getattr(message, 'reasoning_content', None) or "" if reasoning: # If content is empty, log a warning but don't try to extract from reasoning # (the last line of reasoning is usually garbage, not the answer) if not content.strip(): print(f"[DEBUG AGENT] Warning: LLM returned empty content with reasoning. This may indicate an issue.") return content, reasoning # Fallback: try parsing ... tags from content import re think_match = re.match(r'(.*?)(.*)', content, re.DOTALL) if think_match: reasoning = think_match.group(1).strip() final_content = think_match.group(2).strip() return final_content, reasoning # No reasoning found return content, "No reasoning trace found" return content # ============================================================================ # NODES # ============================================================================ def generate_initial_vibe(state: AgentState) -> AgentState: """Node: Generate initial vibe description from uploaded images using VLM""" from prompts import VIBE_EXTRACTION from utils import parse_json_response, extract_vibe_components client = create_openai_client() # Construct message with images content = [{"type": "text", "text": "Analyze these images and extract the vibe:"}] for img in state["images"]: # Convert local file paths to base64 data URLs if needed if img.startswith(('http://', 'https://', 'data:')): # Already a valid URL image_url = img else: # Local file path - convert to base64 import base64 from pathlib import Path img_path = Path(img) if img_path.exists(): with open(img_path, 'rb') as f: img_data = base64.b64encode(f.read()).decode('utf-8') # Determine MIME type from extension ext = img_path.suffix.lower() mime_types = {'.jpg': 'jpeg', '.jpeg': 'jpeg', '.png': 'png', '.gif': 'gif', '.webp': 'webp'} mime = mime_types.get(ext, 'jpeg') image_url = f"data:image/{mime};base64,{img_data}" else: state["reasoning"].append(f"⚠️ Warning: Image file not found: {img}") continue content.append({ "type": "image_url", "image_url": {"url": image_url} }) response = client.chat.completions.create( model=VLM_MODEL, messages=[ {"role": "system", "content": VIBE_EXTRACTION}, {"role": "user", "content": content} ], temperature=0.7, max_tokens=2000 ) vibe_json_str = response.choices[0].message.content # Parse JSON response vibe_json = parse_json_response(vibe_json_str) if not vibe_json: state["reasoning"].append(f"❌ Failed to parse vibe JSON. Raw response: {vibe_json_str[:200]}") # Fallback to simple extraction state["feels_like"] = vibe_json_str state["aesthetic_genre_keywords"] = [] state["mood_atmosphere"] = [] state["core_themes"] = [] state["tropes"] = [] else: # Extract components components = extract_vibe_components(vibe_json) state["aesthetic_genre_keywords"] = components["aesthetic_genre_keywords"] state["mood_atmosphere"] = components["mood_atmosphere"] state["core_themes"] = components["core_themes"] state["tropes"] = components["tropes"] state["feels_like"] = components["feels_like"] state["reasoning"].append(f"✅ Extracted vibe components:\n" f" - Aesthetics: {', '.join(state['aesthetic_genre_keywords'])}\n" f" - Mood: {', '.join(state['mood_atmosphere'])}\n" f" - Themes: {', '.join(state['core_themes'])}\n" f" - Tropes: {', '.join(state['tropes'])}") state["vibe_refinement_count"] = 0 # Only show feels_like to user assistant_message = f"Here's the vibe I'm getting from your images:\n\n{state['feels_like']}\n\nDoes this capture what you're looking for, or would you like me to adjust it?" state["messages"].append({ "role": "assistant", "content": assistant_message }) # Wait for user feedback; when resumed, user_response will contain their reply user_response = interrupt(assistant_message) if user_response: state["messages"].append({"role": "user", "content": user_response}) return state def refine_vibe(state: AgentState) -> AgentState: """Node: Refine vibe based on user feedback - only refines feels_like portion""" from prompts import VIBE_REFINEMENT from utils import strip_thinking_tags print("[DEBUG AGENT] refine_vibe node started") # Get the latest user message (feedback) user_messages = [m for m in state["messages"] if m.get("role") == "user"] print(f"[DEBUG AGENT] Found {len(user_messages)} user messages") if not user_messages: state["reasoning"].append("⚠️ No user feedback found for refinement; skipping refine_vibe step") return state user_feedback = user_messages[-1]["content"] print(f"[DEBUG AGENT] user_feedback: {user_feedback[:50] if user_feedback else 'None'}...") # Use LLM to refine only the feels_like description # Keep other vibe components (aesthetics, themes, tropes) unchanged messages = [ {"role": "system", "content": VIBE_REFINEMENT}, {"role": "user", "content": f"Current 'feels like' description: {state['feels_like']}\n\nUser feedback: {user_feedback}\n\nProvide the refined 'feels like' description (4-5 sentences):"} ] print(f"[DEBUG AGENT] Calling LLM for refinement...") refined_feels_like, reasoning = call_llm(messages, temperature=0.7, include_reasoning=True) print(f"[DEBUG AGENT] LLM returned content: {refined_feels_like[:200] if refined_feels_like else 'None'}...") print(f"[DEBUG AGENT] LLM reasoning: {reasoning[:200] if reasoning else 'None'}...") # Ensure no thinking tags leak into the feels_like refined_feels_like = strip_thinking_tags(refined_feels_like) # Update only the feels_like portion state["feels_like"] = refined_feels_like state["vibe_refinement_count"] += 1 assistant_message = f"I've refined the vibe:\n\n{refined_feels_like}\n\nIs this better, or would you like further adjustments?" print(f"[DEBUG AGENT] Adding assistant message to state, current msg count: {len(state['messages'])}") state["messages"].append({ "role": "assistant", "content": assistant_message }) state["reasoning"].append(f"🧠 REASONING (Vibe Refinement #{state['vibe_refinement_count']}):\n{reasoning}\n") print(f"[DEBUG AGENT] After append, msg count: {len(state['messages'])}") # Wait for user feedback on the refined vibe print(f"[DEBUG AGENT] About to call interrupt()") user_response = interrupt(assistant_message) print(f"[DEBUG AGENT] interrupt() returned: {user_response}") if user_response: state["messages"].append({"role": "user", "content": user_response}) return state def check_vibe_satisfaction(state: AgentState) -> Literal["refine", "retrieve"]: """Conditional edge: Check if user is satisfied with vibe description""" from prompts import VIBE_SATISFACTION_CHECKER # Get the last user message user_messages = [m for m in state["messages"] if m.get("role") == "user"] if not user_messages: # No explicit feedback; default to moving forward return "retrieve" raw_content = user_messages[-1]["content"] # Content may occasionally be a non-string (e.g., list from upstream tools); # normalize to text before passing into the LLM. if isinstance(raw_content, str): last_user_msg = raw_content elif isinstance(raw_content, list): # Join any text-like chunks into a single string representation last_user_msg = " ".join(str(x) for x in raw_content) else: last_user_msg = str(raw_content) # Use LLM to determine satisfaction messages = [ {"role": "system", "content": VIBE_SATISFACTION_CHECKER}, {"role": "user", "content": f"User's response: {last_user_msg}"} ] decision, reasoning = call_llm(messages, temperature=0.0, include_reasoning=True) decision = decision.strip().lower() if decision else "" print(f"[DEBUG] check_vibe_satisfaction - user said: '{last_user_msg}'") print(f"[DEBUG] check_vibe_satisfaction - LLM decision: '{decision}'") state["reasoning"].append(f"🧠 REASONING (Satisfaction Check):\n{reasoning}\n\n→ Decision: {decision}") if "satisfied" in decision and "not_satisfied" not in decision: print(f"[DEBUG] check_vibe_satisfaction -> RETRIEVE (user satisfied)") return "retrieve" else: print(f"[DEBUG] check_vibe_satisfaction -> REFINE (user not satisfied)") return "refine" def retrieve_books(state: AgentState) -> AgentState: """Node: Retrieve books from Modal vector store""" import requests # Construct full vibe query from all components vibe_query = f"{state['feels_like']}\n\nGenres/Aesthetics: {', '.join(state['aesthetic_genre_keywords'])}\nMood: {', '.join(state['mood_atmosphere'])}\nThemes: {', '.join(state['core_themes'])}\nTropes: {', '.join(state['tropes'])}" try: # Call Modal vector store endpoint print(f"DEBUG: Calling Modal URL: {MODAL_VECTOR_STORE_URL}") state["reasoning"].append(f"📚 Calling Modal vector store with full vibe profile") state["reasoning"].append(f"URL: {MODAL_VECTOR_STORE_URL}") response = requests.post( MODAL_VECTOR_STORE_URL, json={ "query": vibe_query, "top_k": NUM_BOOKS_TO_RETRIEVE, "min_books_per_vibe": 1 }, timeout=180 # Long timeout for cold start ) print(f"DEBUG: Response status: {response.status_code}") print(f"DEBUG: Response text: {response.text[:500] if response.text else 'empty'}") if response.status_code == 200: data = response.json() # Extract books from search results with diversity across vibes # Modal returns: {"results": [{"books": [...], "vibe_data": {...}, "score": ...}], ...} # Strategy: Take up to MAX_BOOKS_PER_VIBE from each vibe to ensure diversity MAX_BOOKS_PER_VIBE = 5 books = [] seen = set() # Track seen books for deduplication for result in data.get("results", []): vibe_score = result.get("score", 0) vibe_books = result.get("books", []) books_from_this_vibe = 0 for book in vibe_books: if books_from_this_vibe >= MAX_BOOKS_PER_VIBE: break title = book.get("title", "") author = book.get("author", "") key = (title.lower(), author.lower()) # Skip duplicates if key in seen: continue seen.add(key) books.append({ "title": title, "author": author, "vibe_score": vibe_score # Track which vibe it came from }) books_from_this_vibe += 1 # Fetch extra books to account for filtering (books without descriptions) books = books[:NUM_BOOKS_TO_FETCH] state["reasoning"].append(f"Retrieved {len(books)} books from {len(data.get('results', []))} vibes (max {MAX_BOOKS_PER_VIBE} per vibe)") else: raise Exception(f"HTTP {response.status_code}: {response.text[:200]}") except Exception as e: # Fallback to mock data for development print(f"DEBUG ERROR: Vector store call failed: {e}") import traceback traceback.print_exc() state["reasoning"].append(f"Vector store call failed: {e}. Using mock data.") books = [ {"title": "The Night Circus", "author": "Erin Morgenstern"}, {"title": "The Ocean at the End of the Lane", "author": "Neil Gaiman"}, {"title": "The Starless Sea", "author": "Erin Morgenstern"}, {"title": "Piranesi", "author": "Susanna Clarke"}, {"title": "The House in the Cerulean Sea", "author": "TJ Klune"}, {"title": "Howl's Moving Castle", "author": "Diana Wynne Jones"}, {"title": "Circe", "author": "Madeline Miller"}, {"title": "The Invisible Life of Addie LaRue", "author": "V.E. Schwab"}, {"title": "Mexican Gothic", "author": "Silvia Moreno-Garcia"}, {"title": "The Ten Thousand Doors of January", "author": "Alix E. Harrow"}, {"title": "The Goblin Emperor", "author": "Katherine Addison"}, {"title": "The Priory of the Orange Tree", "author": "Samantha Shannon"}, {"title": "Uprooted", "author": "Naomi Novik"}, {"title": "The Bear and the Nightingale", "author": "Katherine Arden"}, {"title": "The City of Brass", "author": "S.A. Chakraborty"} ] state["retrieved_books"] = books state["reasoning"].append(f"Total books in state: {len(books)}") return state def call_google_books_mcp(title: str, author: str = "") -> Dict[str, Any]: """ Call the Google Books MCP server via Gradio MCP endpoint. Args: title: Book title author: Book author (optional) Returns: Book metadata dict or None if not found """ import requests try: # Gradio MCP endpoint (Streamable HTTP transport) mcp_url = f"{GOOGLE_BOOKS_MCP_URL}/gradio_api/mcp/" # MCP uses JSON-RPC style calls payload = { "jsonrpc": "2.0", "method": "tools/call", "params": { "name": "google_books_mcp_search_book_by_title_author", "arguments": { "title": title, "author": author } }, "id": 1 } response = requests.post( mcp_url, json=payload, headers={ "Content-Type": "application/json", "Accept": "application/json, text/event-stream" }, timeout=30 ) if response.status_code != 200: print(f"[DEBUG] Google Books MCP failed: {response.status_code} - {response.text[:200]}") return None # Parse SSE response for line in response.text.split('\n'): if line.startswith('data: '): try: data = json.loads(line[6:]) if "result" in data: result = data["result"] if isinstance(result, dict): # Check if it's a direct book response if "success" in result and "book" in result: if result.get("success") and result.get("book"): return result["book"] # Check if it's a content array response elif "content" in result: for content_item in result["content"]: if content_item.get("type") == "text": text_content = content_item.get("text", "") if text_content.strip(): try: book_data = json.loads(text_content) if book_data.get("success") and book_data.get("found"): return book_data.get("book") except json.JSONDecodeError: continue return result except json.JSONDecodeError: continue return None except Exception as e: print(f"[DEBUG] Google Books MCP error: {e}") return None def fetch_book_metadata(state: AgentState) -> AgentState: """Node: Fetch metadata for retrieved books via Google Books API""" print(f"[DEBUG AGENT] fetch_book_metadata node started with {len(state.get('retrieved_books', []))} books") enriched_books = [] skipped_books = [] state["reasoning"].append(f"📖 Fetching metadata from Google Books (need {NUM_BOOKS_TO_RETRIEVE} with descriptions)...") for book in state["retrieved_books"]: # Stop once we have enough books with valid descriptions if len(enriched_books) >= NUM_BOOKS_TO_RETRIEVE: print(f"[DEBUG] Reached target of {NUM_BOOKS_TO_RETRIEVE} books, stopping") break try: # Use Google Books MCP server metadata = call_google_books_mcp(book['title'], book['author']) if metadata and metadata.get("title"): description = metadata.get("description", "") # FILTER: Skip books without meaningful descriptions if not description or len(description.strip()) < 50: skipped_books.append(book['title']) print(f"[DEBUG] Skipping '{book['title']}' - no/short description ({len(description.strip()) if description else 0} chars)") continue # Format authors as string authors = metadata.get("authors", []) author_str = ", ".join(authors) if isinstance(authors, list) else authors or book["author"] enriched_books.append({ "title": metadata.get("title", book["title"]), "author": author_str, "description": description, "cover_url": metadata.get("thumbnail"), "isbn": metadata.get("isbn"), "published_year": metadata.get("published_date", "")[:4] if metadata.get("published_date") else None, "page_count": metadata.get("page_count"), "categories": metadata.get("categories", []), "preview_link": metadata.get("preview_link"), "info_link": metadata.get("info_link") }) print(f"[DEBUG] Found metadata for: {book['title']} ({len(description)} chars) [{len(enriched_books)}/{NUM_BOOKS_TO_RETRIEVE}]") else: # No results found - skip skipped_books.append(book['title']) print(f"[DEBUG] Skipping '{book['title']}' - no Google Books results") except Exception as e: # On any error, skip the book skipped_books.append(book['title']) state["reasoning"].append(f"Error fetching metadata for '{book['title']}': {str(e)}") state["books_with_metadata"] = enriched_books if skipped_books: state["reasoning"].append(f"⚠️ Skipped {len(skipped_books)} books without descriptions") state["reasoning"].append(f"✅ Found {len(enriched_books)}/{NUM_BOOKS_TO_RETRIEVE} books with full metadata") return state def _generate_narrowing_question(state: AgentState, question_num: int) -> tuple: """Helper: Generate a narrowing question""" from prompts import NARROWING_QUESTION_GENERATOR books_summary_parts = [] for i, b in enumerate(state["books_with_metadata"], 1): desc = b.get('description', 'No description') cats = ', '.join(b.get('categories', [])) if b.get('categories') else 'Uncategorized' books_summary_parts.append(f"Book {i}: {b['title']} by {b['author']}\n Categories: {cats}\n Description: {desc}") books_summary = "\n\n".join(books_summary_parts) vibe_context = f"Feels like: {state['feels_like']}\nAesthetics: {', '.join(state['aesthetic_genre_keywords'])}\nMood: {', '.join(state['mood_atmosphere'])}\nThemes: {', '.join(state['core_themes'])}" is_last = question_num >= 2 question_context = f"This is question {question_num} of 2." + (" THIS IS THE LAST QUESTION - make it count!" if is_last else "") user_prompt = f"Books to narrow down:\n{books_summary}\n\nVibe:\n{vibe_context}\n\nPrevious preferences: {json.dumps(state.get('user_preferences', {}), indent=2)}\n\n{question_context}\n\nGenerate an either/or question:" messages = [ {"role": "system", "content": NARROWING_QUESTION_GENERATOR}, {"role": "user", "content": user_prompt} ] return call_llm(messages, temperature=0.8, model=FAST_MODEL, include_reasoning=True) def generate_question_1(state: AgentState) -> AgentState: """Node: Generate Q1 and add to messages""" print(f"[DEBUG AGENT] generate_question_1") question, reasoning = _generate_narrowing_question(state, 1) state["narrowing_questions_asked"] = 1 state["q1_question"] = question state["reasoning"].append(f"🧠 REASONING (Narrowing Question #1):\n{reasoning}\n\n→ Question: {question}") assistant_message = f"To help me find the perfect match:\n\n{question}" print(f"[DEBUG AGENT] Q1: {question[:60]}...") state["messages"].append({"role": "assistant", "content": assistant_message}) return state def wait_for_answer_1(state: AgentState) -> AgentState: """Node: Wait for user's answer to Q1""" print(f"[DEBUG AGENT] wait_for_answer_1") user_answer = interrupt("Waiting for Q1 answer") if user_answer: state["messages"].append({"role": "user", "content": user_answer}) state["user_preferences"]["q1"] = { "question": state.get("q1_question", ""), "answer": user_answer } print(f"[DEBUG AGENT] Q1 answered: {user_answer}") return state def generate_question_2(state: AgentState) -> AgentState: """Node: Generate Q2 and add to messages""" print(f"[DEBUG AGENT] generate_question_2") question, reasoning = _generate_narrowing_question(state, 2) state["narrowing_questions_asked"] = 2 state["q2_question"] = question state["reasoning"].append(f"🧠 REASONING (Narrowing Question #2):\n{reasoning}\n\n→ Question: {question}") assistant_message = f"To help me find the perfect match:\n\n{question}" print(f"[DEBUG AGENT] Q2: {question[:60]}...") state["messages"].append({"role": "assistant", "content": assistant_message}) return state def wait_for_answer_2(state: AgentState) -> AgentState: """Node: Wait for user's answer to Q2""" print(f"[DEBUG AGENT] wait_for_answer_2") user_answer = interrupt("Waiting for Q2 answer") if user_answer: state["messages"].append({"role": "user", "content": user_answer}) state["user_preferences"]["q2"] = { "question": state.get("q2_question", ""), "answer": user_answer } print(f"[DEBUG AGENT] Q2 answered: {user_answer}") return state def check_narrowing_complete(state: AgentState) -> Literal["ask_more", "finalize"]: """Conditional edge: Check if we've asked all 2 narrowing questions""" questions_asked = state.get("narrowing_questions_asked", 0) if questions_asked >= 2: return "finalize" return "ask_more" def finalize_books(state: AgentState) -> AgentState: """Node: Use reasoning to select final 3 books based on vibe and preferences""" print(f"[DEBUG AGENT] finalize_books node started") print(f"[DEBUG AGENT] books_with_metadata count: {len(state.get('books_with_metadata', []))}") from prompts import get_book_finalizer_prompt # Build detailed book summary with full descriptions - no truncation books_summary_parts = [] for i, b in enumerate(state["books_with_metadata"]): desc = b.get('description', 'No description available') cats = ', '.join(b.get('categories', [])) if b.get('categories') else 'Uncategorized' books_summary_parts.append(f"{i+1}. {b['title']} by {b['author']}\n Categories: {cats}\n Description: {desc}") books_summary = "\n\n".join(books_summary_parts) prefs_summary = json.dumps(state.get("user_preferences", {}), indent=2) vibe_context = f"Feels like: {state['feels_like']}\nAesthetics: {', '.join(state['aesthetic_genre_keywords'])}\nMood: {', '.join(state['mood_atmosphere'])}\nThemes: {', '.join(state['core_themes'])}\nTropes: {', '.join(state['tropes'])}" user_prompt = f"Vibe:\n{vibe_context}\n\nCandidate Books:\n{books_summary}\n\nUser Preferences (from Q&A):\n{prefs_summary}\n\nSelect the {NUM_FINAL_BOOKS} best matches (return only JSON array):" messages = [ {"role": "system", "content": get_book_finalizer_prompt(NUM_FINAL_BOOKS)}, {"role": "user", "content": user_prompt} ] print(f"[DEBUG AGENT] finalize_books user_prompt:\n{user_prompt}") # Use reasoning model for book selection - this is a complex decision # Increase max_tokens since we're sending full book descriptions selection_response, reasoning = call_llm(messages, temperature=0.3, model=REASONING_MODEL, include_reasoning=True, max_tokens=5000) # Log reasoning even if empty state["reasoning"].append(f"🧠 REASONING (Book Selection):\n{reasoning or 'No reasoning provided'}") # Parse the JSON response - check both content and reasoning for the array try: import re # First try to find JSON array in the response content json_match = re.search(r'\[([\d,\s]+)\]', selection_response) # If not found in content, try to find it in reasoning (some models put answer there) if not json_match and reasoning: json_match = re.search(r'\[([\d,\s]+)\]', reasoning) if json_match: print(f"[DEBUG AGENT] Found JSON in reasoning instead of content") if json_match: indices = json.loads(json_match.group(0)) selected_books = [state["books_with_metadata"][i-1] for i in indices if 0 < i <= len(state["books_with_metadata"])][:NUM_FINAL_BOOKS] else: # Fallback to first 3 books print(f"[DEBUG AGENT] No JSON array found, using first 3 books") selected_books = state["books_with_metadata"][:NUM_FINAL_BOOKS] except Exception as e: state["reasoning"].append(f"❌ Failed to parse book selection: {e}. Using first 3 books.") selected_books = state["books_with_metadata"][:NUM_FINAL_BOOKS] state["final_books"] = selected_books state["reasoning"].append(f"🧠 REASONING (Book Selection):\n{reasoning}\n\n→ Selected: {[b['title'] for b in selected_books]}") return state def generate_soundtrack(state: AgentState) -> AgentState: """Node: Generate ambient soundtrack using ElevenLabs Music API""" print(f"[DEBUG AGENT] generate_soundtrack node started") import requests import tempfile ELEVENLABS_API_KEY = os.getenv("ELEVENLABS_API_KEY") print(f"[DEBUG AGENT] ELEVENLABS_API_KEY present: {bool(ELEVENLABS_API_KEY)}") if not ELEVENLABS_API_KEY: print(f"[DEBUG AGENT] No ELEVENLABS_API_KEY - skipping") state["reasoning"].append("⚠️ ELEVENLABS_API_KEY not set - skipping soundtrack generation") state["soundtrack_url"] = "" return state try: # Build vibe context for music prompt generation vibe_context = { "feels_like": state["feels_like"], "mood_atmosphere": state["mood_atmosphere"], "aesthetic_genre_keywords": state["aesthetic_genre_keywords"], "core_themes": state["core_themes"], "tropes": state["tropes"] } print(f"[DEBUG AGENT] vibe_context built: {list(vibe_context.keys())}") # Use LLM to generate music prompt from vibe context from prompts import MUSIC_PROMPT_GENERATION messages = [ {"role": "system", "content": MUSIC_PROMPT_GENERATION}, {"role": "user", "content": f"Generate a music prompt based on this vibe:\n{json.dumps(vibe_context, indent=2)}"} ] print(f"[DEBUG AGENT] Calling LLM for music prompt...") music_prompt, reasoning = call_llm(messages, temperature=0.7, model=FAST_MODEL, include_reasoning=True) print(f"[DEBUG AGENT] Music prompt generated: {music_prompt[:100] if music_prompt else 'None'}...") state["reasoning"].append(f"🎵 Music prompt: {music_prompt}") # Call ElevenLabs Music API directly print(f"[DEBUG AGENT] Calling ElevenLabs Music API...") state["reasoning"].append(f"🎵 Calling ElevenLabs Music API...") response = requests.post( "https://api.elevenlabs.io/v1/music", headers={ "xi-api-key": ELEVENLABS_API_KEY, "Content-Type": "application/json" }, json={ "prompt": music_prompt, "music_length_ms": 90000, # 1:30 minute "force_instrumental": True # No vocals, just ambient music }, timeout=120 # Music generation can take a while ) print(f"[DEBUG AGENT] ElevenLabs response status: {response.status_code}") if response.status_code == 200: print(f"[DEBUG AGENT] Success! Response size: {len(response.content)} bytes") # Save the audio data to a temp file temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.mp3') temp_file.write(response.content) temp_file.close() print(f"[DEBUG AGENT] Saved to temp file: {temp_file.name}") state["soundtrack_url"] = temp_file.name state["reasoning"].append(f"✅ Generated soundtrack successfully ({len(response.content)} bytes)") else: print(f"[DEBUG AGENT] ElevenLabs API error: {response.status_code} - {response.text[:500]}") state["reasoning"].append(f"❌ ElevenLabs API error: {response.status_code} - {response.text[:200]}") state["soundtrack_url"] = "" except Exception as e: import traceback print(f"[DEBUG AGENT] Exception in generate_soundtrack: {e}") traceback.print_exc() state["reasoning"].append(f"❌ Failed to generate soundtrack: {e}") state["soundtrack_url"] = "" print(f"[DEBUG AGENT] generate_soundtrack finished, soundtrack_url: {state.get('soundtrack_url', 'not set')}") return state def present_final_results(state: AgentState) -> AgentState: """Node: Format and present final results to user""" # Format books for display books_text = "Here are your personalized book recommendations:\n\n" for i, book in enumerate(state["final_books"], 1): books_text += f"{i}. **{book['title']}** by {book['author']}\n" state["messages"].append({ "role": "assistant", "content": books_text + f"\n\nI'm also generating a soundtrack that matches your vibe! Scroll down for all the goodies ⬇️" }) state["reasoning"].append("Presented final results to user") return state # ============================================================================ # GRAPH CONSTRUCTION # ============================================================================ def create_agent_graph(): """Create and compile the LangGraph workflow with interrupts for user input""" from langgraph.checkpoint.memory import MemorySaver # Initialize graph workflow = StateGraph(AgentState) # Add nodes workflow.add_node("generate_initial_vibe", generate_initial_vibe) workflow.add_node("refine_vibe", refine_vibe) workflow.add_node("retrieve_books", retrieve_books) workflow.add_node("fetch_metadata", fetch_book_metadata) workflow.add_node("generate_q1", generate_question_1) workflow.add_node("wait_a1", wait_for_answer_1) workflow.add_node("generate_q2", generate_question_2) workflow.add_node("wait_a2", wait_for_answer_2) workflow.add_node("finalize_books", finalize_books) workflow.add_node("generate_soundtrack", generate_soundtrack) workflow.add_node("present_results", present_final_results) # Set entry point workflow.set_entry_point("generate_initial_vibe") # After initial vibe, check if user is satisfied or wants refinement workflow.add_conditional_edges( "generate_initial_vibe", check_vibe_satisfaction, { "refine": "refine_vibe", "retrieve": "retrieve_books" } ) # After refinement, check again if user is satisfied workflow.add_conditional_edges( "refine_vibe", check_vibe_satisfaction, { "refine": "refine_vibe", "retrieve": "retrieve_books" } ) # Sequential: retrieve -> fetch -> generate Q1 -> wait A1 -> generate Q2 -> wait A2 -> finalize workflow.add_edge("retrieve_books", "fetch_metadata") workflow.add_edge("fetch_metadata", "generate_q1") workflow.add_edge("generate_q1", "wait_a1") workflow.add_edge("wait_a1", "generate_q2") workflow.add_edge("generate_q2", "wait_a2") workflow.add_edge("wait_a2", "finalize_books") # Sequential: finalize -> soundtrack -> present workflow.add_edge("finalize_books", "generate_soundtrack") workflow.add_edge("generate_soundtrack", "present_results") workflow.add_edge("present_results", END) # Compile with checkpointer for state persistence memory = MemorySaver() return workflow.compile(checkpointer=memory) # ============================================================================ # MAIN INTERFACE # ============================================================================ # Global graph instance with persistent checkpointer _GRAPH_INSTANCE = None def get_graph(): """Get or create the compiled graph with checkpointer""" global _GRAPH_INSTANCE if _GRAPH_INSTANCE is None: print(f"[DEBUG AGENT] Creating NEW graph instance!") _GRAPH_INSTANCE = create_agent_graph() else: print(f"[DEBUG AGENT] Reusing existing graph instance") return _GRAPH_INSTANCE def reset_agent(): """Reset the agent by clearing the graph instance""" global _GRAPH_INSTANCE _GRAPH_INSTANCE = None def run_agent(images: List[str], user_message: str = None, thread_id: str = "main"): """ Main interface to run the agent with interrupt-based human-in-the-loop Args: images: List of image URLs/base64 for initial upload user_message: User's message (for resuming after interrupt) thread_id: Unique identifier for the user session (required for multi-user support) Returns: Updated state with agent's response """ from langgraph.types import Command graph = get_graph() thread_config = {"configurable": {"thread_id": thread_id}} # Initialize state if new conversation (images provided) if images and len(images) > 0: initial_state = AgentState( images=images, messages=[], aesthetic_genre_keywords=[], mood_atmosphere=[], core_themes=[], tropes=[], feels_like="", vibe_refinement_count=0, retrieved_books=[], books_with_metadata=[], q1_question="", q2_question="", user_preferences={}, final_books=[], soundtrack_url="", reasoning=[] ) # Start the graph - it will stop at first interrupt() result = graph.invoke(initial_state, thread_config) return result # Resume with user's message if user_message: # Check current state before resuming current_state = graph.get_state(thread_config) print(f"[DEBUG AGENT] State BEFORE resume:") print(f"[DEBUG AGENT] messages count: {len(current_state.values.get('messages', []))}") for i, m in enumerate(current_state.values.get('messages', [])): print(f"[DEBUG AGENT] msg[{i}]: {m.get('role')} - {m.get('content', '')[:60]}...") print(f"[DEBUG AGENT] q1_question: '{current_state.values.get('q1_question', '')[:50] if current_state.values.get('q1_question') else 'EMPTY'}'") # Resume from the last interrupt; the value passed to Command(resume=...) # is what the corresponding interrupt(...) call will return inside the node. print(f"[DEBUG AGENT] Resuming graph with user_message: {user_message[:50]}...") result = graph.invoke(Command(resume=user_message), thread_config) print(f"[DEBUG AGENT] graph.invoke returned: {type(result)}, keys: {list(result.keys()) if hasattr(result, 'keys') else 'N/A'}") print(f"[DEBUG AGENT] result has {len(result.get('messages', []))} messages") # Remove __interrupt__ key if present before returning if "__interrupt__" in result: result = {k: v for k, v in result.items() if k != "__interrupt__"} return result return None