from __future__ import annotations import json import os import re import sys import time from datetime import datetime, timezone from pathlib import Path import streamlit as st ROOT = Path(__file__).resolve().parent SRC = ROOT / "src" if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) from qa_langgraph.export import build_run_export from qa_langgraph.graph import build_graph from qa_langgraph.registry import AGENTS, default_agent_configs from qa_langgraph.run_logger import read_stage_traces_jsonl, write_run_export from qa_langgraph.sample_scenarios import CUSTOM_SCENARIO, load_sample_scenario, scenario_options from qa_langgraph.service_client import DEFAULT_SERVICE_URL, normalize_service_url, validate_agent_service def init_session() -> None: default_service_url = normalize_service_url(os.getenv("QA_AGENT_SERVICE_URL", DEFAULT_SERVICE_URL)) st.session_state.setdefault("agent_configs", default_agent_configs()) st.session_state.setdefault("run_state", {}) st.session_state.setdefault("run_export", None) st.session_state.setdefault("run_status", "not run") st.session_state.setdefault("current_run_id", "") st.session_state.setdefault("stage_traces_jsonl", "") st.session_state.setdefault("selected_scenario", CUSTOM_SCENARIO) st.session_state.setdefault("title", "") st.session_state.setdefault("requirements_text", "") st.session_state.setdefault("qa_agent_service_url", default_service_url) def agent_status_rows(stage_traces: list[dict]) -> list[dict]: latest: dict[str, dict] = {} for trace in stage_traces: latest[trace.get("agent_key", "")] = trace rows = [] for key, agent in AGENTS.items(): trace = latest.get(key, {}) rows.append( { "agent": agent.name, "status": trace.get("status", "not started"), "iteration": trace.get("iteration", ""), "duration_ms": trace.get("duration_ms", ""), } ) return rows def safe_filename_part(value: str, fallback: str = "scenario") -> str: cleaned = re.sub(r"[^A-Za-z0-9]+", "_", value.strip()).strip("_") return cleaned or fallback def download_basename(export: dict) -> str: title = safe_filename_part(export.get("title", "scenario")) created_at = export.get("run_metadata", {}).get("created_at", "") try: timestamp = datetime.fromisoformat(created_at.replace("Z", "+00:00")).strftime("%Y%m%d-%H%M%S") except ValueError: timestamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") return f"LangGraph_{title}_{timestamp}" def _dot_label(label: str) -> str: return label.replace('"', '\\"') def runtime_path_edges(stage_traces: list[dict], orchestration_decision: dict) -> set[tuple[str, str]]: edges: set[tuple[str, str]] = set() ordered_nodes = [trace.get("agent_key") for trace in stage_traces if trace.get("agent_key")] if ordered_nodes: edges.add(("START", ordered_nodes[0])) for source, target in zip(ordered_nodes, ordered_nodes[1:]): edges.add((source, target)) decision = orchestration_decision.get("decision") target_agent = orchestration_decision.get("target_agent") if ordered_nodes and ordered_nodes[-1] == "orchestrator": if decision == "finish": edges.add(("orchestrator", "END")) elif decision == "fail": edges.add(("orchestrator", "END_FAIL")) elif decision == "backtrack" and target_agent in {"requirements_analyst", "test_designer", "reviewer"}: edges.add(("orchestrator", target_agent)) return edges def build_workflow_dot(stage_traces: list[dict] | None = None, orchestration_decision: dict | None = None) -> str: active_edges = runtime_path_edges(stage_traces or [], orchestration_decision or {}) completed_nodes = {trace.get("agent_key") for trace in (stage_traces or []) if trace.get("status") == "completed"} failed_nodes = {trace.get("agent_key") for trace in (stage_traces or []) if trace.get("status") == "failed"} nodes = { "START": {"label": "START", "shape": "circle", "fillcolor": "#1e293b", "fontcolor": "#f8fafc"}, "requirements_analyst": { "label": "Requirements Analyst\\n(service)", "shape": "box", "fillcolor": "#0b5fff" if "requirements_analyst" in completed_nodes else "#f8fafc", "fontcolor": "#ffffff" if "requirements_analyst" in completed_nodes else "#0f172a", }, "test_designer": { "label": "Test Designer\\n(service)", "shape": "box", "fillcolor": "#0b5fff" if "test_designer" in completed_nodes else "#f8fafc", "fontcolor": "#ffffff" if "test_designer" in completed_nodes else "#0f172a", }, "reviewer": { "label": "Reviewer\\n(service)", "shape": "box", "fillcolor": "#0b5fff" if "reviewer" in completed_nodes else "#f8fafc", "fontcolor": "#ffffff" if "reviewer" in completed_nodes else "#0f172a", }, "orchestrator": { "label": "Orchestrator\\n(local deterministic)", "shape": "diamond", "fillcolor": "#f59e0b" if "orchestrator" in completed_nodes else "#fef3c7", "fontcolor": "#0f172a", }, "END": {"label": "END\\nfinish", "shape": "doublecircle", "fillcolor": "#16a34a", "fontcolor": "#ffffff"}, "END_FAIL": {"label": "END\\nfail", "shape": "doublecircle", "fillcolor": "#dc2626", "fontcolor": "#ffffff"}, } for node_key in failed_nodes: if node_key in nodes: nodes[node_key]["fillcolor"] = "#dc2626" nodes[node_key]["fontcolor"] = "#ffffff" edges = [ ("START", "requirements_analyst", ""), ("requirements_analyst", "test_designer", ""), ("test_designer", "reviewer", ""), ("reviewer", "orchestrator", ""), ("orchestrator", "END", "finish"), ("orchestrator", "requirements_analyst", "backtrack"), ("orchestrator", "test_designer", "backtrack"), ("orchestrator", "reviewer", "backtrack"), ("orchestrator", "END_FAIL", "fail"), ] lines = [ "digraph LangGraphWorkflow {", ' graph [rankdir=LR, bgcolor="#f8fafc", fontname="Helvetica", nodesep="0.55", ranksep="0.75", pad="0.25"];', ' node [style="rounded,filled", color="#0f172a", penwidth=1.5, fontname="Helvetica"];', ' edge [color="#334155", penwidth=1.9, fontname="Helvetica", fontsize=11, fontcolor="#0f172a"];', ] for node_key, attrs in nodes.items(): lines.append( f' {node_key} [label="{_dot_label(attrs["label"])}", shape="{attrs["shape"]}", fillcolor="{attrs["fillcolor"]}", fontcolor="{attrs["fontcolor"]}"];' ) for source, target, label in edges: is_active = (source, target) in active_edges color = "#0b5fff" if is_active else "#334155" penwidth = "3.4" if is_active else "1.9" fontcolor = "#0b5fff" if is_active else "#0f172a" label_attr = f', label="{label}", fontcolor="{fontcolor}"' if label else "" lines.append(f' {source} -> {target} [color="{color}", penwidth={penwidth}{label_attr}];') lines.append("}") return "\n".join(lines) def runtime_path_summary(stage_traces: list[dict], orchestration_decision: dict) -> str: ordered_names = [AGENTS[trace["agent_key"]].name for trace in stage_traces if trace.get("agent_key") in AGENTS] if not ordered_names: return "No run has been executed yet." summary = " -> ".join(ordered_names) decision = orchestration_decision.get("decision") if decision == "finish": return f"{summary} -> END (finish)" if decision == "fail": return f"{summary} -> END (fail)" if decision == "backtrack" and orchestration_decision.get("target_agent") in AGENTS: target_name = AGENTS[orchestration_decision["target_agent"]].name return f"{summary} -> backtrack to {target_name}" return summary def render_service_help(service_url: str) -> None: rest_url = normalize_service_url(service_url) st.markdown("**Use REST here, not MCP.**") st.caption("LangGraph calls FastAPI REST endpoints. MCP on port 8008 is for MCP clients and does not expose Swagger/OpenAPI.") st.table( [ { "server": "Hosted REST/FastAPI", "default": "https://helmfridsson-qa-agent-service.hf.space/api/v1", "paths": "/health, /docs, /openapi.json, /agents/{agent_key}/invoke", }, { "server": "Local REST/FastAPI", "default": "http://127.0.0.1:8009/api/v1", "paths": "/health, /docs, /openapi.json, /agents/{agent_key}/invoke", }, {"server": "Local MCP", "default": "http://127.0.0.1:8008/mcp", "paths": "/mcp only, no Swagger/OpenAPI"}, ] ) st.caption("You can paste the full REST base URL, the service root, `/docs`, `/health`, or the Hugging Face Space page URL. The app normalizes it to `/api/v1`.") st.markdown( f""" REST links for the configured URL: - Health: [{rest_url}/health]({rest_url}/health) - Swagger: [{rest_url}/docs]({rest_url}/docs) - OpenAPI JSON: [{rest_url}/openapi.json]({rest_url}/openapi.json) """ ) st.code( """cd /Users/tomashelmfridsson/workspace/qa-agent-service PYTHONPATH=src uvicorn api:app --reload --port 8009""", language="bash", ) st.caption("The PYTHONPATH part fixes ModuleNotFoundError when qa-agent-service is not installed in the active Python environment.") st.caption("If the hosted /api/v1 health check returns Streamlit HTML, the Space is running the monitor UI but not exposing the REST API surface yet.") def run_workflow(title: str, requirements_text: str, global_config: dict, agent_configs: list[dict], max_rounds: int) -> None: if not title.strip(): st.error("Scenario title is required.") return if not requirements_text.strip(): st.error("Requirement text is required.") return if any(config.get("execution_mode") not in {"Deterministic", "Disabled"} for config in agent_configs): validate_agent_service(global_config) run_id = f"run_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}" state = { "run_metadata": { "run_id": run_id, "created_at": datetime.now(timezone.utc).isoformat(), "started_monotonic_ms": int(time.time() * 1000), "app_version": "langgraph-qa-compare-0.1.0", }, "title": title.strip(), "source_requirements": requirements_text.strip(), "requirements": [], "test_designs": [], "generated_artifacts": [], "iterations": 1, "run_controls": { "max_rounds": max_rounds, "max_feedback_messages": 0, "max_feedback_per_agent_pair": 0, }, "global_llm_config": global_config, "agent_configs": agent_configs, "stage_traces": [], "working_memory": {"shared": {}, "agent_private": {}, "timeline": []}, "errors": [], } st.session_state.current_run_id = run_id st.session_state.run_status = "running" st.session_state.run_state = state st.session_state.run_export = None st.session_state.stage_traces_jsonl = "" graph = build_graph() final_state = state status_slot = st.empty() traces_slot = st.empty() memory_slot = st.empty() for snapshot in graph.stream(state, {"recursion_limit": max(8, max_rounds * 6)}, stream_mode="values"): final_state = snapshot st.session_state.run_state = final_state st.session_state.run_status = "running" traces = final_state.get("stage_traces", []) memory = final_state.get("working_memory", {"shared": {}, "agent_private": {}, "timeline": []}) status_slot.info(f"Running... {len(traces)} stage trace(s), iteration {final_state.get('iterations', 1)}") traces_slot.dataframe(agent_status_rows(traces), use_container_width=True) memory_slot.json( { "shared": memory.get("shared", {}), "timeline": memory.get("timeline", []), } ) export = build_run_export(final_state) export_dict = export.model_dump(mode="json") write_run_export(run_id, export_dict) st.session_state.run_state = final_state st.session_state.run_export = export_dict st.session_state.run_status = export_dict.get("run_metadata", {}).get("status", "partial") st.session_state.stage_traces_jsonl = read_stage_traces_jsonl(run_id) init_session() st.set_page_config(page_title="LangGraph QA Compare", layout="wide") st.markdown( """ """, unsafe_allow_html=True, ) st.title("LangGraph QA Compare") with st.sidebar: st.header("QA Agent Service") service_url = st.text_input("REST API URL", value=st.session_state.get("qa_agent_service_url", DEFAULT_SERVICE_URL)) st.session_state.qa_agent_service_url = normalize_service_url(service_url) service_timeout_seconds = st.number_input("Service timeout seconds", min_value=10, max_value=600, value=150, step=10) max_rounds = st.number_input("Max rounds", min_value=1, max_value=8, value=2, step=1) global_config = { "qa_agent_service_url": st.session_state.qa_agent_service_url, "service_timeout_seconds": int(service_timeout_seconds), } if st.button("Validate QA Agent Service"): try: validate_agent_service(global_config) st.success("QA Agent Service is reachable.") except Exception as exc: st.error(str(exc)) with st.expander("REST vs MCP endpoints", expanded=False): render_service_help(service_url) if st.button("Clear session"): st.session_state.clear() st.rerun() tab_input, tab_agents, tab_flow, tab_runtime, tab_memory, tab_results, tab_downloads = st.tabs( ["Scenario Input", "Agent Configuration", "Flow Map", "Runtime Activity", "Working Memory", "Results", "Downloads"] ) with tab_input: selected_scenario = st.selectbox( "Predefined scenario", scenario_options(), index=scenario_options().index(st.session_state.get("selected_scenario", CUSTOM_SCENARIO)), ) if st.button("Load selected scenario"): title_value, requirements_value = load_sample_scenario(selected_scenario) st.session_state.selected_scenario = selected_scenario st.session_state.title = title_value st.session_state.requirements_text = requirements_value st.rerun() title = st.text_input("Scenario title", value=st.session_state.get("title", "")) requirements_text = st.text_area( "Raw requirements", value=st.session_state.get("requirements_text", ""), height=300, ) if st.button("Run workflow", type="primary"): st.session_state.title = title st.session_state.requirements_text = requirements_text try: run_workflow(title, requirements_text, global_config, st.session_state.agent_configs, int(max_rounds)) st.success("Workflow finished. Check Runtime Activity and Results.") except Exception as exc: st.session_state.run_status = "failed" st.error(str(exc)) with tab_agents: updated_configs = [] for config in st.session_state.agent_configs: key = config["agent_key"] agent_definition = AGENTS[key] with st.expander(config["agent_name"], expanded=False): mode_options = ["Deterministic", "Disabled"] if key == "orchestrator" else ["Service-backed", "Deterministic", "Disabled"] current_mode = config.get("execution_mode", "Deterministic" if key == "orchestrator" else "Service-backed") if current_mode == "LLM-backed": current_mode = "Service-backed" if current_mode not in mode_options: current_mode = mode_options[0] execution_mode = st.selectbox( "Execution mode", mode_options, index=mode_options.index(current_mode), key=f"{key}_mode", ) provider_strategy = st.selectbox( "Caller provider override", ["global", "ollama", "openai"], index=["global", "ollama", "openai"].index(config.get("provider_strategy", "global")), key=f"{key}_provider", ) model_override = st.text_input("Caller model override", value=config.get("model_override", ""), key=f"{key}_model") agent_timeout = st.number_input( "Service call timeout seconds", min_value=10, max_value=600, value=int(config.get("timeout_seconds", 150)), step=10, key=f"{key}_timeout", ) st.markdown("Default directives") st.markdown( f"
{agent_definition.default_directives}
", unsafe_allow_html=True, ) directives = st.text_area("Additional directives", value=config.get("directives", ""), key=f"{key}_directives") updated = dict(config) updated.update( { "execution_mode": execution_mode, "provider_strategy": provider_strategy, "model_override": model_override, "timeout_seconds": int(agent_timeout), "directives": directives, } ) updated_configs.append(updated) st.session_state.agent_configs = updated_configs run_state = st.session_state.get("run_state", {}) stage_traces = run_state.get("stage_traces", []) orchestration_decision = run_state.get("orchestration_decision", {}) with tab_flow: col_static, col_runtime = st.columns(2) with col_static: st.subheader("Workflow structure") st.caption("Static LangGraph routing, including the orchestrator's conditional backtracking paths.") st.graphviz_chart(build_workflow_dot(), use_container_width=True) with col_runtime: st.subheader("Latest run path") st.caption("Highlighted edges show the path taken by the current run state.") st.graphviz_chart(build_workflow_dot(stage_traces, orchestration_decision), use_container_width=True) st.code(runtime_path_summary(stage_traces, orchestration_decision), language="text") if orchestration_decision: st.json( { "decision": orchestration_decision.get("decision"), "target_agent": orchestration_decision.get("target_agent"), "reason": orchestration_decision.get("reason"), "feedback": orchestration_decision.get("feedback"), } ) with tab_runtime: export = st.session_state.get("run_export") status = export.get("run_metadata", {}).get("status") if export else st.session_state.get("run_status", "not run") st.metric("Current run status", status) if st.session_state.get("current_run_id"): st.caption(f"Run ID: {st.session_state.current_run_id}") st.metric("Iteration count", run_state.get("iterations", 0)) st.dataframe(agent_status_rows(stage_traces), use_container_width=True) st.caption(runtime_path_summary(stage_traces, orchestration_decision)) st.json(stage_traces) with tab_memory: memory = run_state.get("working_memory", {"shared": {}, "timeline": []}) st.metric("Current run status", st.session_state.get("run_status", "not run")) if st.session_state.get("current_run_id"): st.caption(f"Run ID: {st.session_state.current_run_id}") st.subheader("Shared") st.json(memory.get("shared", {})) st.subheader("Timeline") st.json(memory.get("timeline", [])) with tab_results: st.subheader("Requirements") st.dataframe(run_state.get("requirements", []), use_container_width=True) st.subheader("Test cases") st.dataframe(run_state.get("test_designs", []), use_container_width=True) st.subheader("Review") st.json(run_state.get("review", {})) st.subheader("Orchestration decision") st.json(orchestration_decision) st.subheader("Final JSON preview") st.json(st.session_state.get("run_export") or {}) with tab_downloads: export = st.session_state.get("run_export") if export: basename = download_basename(export) st.download_button( "Download run_export.json", data=json.dumps(export, indent=2, ensure_ascii=False), file_name=f"{basename}_run_export.json", mime="application/json", ) st.download_button( "Download stage_traces.jsonl", data=st.session_state.get("stage_traces_jsonl", ""), file_name=f"{basename}_stage_traces.jsonl", mime="application/x-ndjson", ) else: st.info("Run the workflow to generate downloads.")