update langgraph + gradio

app_tool.py

@@ -0,0 +1,420 @@
+import gradio as gr
+from langgraph.graph import StateGraph, END
+from typing import TypedDict, Annotated
+import operator
+from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
+from langchain_openai import ChatOpenAI
+import asyncio
+import time
+import os
+
+# 상태 정의
+class ChatState(TypedDict):
+    messages: Annotated[list[BaseMessage], operator.add]
+    current_response: str
+    agent_type: str
+    context: dict
+    step_info: str
+
+# LLM 초기화
+llm = ChatOpenAI(
+    model="gpt-3.5-turbo",
+    temperature=0.7,
+    # api_key="your-openai-api-key"  # 실제 API 키를 설정하세요
+)
+
+def step1_analyzer_node(state: ChatState) -> ChatState:
+    """1단계: 메시지 분석 노드"""
+    last_message = state["messages"][-1].content
+
+    # 분석 시뮬레이션 (실제로는 더 복잡한 로직)
+    analysis = {
+        "length": len(last_message),
+        "has_question": "?" in last_message,
+        "language": "korean" if any(ord(c) > 127 for c in last_message) else "english",
+        "sentiment": "positive" if any(word in last_message.lower() for word in ["좋", "감사", "고마워", "thanks", "good"]) else "neutral"
+    }
+
+    return {
+        "messages": [],
+        "current_response": "",
+        "agent_type": "",
+        "context": {"analysis": analysis},
+        "step_info": f"📊 **1단계 완료** - 메시지 분석\n- 길이: {analysis['length']}자\n- 질문 포함: {'예' if analysis['has_question'] else '아니오'}\n- 언어: {analysis['language']}\n- 감정: {analysis['sentiment']}"
+    }
+
+def step2_classifier_node(state: ChatState) -> ChatState:
+    """2단계: 의도 분류 노드"""
+    last_message = state["messages"][-1].content.lower()
+    analysis = state["context"]["analysis"]
+
+    # 의도 분류 로직
+    if any(word in last_message for word in ["코드", "프로그래밍", "python", "개발", "함수", "클래스"]):
+        agent_type = "programmer"
+        confidence = 0.9
+    elif any(word in last_message for word in ["날씨", "뉴스", "정보", "검색", "찾아"]):
+        agent_type = "informer"
+        confidence = 0.8
+    elif any(word in last_message for word in ["계산", "수학", "더하기", "빼기", "곱하기", "나누기"]):
+        agent_type = "calculator"
+        confidence = 0.95
+    elif any(word in last_message for word in ["창작", "시", "소설", "이야기", "글"]):
+        agent_type = "creative"
+        confidence = 0.85
+    else:
+        agent_type = "general"
+        confidence = 0.7
+
+    context = state["context"]
+    context["classification"] = {
+        "agent_type": agent_type,
+        "confidence": confidence
+    }
+
+    return {
+        "messages": [],
+        "current_response": "",
+        "agent_type": agent_type,
+        "context": context,
+        "step_info": f"🎯 **2단계 완료** - 의도 분류\n- 분류 결과: {agent_type}\n- 신뢰도: {confidence:.1%}\n- 다음 단계: {'전문 처리' if confidence > 0.8 else '일반 처리'}"
+    }
+
+def step3_context_enricher_node(state: ChatState) -> ChatState:
+    """3단계: 컨텍스트 강화 노드"""
+    agent_type = state["agent_type"]
+
+    # 에이전트 타입별 컨텍스트 강화
+    enriched_context = {
+        "programmer": {
+            "system_prompt": "당신은 경험이 풍부한 시니어 개발자입니다. 코드 예시와 함께 명확하고 실용적인 답변을 제공하세요.",
+            "tools": ["코드 실행", "문서 검색", "베스트 프랙티스"],
+            "style": "기술적이고 정확한"
+        },
+        "informer": {
+            "system_prompt": "당신은 정보 전문가입니다. 정확하고 최신의 정보를 구조화된 형태로 제공하세요.",
+            "tools": ["웹 검색", "팩트 체크", "데이터 분석"],
+            "style": "객관적이고 상세한"
+        },
+        "calculator": {
+            "system_prompt": "당신은 수학 전문가입니다. 계산 과정을 단계별로 설명하고 정확한 답을 제공하세요.",
+            "tools": ["수식 계산", "그래프 생성", "통계 분석"],
+            "style": "논리적이고 체계적인"
+        },
+        "creative": {
+            "system_prompt": "당신은 창작 전문가입니다. 상상력이 풍부하고 감성적인 콘텐츠를 제작하세요.",
+            "tools": ["스토리텔링", "시각적 묘사", "감정 표현"],
+            "style": "창의적이고 감성적인"
+        },
+        "general": {
+            "system_prompt": "당신은 친근하고 도움이 되는 AI 어시스턴트입니다. 자연스럽고 이해하기 쉬운 답변을 제공하세요.",
+            "tools": ["일반 대화", "정보 제공", "문제 해결"],
+            "style": "친근하고 자연스러운"
+        }
+    }
+
+    context = state["context"]
+    context["enriched"] = enriched_context.get(agent_type, enriched_context["general"])
+
+    return {
+        "messages": [],
+        "current_response": "",
+        "agent_type": agent_type,
+        "context": context,
+        "step_info": f"🔧 **3단계 완료** - 컨텍스트 강화\n- 에이전트: {agent_type}\n- 스타일: {context['enriched']['style']}\n- 활용 도구: {', '.join(context['enriched']['tools'][:2])}"
+    }
+
+def step4_response_generator_node(state: ChatState) -> ChatState:
+    """4단계: 응답 생성 노드"""
+    enriched_context = state["context"]["enriched"]
+    system_prompt = enriched_context["system_prompt"]
+
+    # 시스템 프롬프트와 함께 메시지 구성
+    messages = [HumanMessage(content=system_prompt)] + state["messages"]
+
+    try:
+        response = llm.invoke(messages)
+
+        # 에이전트 타입에 따른 아이콘 설정
+        icons = {
+            "programmer": "💻",
+            "informer": "📚",
+            "calculator": "🔢",
+            "creative": "🎨",
+            "general": "💬"
+        }
+
+        icon = icons.get(state["agent_type"], "💬")
+        final_response = f"{icon} **[{state['agent_type'].upper()}]**\n\n{response.content}"
+
+        return {
+            "messages": [response],
+            "current_response": final_response,
+            "agent_type": state["agent_type"],
+            "context": state["context"],
+            "step_info": f"✅ **4단계 완료** - 응답 생성\n- 최종 응답 준비됨\n- 응답 길이: {len(response.content)}자"
+        }
+
+    except Exception as e:
+        error_msg = f"❌ 응답 생성 중 오류 발생: {str(e)}"
+        return {
+            "messages": [AIMessage(content=error_msg)],
+            "current_response": error_msg,
+            "agent_type": state["agent_type"],
+            "context": state["context"],
+            "step_info": f"❌ **4단계 실패** - 오류 발생\n- 오류: {str(e)}"
+        }
+
+def should_continue_to_classifier(state: ChatState) -> str:
+    return "classifier"
+
+def should_continue_to_enricher(state: ChatState) -> str:
+    return "enricher"
+
+def should_continue_to_generator(state: ChatState) -> str:
+    return "generator"
+
+def should_end(state: ChatState) -> str:
+    return END
+
+# LangGraph 워크플로우 생성
+def create_enhanced_workflow():
+    workflow = StateGraph(ChatState)
+
+    # 4개 노드 추가
+    workflow.add_node("analyzer", step1_analyzer_node)
+    workflow.add_node("classifier", step2_classifier_node)
+    workflow.add_node("enricher", step3_context_enricher_node)
+    workflow.add_node("generator", step4_response_generator_node)
+
+    # 시작점 설정
+    workflow.set_entry_point("analyzer")
+
+    # 순차적 엣지 추가
+    workflow.add_conditional_edges(
+        "analyzer",
+        should_continue_to_classifier,
+        {"classifier": "classifier"}
+    )
+
+    workflow.add_conditional_edges(
+        "classifier",
+        should_continue_to_enricher,
+        {"enricher": "enricher"}
+    )
+
+    workflow.add_conditional_edges(
+        "enricher",
+        should_continue_to_generator,
+        {"generator": "generator"}
+    )
+
+    workflow.add_conditional_edges(
+        "generator",
+        should_end,
+        {END: END}
+    )
+
+    return workflow.compile()
+
+# 글로벌 워크플로우 인스턴스
+enhanced_workflow = create_enhanced_workflow()
+
+def stream_chatbot_response(message, history):
+    """각 단계를 누적해서 실시간 표시"""
+    if not message.strip():
+        yield "", history
+        return
+
+    # 메시지 히스토리를 LangChain 메시지로 변환
+    messages = []
+    for human_msg, ai_msg in history:
+        if human_msg and not "📊" in human_msg and not "🎯" in human_msg:
+            messages.append(HumanMessage(content=human_msg))
+        if ai_msg and not "📊" in ai_msg and not "🎯" in ai_msg and not "⚡" in ai_msg:
+            messages.append(AIMessage(content=ai_msg))
+
+    # 현재 메시지 추가
+    messages.append(HumanMessage(content=message))
+
+    try:
+        # 초기 상태 설정
+        initial_state = {
+            "messages": messages,
+            "current_response": "",
+            "agent_type": "",
+            "context": {},
+            "step_info": ""
+        }
+
+        # 워크플로우를 스트림으로 실행
+        current_history = history.copy()
+
+        # 누적될 단계 정보들
+        accumulated_steps = []
+        accumulated_steps.append("🔄 **AI 처리 과정 시작**\n")
+
+        current_history.append((message, "\n".join(accumulated_steps)))
+        yield "", current_history
+        time.sleep(0.3)
+
+        # 각 노드를 순차적으로 실행하면서 중간 결과를 누적 표시
+        step_counter = 1
+        for step_result in enhanced_workflow.stream(initial_state):
+            node_name = list(step_result.keys())[0]
+            node_result = step_result[node_name]
+
+            if "step_info" in node_result and node_result["step_info"]:
+                # 현재 단계 정보를 누적 리스트에 추가
+                step_info = node_result["step_info"]
+                accumulated_steps.append(f"\n{'='*50}")
+                accumulated_steps.append(step_info)
+
+                # 진행률 표시
+                progress_bar = "▓" * step_counter + "░" * (4 - step_counter)
+                accumulated_steps.append(f"\n📈 **진행률**: [{progress_bar}] {step_counter}/4 단계")
+
+                # 누적된 모든 단계 정보를 표시
+                full_message = "\n".join(accumulated_steps)
+                current_history[-1] = (message, full_message)
+                yield "", current_history
+                time.sleep(0.8)  # 시각적 효과를 위한 지연
+                step_counter += 1
+
+        # 최종 응답을 위해 다시 워크플로우 실행 (결과 획득용)
+        final_result = None
+        for step_result in enhanced_workflow.stream(initial_state):
+            final_result = step_result
+
+        if final_result:
+            final_node_result = list(final_result.values())[0]
+            final_response = final_node_result.get("current_response", "응답을 생성할 수 없습니다.")
+
+            # 최종 응답을 누적 정보에 추가
+            accumulated_steps.append(f"\n{'='*50}")
+            accumulated_steps.append("🎉 **최종 응답**")
+            accumulated_steps.append(f"\n{final_response}")
+
+            # 완료된 전체 과정 표시
+            full_message = "\n".join(accumulated_steps)
+            current_history[-1] = (message, full_message)
+            yield "", current_history
+
+    except Exception as e:
+        error_response = f"❌ **오류 발생**\n{str(e)}"
+        current_history = history.copy()
+        current_history.append((message, error_response))
+        yield "", current_history
+
+def clear_chat():
+    """채팅 히스토리 초기화"""
+    return []
+
+# Gradio 인터페이스 생성
+def create_enhanced_gradio_interface():
+    with gr.Blocks(title="Enhanced LangGraph 챗봇", theme=gr.themes.Soft()) as demo:
+        gr.Markdown(
+            """
+            # 🚀 Enhanced LangGraph + Gradio 챗봇
+
+            **4단계 처리 과정을 실시간으로 확인할 수 있는 AI 챗봇**
+
+            **처리 단계:**
+            1. 📊 **메시지 분석** - 입력 메시지의 특성 분석
+            2. 🎯 **의도 분류** - 사용자 의도에 따른 에이전트 선택
+            3. 🔧 **컨텍스트 강화** - 전문 도메인별 컨텍스트 설정
+            4. ✅ **응답 생성** - 최적화된 답변 생성
+
+            **지원 에이전트:** 💻 프로그래머 | 📚 정보전문가 | 🔢 계산기 | 🎨 창작가 | 💬 일반대화
+            """
+        )
+
+        # 챗봇 컴포넌트
+        chatbot = gr.Chatbot(
+            value=[],
+            height=600,
+            show_label=False,
+            container=True,
+        )
+
+        with gr.Row():
+            msg = gr.Textbox(
+                placeholder="메시지를 입력하세요... (각 처리 단계가 실시간으로 표시됩니다)",
+                show_label=False,
+                scale=4,
+                container=False
+            )
+            submit_btn = gr.Button("전송", scale=1, variant="primary")
+            clear_btn = gr.Button("초기화", scale=1, variant="secondary")
+
+        # 상태 표시
+        with gr.Row():
+            gr.Markdown("💡 **팁**: 다양한 주제로 대화해보세요. 각 단계별 처리 과정을 확인할 수 있습니다!")
+
+        # 이벤트 핸들러 - 스트리밍 방식으로 변경
+        submit_btn.click(
+            stream_chatbot_response,
+            inputs=[msg, chatbot],
+            outputs=[msg, chatbot]
+        )
+
+        msg.submit(
+            stream_chatbot_response,
+            inputs=[msg, chatbot],
+            outputs=[msg, chatbot]
+        )
+
+        # 초기화 버튼
+        clear_btn.click(
+            clear_chat,
+            outputs=[chatbot]
+        )
+
+        # 카테고리별 예제 질문들
+        with gr.Row():
+            with gr.Column():
+                gr.Markdown("### 💻 프로그래밍")
+                gr.Examples(
+                    examples=[
+                        "Python으로 피보나치 수열 함수 만드는 방법?",
+                        "딕셔너리와 리스트의 차이점을 알려줘",
+                        "클래스와 객체에 대해 설명해줘"
+                    ],
+                    inputs=msg
+                )
+
+            with gr.Column():
+                gr.Markdown("### 🔢 계산/수학")
+                gr.Examples(
+                    examples=[
+                        "25 곱하기 37은 얼마야?",
+                        "복리 계산 방법을 알려줘",
+                        "삼각함수에 대해 설명해줘"
+                    ],
+                    inputs=msg
+                )
+
+            with gr.Column():
+                gr.Markdown("### 🎨 창작")
+                gr.Examples(
+                    examples=[
+                        "봄에 대한 짧은 시를 써줘",
+                        "우주 여행 이야기를 만들어줘",
+                        "창의적인 아이디어를 제안해줘"
+                    ],
+                    inputs=msg
+                )
+
+    return demo
+
+if __name__ == "__main__":
+    # 사용 방법:
+    # 1. OpenAI API 키를 설정하세요
+    # os.environ["OPENAI_API_KEY"] = "your-api-key-here"
+
+    # 2. 필요한 패키지 설치:
+    # pip install gradio langgraph langchain-openai langchain-core
+
+    # 3. 인터페이스 실행
+    demo = create_enhanced_gradio_interface()
+    demo.launch(
+    )

pyproject.toml

@@ -6,8 +6,11 @@ readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
     "gradio[mcp]==5.37",
+    "ipython>=9.4.0",
     "langchain-mcp-adapters>=0.1.9",
+    "langchain-teddynote>=0.3.45",
     "langchain[openai]>=0.3.27",
     "langgraph>=0.5.4",
+    "notebook>=7.4.4",
     "textblob>=0.19.0",
 ]