-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathapp.py
More file actions
78 lines (57 loc) · 2.31 KB
/
app.py
File metadata and controls
78 lines (57 loc) · 2.31 KB
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
import json
from json import JSONDecodeError
from typing import Any, Dict
import streamlit as st
from graph.main import graph
st.set_page_config(page_title="Grievance Intake", page_icon="📥")
st.title("Grievance Intake")
st.caption("Provide grievances as JSON. Messages are moderated and stored when approved.")
def render_chat_log():
for entry in st.session_state.chat_log:
with st.chat_message(entry["role"]):
st.markdown(entry["content"])
def record_message(role: str, content: str) -> None:
st.session_state.chat_log.append({"role": role, "content": content})
def validate_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
errors = []
level = payload.get("level")
message = payload.get("message")
if not isinstance(level, int) or not 1 <= level <= 5:
errors.append("`level` must be an integer between 1 and 5.")
if not isinstance(message, str) or not message.strip():
errors.append("`message` must be a non-empty string.")
if errors:
raise ValueError(" ".join(errors))
return {"level": level, "message": message.strip()}
if "chat_log" not in st.session_state:
st.session_state.chat_log = []
with st.sidebar:
st.markdown(
"Example payload:\n"
"```json\n"
'{\n "level": 3,\n "message": "Build pipeline flakes on Mondays."\n}\n'
"```"
)
prompt = st.chat_input('Enter grievance JSON e.g. {"level": 2, "message": "..."}')
if prompt:
record_message("user", prompt)
try:
payload = json.loads(prompt)
if not isinstance(payload, dict):
raise ValueError("Payload must be a JSON object with `level` and `message`.")
cleaned = validate_payload(payload)
except (JSONDecodeError, ValueError) as exc:
record_message("assistant", f"Submission rejected: {exc}")
else:
result = graph.invoke({"request": cleaned})
response = result.get("response", "Workflow completed.")
if result.get("decision") == "approved":
storage_path = result.get("storage_path")
if storage_path:
response += f"\nSaved to `{storage_path}`."
else:
reason = result.get("reason")
if reason:
response += f"\nReason: {reason}"
record_message("assistant", response)
render_chat_log()