Chainlit & Dash

🧠 Thought: 사용자가 모델 성능을 물어보고 있습니다. MLflow에서 최근 실험 결과를 조회해야 합니다.
🔧 Action: query_mlflow_experiments(model_name="predictive_maintenance_v3")
📋 Observation: {"accuracy": 0.94, "f1_score": 0.91, "drift_detected": false, ...}
🧠 Thought: 모델 성능이 양호합니다. 사용자에게 결과를 요약하겠습니다.
💬 Answer: 예지보전 모델(v3)의 최근 성능은...

import chainlit as cl
from langchain_databricks import ChatDatabricks
from langgraph.prebuilt import create_react_agent

# Tool 정의
from langchain_core.tools import tool

@tool
def query_model_metrics(model_name: str) -> str:
    """MLflow에서 모델의 최신 메트릭을 조회합니다."""
    import mlflow
    client = mlflow.tracking.MlflowClient()
    versions = client.search_model_versions(f"name='{model_name}'")
    if versions:
        run = client.get_run(versions[0].run_id)
        return str(run.data.metrics)
    return "모델을 찾을 수 없습니다."

@tool
def check_data_drift(table_name: str) -> str:
    """Unity Catalog 모니터에서 데이터 드리프트 상태를 확인합니다."""
    from databricks.sdk import WorkspaceClient
    w = WorkspaceClient()
    # 모니터 조회 로직
    return "드리프트 미감지. 모든 피처가 정상 범위 내에 있습니다."

# Agent 구성
model = ChatDatabricks(endpoint="databricks-meta-llama-3-3-70b-instruct")
tools = [query_model_metrics, check_data_drift]
agent = create_react_agent(model, tools)

@cl.on_chat_start
async def start():
    """채팅 시작 시 초기화"""
    await cl.Message(
        content="안녕하세요! MLOps Agent입니다. 모델 성능, 드리프트, 배포 상태에 대해 물어보세요."
    ).send()

@cl.on_message
async def main(message: cl.Message):
    """사용자 메시지 처리"""
    # Agent 실행 (중간 Step 자동 표시)
    result = await cl.make_async(agent.invoke)(
        {"messages": [("user", message.content)]}
    )

    # 최종 응답 전송
    final_message = result["messages"][-1].content
    await cl.Message(content=final_message).send()

import chainlit as cl

@cl.on_message
async def main(message: cl.Message):
    # Step을 명시적으로 표시
    async with cl.Step(name="DB 조회", type="tool") as step:
        step.input = "sales 테이블에서 최근 분기 데이터 조회"
        result = query_database("SELECT * FROM sales WHERE quarter = '2025Q4'")
        step.output = f"조회 결과: {len(result)}건"

    async with cl.Step(name="분석", type="llm") as step:
        step.input = f"조회된 {len(result)}건의 데이터를 분석"
        analysis = llm.invoke(f"다음 데이터를 분석해주세요: {result}")
        step.output = analysis

    await cl.Message(content=analysis).send()

import dash
from dash import dcc, html, Input, Output, State
import plotly.express as px

app = dash.Dash(__name__)

app.layout = html.Div([
    html.H1("모델 모니터링 대시보드"),

    # 모델 선택
    dcc.Dropdown(
        id="model-selector",
        options=[
            {"label": "예지보전 v3", "value": "pred_maint_v3"},
            {"label": "이상탐지 v2", "value": "anomaly_v2"},
        ],
        value="pred_maint_v3"
    ),

    # 차트 영역
    dcc.Graph(id="accuracy-chart"),
    dcc.Graph(id="drift-chart"),

    # 채팅 영역 (간단한 Q&A)
    html.Div([
        dcc.Input(id="chat-input", type="text", placeholder="Agent에게 질문하세요..."),
        html.Button("전송", id="send-button"),
        html.Div(id="chat-output")
    ])
])

@app.callback(
    Output("accuracy-chart", "figure"),
    Input("model-selector", "value")
)
def update_accuracy_chart(model_name):
    """모델 선택 시 정확도 차트 업데이트"""
    # MLflow에서 메트릭 조회
    data = get_model_metrics(model_name)
    fig = px.line(data, x="date", y="accuracy", title=f"{model_name} 정확도 추이")
    return fig

@app.callback(
    Output("chat-output", "children"),
    Input("send-button", "n_clicks"),
    State("chat-input", "value"),
    prevent_initial_call=True
)
def handle_chat(n_clicks, message):
    """Agent 대화 처리"""
    if message:
        response = agent.invoke(message)
        return html.P(response)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)