Integrating Human Feedback in Chatbots with Phoenix Framework
Hi there, I’m running into an issue of integrating human feedback feature in my chatbot with Phoenix. It seems like I’m not persisting the right span_id and thus I can’t see the feedback in the right span.
import streamlit as st
from rag import RagSystem
import time
import httpx
from opentelemetry import trace
import logging
import phoenix as px
from phoenix.otel import register
logger = logging.getLogger(__name__)
class Feedback:
def __init__(self, message_idx):
self.message_idx = message_idx
self.key_prefix = f"feedback_{message_idx}"
# Retrieve the span_id from chat history
self.span_id = st.session_state.chat_history[self.message_idx].get("span_id")
if not self.span_id:
logger.warning(f"⚠️ No valid span ID found for message {message_idx}")
def _send_phoenix_annotation(self, label: str, score: float, explanation: str = ""):
"""Send feedback annotation to Phoenix."""
if not self.span_id:
logger.warning("⚠️ Skipping feedback annotation: No valid span ID available.")
return False
feedback_data = {
"label": label,
"score": score,
"explanation": explanation,
"timestamp": time.time()
}
if not hasattr(st.session_state, 'feedback_store'):
st.session_state.feedback_store = []
st.session_state.feedback_store.append(feedback_data)
try:
client = httpx.Client(timeout=5.0)
logger.info(f"📡 Sending feedback to Phoenix server for span {self.span_id}")
annotation_payload = {
"data": [{
"span_id": self.span_id,
"name": "user feedback",
"annotator_kind": "HUMAN",
"result": {
"label": label,
"score": score,
"explanation": explanation
},
"metadata": {}
}]
}
phoenix_url = "https://app.phoenix.arize.com/v1/span_annotations?sync=false"
response = client.post(
phoenix_url,
json=annotation_payload,
headers={"Content-Type": "application/json", "api_key": "<your phoenix api key>"}
)
if response.status_code == 200:
logger.info("✅ Successfully sent feedback to Phoenix server")
else:
logger.warning(f"⚠️ Phoenix server returned status {response.status_code}: {response.text}")
except httpx.RequestError as e:
logger.warning(f"⚠️ Could not connect to Phoenix server: {e}")
return True
def render(self):
"""Render feedback buttons and text input."""
st.markdown("##### How was this response?")
# Thumbs up/down buttons in a row
col1, col2, _ = st.columns([1, 1, 4])
with col1:
if st.button("👍", key=f"{self.key_prefix}_up"):
st.session_state.chat_history[self.message_idx]["feedback"] = {
"rating": "positive",
"text": st.session_state.get(f"{self.key_prefix}_text", "")
}
self._send_phoenix_annotation("thumbs_up", 1.0)
st.success("Thanks for the positive feedback!")
with col2:
if st.button("👎", key=f"{self.key_prefix}_down"):
st.session_state.chat_history[self.message_idx]["feedback"] = {
"rating": "negative",
"text": st.session_state.get(f"{self.key_prefix}_text", "")
}
self._send_phoenix_annotation("thumbs_down", 0.0)
st.error("Thanks for the feedback! Please let us know how we can improve.")
# Text feedback in a separate container
with st.container():
st.markdown("##### Additional Comments (optional)")
feedback_text = st.text_area(
"Share your thoughts about this response",
key=f"{self.key_prefix}_text",
label_visibility="collapsed",
placeholder="Enter your feedback here..."
)
if st.button("Submit Feedback", key=f"{self.key_prefix}_submit"):
if feedback_text:
if "feedback" in st.session_state.chat_history[self.message_idx]:
st.session_state.chat_history[self.message_idx]["feedback"]["text"] = feedback_text
score = 1.0 if st.session_state.chat_history[self.message_idx]["feedback"].get("rating") == "positive" else 0.0
self._send_phoenix_annotation("text_feedback", score, feedback_text)
st.success("Thank you for your detailed feedback!")
else:
st.session_state.chat_history[self.message_idx]["feedback"] = {
"rating": None,
"text": feedback_text
}
self._send_phoenix_annotation("text_feedback", 0.5, feedback_text)
st.success("Thank you for your feedback!")
else:
st.info("Please enter some feedback text before submitting.")
class ChatInterface:
def __init__(self):
self.rag_system = st.session_state.rag_system
def _display_message(self, role, content, message_idx=None):
"""Display a chat message with appropriate styling."""
with st.chat_message(role):
st.markdown(content)
if role == "assistant" and message_idx is not None:
st.markdown("---") # Add a separator line
Feedback(message_idx).render()
def _process_user_input(self, user_input: str) -> str:
"""Process user input and get response from RAG system."""
try:
with st.spinner("Thinking..."):
# Start a tracing span
with trace.get_tracer(__name__).start_as_current_span("RAG Query Processing") as span:
span.set_attribute("rag.query", user_input)
# Get response from RAG
response = self.rag_system.get_response(user_input)
# Extract the span_id
span_id = span.get_span_context().span_id.to_bytes(8, "big").hex()
# Store response with span_id in chat history
st.session_state.chat_history.append({
"role": "assistant",
"content": response,
"feedback": {"rating": None, "text": ""},
"span_id": span_id # Store span_id here
})
return response
except Exception as e:
logger.error(f"❌ Error getting response: {str(e)}", exc_info=True)
st.error("I apologize, but I encountered an error. Please try again.")
return "I apologize, but I encountered an error while processing your question. Please try again."
def render(self):
"""Render the chat interface."""
# Display chat history
for i, message in enumerate(st.session_state.chat_history):
self._display_message(
message["role"],
message["content"],
i if message["role"] == "assistant" else None
)
# Chat input
if user_input := st.chat_input("Ask a question about the Roman Empire"):
# Add user message to chat
st.session_state.chat_history.append({
"role": "user",
"content": user_input
})
self._display_message("user", user_input)
# Get and display assistant response
response = self._process_user_input(user_input)
st.session_state.chat_history.append({
"role": "assistant",
"content": response,
"feedback": {"rating": None, "text": ""}
})
self._display_message("assistant", response, len(st.session_state.chat_history) - 1)
