Spaces:
Runtime error
Runtime error
File size: 4,677 Bytes
35ac7e4 |
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 |
from __future__ import annotations
import asyncio
import tempfile
from pathlib import Path
from typing import Optional
import gradio as gr
from conversation_storyline.io import load_messages, load_messages_from_text
from conversation_storyline.pipeline import run_pipeline
from conversation_storyline.plots import (
load_graph_json,
load_interactions_df,
plot_reply_sankey,
plot_sentiment_histogram,
plot_sentiment_over_time,
plot_speaker_activity_heatmap,
plot_speaker_topic_heatmap,
plot_topic_shift_timeline,
)
def get_backend(name: str):
if name == "openai":
from conversation_storyline.llm_backends.openai_backend import OpenAIBackend
return OpenAIBackend()
elif name == "outlines":
from conversation_storyline.llm_backends.outlines_backend import OutlinesBackend
return OutlinesBackend()
else:
raise ValueError("backend inválido")
async def _run(file_path: Optional[str], transcript_text: str, backend: str):
transcript_text = (transcript_text or "").strip()
if transcript_text:
msgs = load_messages_from_text(transcript_text)
else:
if not file_path:
raise ValueError("Debes pegar un transcript o subir un archivo.")
msgs = load_messages(file_path)
b = get_backend(backend)
outdir = Path(tempfile.mkdtemp(prefix="storyline_"))
await run_pipeline(msgs, b, str(outdir))
png = outdir / "storyline.png"
html = outdir / "storyline.html"
graph = outdir / "graph.json"
interactions = outdir / "interactions.jsonl"
metrics = outdir / "metrics.parquet"
html_inline = html.read_text(encoding="utf-8", errors="ignore") if html.exists() else None
figs = [None] * 5
try:
df = load_interactions_df(outdir)
g = load_graph_json(outdir)
figs = [
plot_sentiment_over_time(df),
plot_sentiment_histogram(df),
plot_speaker_topic_heatmap(df),
plot_speaker_activity_heatmap(df),
plot_reply_sankey(g),
]
topic_shift_fig = plot_topic_shift_timeline(df)
except Exception:
topic_shift_fig = None
return (
str(png) if png.exists() else None,
html_inline,
str(html) if html.exists() else None,
str(graph) if graph.exists() else None,
str(interactions) if interactions.exists() else None,
str(metrics) if metrics.exists() else None,
figs[0],
figs[1],
figs[2],
figs[3],
figs[4],
topic_shift_fig,
)
def run_ui(file_obj, transcript_text: str, backend: str):
file_path = file_obj.name if file_obj is not None else None
return asyncio.run(_run(file_path, transcript_text, backend))
with gr.Blocks(title="Conversation Storyline – v4") as demo:
gr.Markdown("# Conversation Storyline – v4\nPega un transcript o sube TXT/CSV.")
with gr.Row():
f = gr.File(label="Upload (.txt o .csv)")
backend = gr.Dropdown(choices=["openai", "outlines"], value="openai", label="Backend LLM")
transcript_text = gr.Textbox(label="O pega aquí el transcript", lines=10)
btn = gr.Button("Run", variant="primary")
with gr.Tabs():
with gr.Tab("Storyline"):
with gr.Row():
out_png = gr.Image(label="Storyline (PNG)", type="filepath")
out_story_html = gr.HTML(label="Storyline (HTML embebido)")
out_html_file = gr.File(label="Storyline HTML (descarga)")
with gr.Tab("Analítica"):
out_sentiment = gr.Plot(label="Sentiment timeline")
out_hist = gr.Plot(label="Sentiment histogram")
out_topic_heat = gr.Plot(label="Speaker × topic heatmap")
out_activity_heat = gr.Plot(label="Speaker activity heatmap")
out_topic_shifts = gr.Plot(label="Topic shifts timeline")
with gr.Tab("Grafo"):
out_sankey = gr.Plot(label="Sankey replies")
out_graph = gr.File(label="Graph JSON")
with gr.Tab("Artifacts"):
out_interactions = gr.File(label="interactions.jsonl")
out_metrics = gr.File(label="metrics.parquet")
btn.click(
fn=run_ui,
inputs=[f, transcript_text, backend],
outputs=[
out_png,
out_story_html,
out_html_file,
out_graph,
out_interactions,
out_metrics,
out_sentiment,
out_hist,
out_topic_heat,
out_activity_heat,
out_sankey,
out_topic_shifts,
],
)
if __name__ == "__main__":
demo.launch()
|