""" graph_demo.py — 웹 데모용 그래프 어댑터. backend/app/pipeline/graph_main.py를 import해서 사용하되, 웹 챗봇 환경에 맞게 다음 두 가지를 처리한다: 1. interrupt_before 우회 - 원본은 process_slot_answer 등 3개 노드 직전에 멈춰서 인간 입력을 기다리는 구조. - 웹에서는 매 요청이 (유저 메시지) → (AI 응답) 한 쌍으로 완결돼야 함. - 따라서 stream API + Command(resume) 패턴 대신, 명시적인 상태 inject로 처리. 2. 세션 누적 (localStorage 기반) - 백엔드는 stateless하게 동작. - 클라이언트가 이전 세션의 UserProfile + book_experiences + summary를 보내주면 그것을 initial_state에 inject한 뒤 그래프 실행. - 그래프 종료 후 업데이트된 프로파일을 다시 클라이언트에 반환. 이렇게 하면 ChromaDB 영속성 없이도 "세션을 넘나드는 프로파일 누적"의 시각적 효과를 정확히 재현할 수 있음. """ from __future__ import annotations import os import uuid from typing import Any, Optional from langchain_core.messages import HumanMessage, AIMessage from app.pipeline import graph_main from app.state.state import ( UserProfile, BookExperience, Phase, SLOT_NAMES, ) # ──────────────────────────────────────────────────────────────────────────── # Langfuse 트레이싱 (환경변수가 있을 때만 활성화) # ──────────────────────────────────────────────────────────────────────────── _langfuse_handler = None try: if os.environ.get("LANGFUSE_PUBLIC_KEY") and os.environ.get("LANGFUSE_SECRET_KEY"): from langfuse.langchain import CallbackHandler _langfuse_handler = CallbackHandler( public_key=os.environ["LANGFUSE_PUBLIC_KEY"], secret_key=os.environ["LANGFUSE_SECRET_KEY"], host=os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com"), ) print("[graph_demo] Langfuse callback handler initialized") else: print("[graph_demo] Langfuse env vars not set — tracing disabled") except Exception as e: print(f"[graph_demo] Langfuse init failed (continuing without tracing): {e}") _langfuse_handler = None # ──────────────────────────────────────────────────────────────────────────── # 유저별 그래프 인스턴스 관리 # ──────────────────────────────────────────────────────────────────────────── # 유저 ID별로 ChromaDB 경로를 분리해 세션 기억을 격리한다. # HF Spaces 무료 플랜은 슬립 후 /tmp가 초기화되므로 재시작 시 데이터가 초기화됨. _DEMO_CHROMA_BASE = os.environ.get("DEMO_CHROMA_BASE", "/tmp/peekabook_demo_chroma") _graphs: dict[str, Any] = {} def _get_graph(user_id: str): if user_id not in _graphs: chroma_path = os.path.join(_DEMO_CHROMA_BASE, user_id) _graphs[user_id] = graph_main.create_app( chroma_db_path=chroma_path, use_genre_filter=True, ) return _graphs[user_id] # ──────────────────────────────────────────────────────────────────────────── # 프로파일 직렬화 / 역직렬화 헬퍼 # ──────────────────────────────────────────────────────────────────────────── def serialize_profile_payload(state: dict) -> dict: """그래프 실행 후 클라이언트에 돌려줄 누적 가능한 프로파일 페이로드. 이 dict가 그대로 localStorage에 JSON으로 저장되고, 다음 세션 시작 시 그대로 받아서 deserialize_profile_payload로 복원된다. """ profile: UserProfile = state.get("user_profile") or UserProfile() book_experiences: list[BookExperience] = state.get("book_experiences") or [] return { "user_profile": profile.to_display_dict(), # state.py에 이미 정의됨 "book_experiences": [ { "book_name": be.book_name, "impression": be.impression, "context": be.context, } for be in book_experiences ], "summary": state.get("summary", "") or "", "reflection": state.get("reflection", "") or "", } def deserialize_profile_payload(payload: Optional[dict]) -> dict: """클라이언트가 보낸 페이로드를 그래프 initial_state 형태로 변환.""" if not payload: return { "user_profile": UserProfile(), "book_experiences": [], "summary": "", "reflection": "", } profile = UserProfile.from_dict(payload.get("user_profile") or {}) book_experiences = [ BookExperience( book_name=be.get("book_name", ""), impression=be.get("impression", ""), context=be.get("context", ""), ) for be in (payload.get("book_experiences") or []) ] return { "user_profile": profile, "book_experiences": book_experiences, "summary": payload.get("summary", "") or "", "reflection": payload.get("reflection", "") or "", } # ──────────────────────────────────────────────────────────────────────────── # 핵심: 한 턴 실행 # ──────────────────────────────────────────────────────────────────────────── async def run_one_turn( user_message: str, session_id: str, user_id: str = "default", prior_profile_payload: Optional[dict] = None, thread_id: Optional[str] = None, ) -> dict: """유저 메시지 한 번에 대해 그래프를 한 턴 실행하고 결과를 반환. Parameters ---------- user_message : 유저가 방금 입력한 메시지. session_id : 현재 세션 ID (클라이언트가 발급/관리). prior_profile_payload : 이전 세션에서 누적된 프로파일 (없으면 빈 상태로 시작). thread_id : LangGraph checkpoint thread (같은 브라우저 세션 내 멀티턴용). 클라이언트가 첫 메시지에서 None을 보내면 새로 만들어 응답에 포함. Returns ------- { "ai_response": str, # 사용자에게 보여줄 AI 답변 "recommendations": list, # 책 추천 결과 (있을 때) "phase": str, # 현재 phase (UI에서 단계 표시용) "profile_payload": dict, # localStorage 저장용 누적 페이로드 "thread_id": str, # 같은 세션 내 다음 호출에 그대로 전달 "session_done": bool, # 그래프가 END에 도달했는지 } """ if thread_id is None: thread_id = str(uuid.uuid4()) config: dict = {"configurable": {"thread_id": thread_id}} if _langfuse_handler is not None: config["callbacks"] = [_langfuse_handler] # 데모 식별용 메타데이터 — Langfuse 대시보드에서 필터링 가능 config["metadata"] = { "user_id": user_id, "session_id": session_id, "langfuse_session_id": session_id, "langfuse_user_id": user_id, } compiled_graph = _get_graph(user_id) # 현재 checkpoint 상태를 확인 — 같은 thread에서 이어지는 호출인지 첫 호출인지 판단 current_snapshot = compiled_graph.get_state(config) # 현재 checkpoint 상태를 확인 — 같은 thread에서 이어지는 호출인지 첫 호출인지 판단 current_snapshot = compiled_graph.get_state(config) is_first_turn_in_thread = ( current_snapshot is None or not current_snapshot.values ) if is_first_turn_in_thread: # 첫 턴: initial_state에 prior profile inject 후 시작 prior = deserialize_profile_payload(prior_profile_payload) initial = dict(graph_main.initial_state) # 얕은 복사 initial["session_id"] = session_id initial["messages"] = [HumanMessage(content=user_message)] # 세션 2부터는 슬롯 빈 상태로 시작 initial["user_profile"] = UserProfile() initial["book_experiences"] = [] # 이전 세션 요약만 컨텍스트로 주입 if prior.get("summary"): initial["summary"] = f"[이전 세션 요약] {prior['summary']}" if prior.get("reflection"): initial["reflection"] = f"[이전 세션 인사이트] {prior['reflection']}" # 누적 프로파일이 있으면 슬롯필링 일부를 건너뛰고 바로 추천으로 가는 것도 # 가능하지만, 데모에서는 "기억하고 있음을 보여주는" 효과가 더 중요하므로 # 정상 흐름을 유지하되 generate_slot_question 노드가 이미 채워진 슬롯을 # 인지하도록 둔다 (그래프 로직이 filled_slots()를 이미 확인함). await compiled_graph.ainvoke(initial, config=config) else: # 같은 thread의 후속 턴: interrupt_before로 멈춰있는 상태에 유저 메시지만 추가 compiled_graph.update_state( config, {"messages": [HumanMessage(content=user_message)]}, ) await compiled_graph.ainvoke(None, config=config) # 실행 후 상태 조회 snapshot = compiled_graph.get_state(config) state_values = snapshot.values # next가 비어있으면 그래프가 END에 도달한 것 session_done = not snapshot.next # 최종 응답은 messages 마지막 항목에서 읽음 (rag_llm/api_tool_calling이 messages에 기록) # 프로파일링 중간 턴은 ai_response를 fallback으로 사용 messages = state_values.get("messages", []) last_ai_message = next( (m.content for m in reversed(messages) if isinstance(m, AIMessage)), state_values.get("ai_response", ""), ) return { "ai_response": last_ai_message, "recommendations": state_values.get("recommendations", []), "phase": _phase_to_str(state_values.get("phase")), "profile_payload": serialize_profile_payload(state_values), "thread_id": thread_id, "session_done": session_done, } def _phase_to_str(phase) -> str: """Phase enum 또는 문자열을 안전하게 문자열로 변환.""" if phase is None: return "" if isinstance(phase, Phase): return phase.value return str(phase) # ──────────────────────────────────────────────────────────────────────────── # 헬스체크 — HF Spaces 슬립 방지용 # ──────────────────────────────────────────────────────────────────────────── def healthcheck() -> dict: """그래프가 정상 컴파일됐는지만 가볍게 확인.""" return { "graph_compiled": True, "slot_names": SLOT_NAMES, }