Spaces:
Running
Running
File size: 42,199 Bytes
e422038 fce43bd e422038 c2d7928 e422038 c2d7928 e422038 c2d7928 e422038 c2d7928 e422038 c2d7928 e422038 c2d7928 e422038 c2d7928 e422038 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 |
"""
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 <think>...</think> tags from content
import re
think_match = re.match(r'<think>(.*?)</think>(.*)', 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
|