Merge pull request #6 from aws-samples/fix/feedback

Fix/feedback

Author: najmia

README.md

@@ -91,13 +91,15 @@ curl -X POST http://localhost:8080/invocations \
   -H "Content-Type: application/json" \
   -d '{"input": {"prompt": "Hello", "conversation_id": "<conversation_id>"}}' 
 
-curl -X POST http://localhost:8080/api/v1/feedback \
+curl -X POST http://localhost:8080/api/v1/invocations \
   -H "Content-Type: application/json" \
   -d '{
-    "run_id": "<run-id>",
-    "session_id": "<session-id>", 
-    "score": 1.0,
-    "comment": "Great response!"
+    "feedback": {
+      "run_id": "<run-id>",
+      "session_id": "<session-id>", 
+      "score": 1.0,
+      "comment": "Great response!"
+    }
   }'
 
 

cx-agent-backend/src/domain/services/agent_service.py

@@ -24,6 +24,7 @@ class AgentRequest:
     user_id: str
     model: str
     session_id: str | None = None
+    trace_id: str | None = None
 
 
 @dataclass(frozen=True)

cx-agent-backend/src/domain/services/conversation_service.py

@@ -1,12 +1,17 @@
 """Domain service for conversation business logic."""
 
+import logging
+import os
 from uuid import UUID
 
+from langfuse import get_client, Langfuse
+
 from domain.entities.conversation import Conversation, Message
 from domain.repositories.conversation_repository import ConversationRepository
 from domain.services.agent_service import AgentRequest, AgentService, AgentType
 from domain.services.guardrail_service import GuardrailAssessment, GuardrailService
 
+logger = logging.getLogger(__name__)
 
 class ConversationService:
     """Service for conversation business logic."""
@@ -67,6 +72,7 @@ async def send_message(
             user_id=conversation.user_id,
             model=model,
             session_id=str(conversation.id),
+            trace_id=None,  # Can be set from FastAPI layer
         )
         agent_response = await self._agent_service.process_request(agent_request)
 
@@ -109,23 +115,40 @@ async def get_user_conversations(self, user_id: str) -> list[Conversation]:
 
     async def log_feedback(self, user_id: str, session_id: str, message_id: str, score: int, comment: str = "") -> None:
         """Log user feedback to Langfuse."""
+        
+        # Log feedback attempt
+        feedback_msg = f"[FEEDBACK] Attempting to log feedback - user_id: {user_id}, session_id: {session_id}, message_id: {message_id}, score: {score}"
+        logger.info(feedback_msg)
+        
         try:
-            import os
-            from langfuse import Langfuse
+            
+            logger.info("[FEEDBACK] Langfuse config - enabled: %s, host: %s", 
+                       self._langfuse_config.get("enabled"), 
+                       self._langfuse_config.get("host"))
 
             if self._langfuse_config.get("enabled"):
+                logger.info("[FEEDBACK] Langfuse is enabled, setting environment variables")
                 os.environ["LANGFUSE_SECRET_KEY"] = self._langfuse_config.get("secret_key")
                 os.environ["LANGFUSE_PUBLIC_KEY"] = self._langfuse_config.get("public_key")
                 os.environ["LANGFUSE_HOST"] = self._langfuse_config.get("host")
 
-                langfuse = Langfuse()
+                langfuse = get_client()
+                predefined_trace_id = Langfuse.create_trace_id(seed=session_id)
 
-                langfuse.create_score(
-                    trace_id=str(f"{user_id}_{session_id}"),
-                    name="user-feedback",
-                    value=score,
-                    data_type="NUMERIC",
-                    comment=comment,
-                )
+                logger.info("[FEEDBACK] Calling span.score_trace")
+                with langfuse.start_as_current_span(
+                    name="langchain-request",
+                    trace_context={"trace_id": predefined_trace_id}
+                ) as span:
+                    result = span.score_trace(
+                        name="user-feedback",
+                        value=score,
+                        data_type="NUMERIC",
+                        comment=comment
+                    )
+                
+                logger.info("[FEEDBACK] Successfully created score: %s", result)
+            else:
+                logger.info("[FEEDBACK] Langfuse is not enabled in config")
         except Exception as e:
-            print(f"Failed to log feedback to Langfuse: {e}")
+            logger.error(f"[FEEDBACK] Failed to log feedback to Langfuse: {e}")

cx-agent-backend/src/infrastructure/adapters/langgraph_agent_service.py

@@ -3,9 +3,9 @@
 from langchain_core.messages import AIMessage, HumanMessage
 from langchain_core.runnables import RunnableConfig
 import os
-import base64
 import logging
 from langgraph.prebuilt import create_react_agent
+from langfuse import get_client, Langfuse
 from langfuse.langchain import CallbackHandler
 
 logger = logging.getLogger(__name__)
@@ -76,7 +76,20 @@ def _create_agent(self, agent_type: AgentType, model: str) -> any:
 
     async def process_request(self, request: AgentRequest) -> AgentResponse:
         """Process request through appropriate agent."""
-        logger.info(f"Processing request for user {request.user_id}, session {request.session_id}, agent type {request.agent_type}")
+        logger.info(
+            "Processing request for user %s, session %s, agent type %s",
+            request.user_id,
+            request.session_id,
+            request.agent_type,
+        )
+        
+        # Use trace_id from request if provided, otherwise create one
+        langfuse = None
+        predefined_trace_id = getattr(request, 'trace_id', None)
+        if self._langfuse_config.get("enabled"):
+            langfuse = get_client()
+            if not predefined_trace_id:
+                predefined_trace_id = Langfuse.create_trace_id(seed=request.session_id)
 
         # Check input guardrails if enabled
         if self._guardrail_service and request.messages:
@@ -115,28 +128,51 @@ async def process_request(self, request: AgentRequest) -> AgentResponse:
         # Create config with Langfuse callback if enabled
         callbacks = []
         trace_id = None
+        response = None
 
         if self._langfuse_config.get("enabled"):
             os.environ["LANGFUSE_SECRET_KEY"] = self._langfuse_config.get("secret_key")
             os.environ["LANGFUSE_PUBLIC_KEY"] = self._langfuse_config.get("public_key")
             os.environ["LANGFUSE_HOST"] = self._langfuse_config.get("host")
 
+            trace_id = predefined_trace_id
+            
             langfuse_handler = CallbackHandler()
-            callbacks.append(langfuse_handler)
-            trace_id = str(f"{request.user_id}_{request.session_id}")
-
-        config = RunnableConfig(
-            configurable={
-                "thread_id": f"{request.user_id}_{request.session_id}",
-                "user_id": request.user_id,
-            },
-            callbacks=callbacks,
-        )
-
-        # Invoke agent
-        logger.debug(f"Invoking agent with {len(lc_messages)} messages")
-        response = await agent.ainvoke({"messages": lc_messages}, config=config)
-        logger.debug(f"Agent response contains {len(response['messages'])} messages")
+            
+            with langfuse.start_as_current_span(
+                name="langchain-request",
+                trace_context={"trace_id": predefined_trace_id}
+            ) as span:
+                span.update_trace(
+                    user_id=request.user_id,
+                    input={"messages": [msg.content for msg in request.messages]}
+                )
+                
+                config = RunnableConfig(
+                    configurable={
+                        "thread_id": f"{request.session_id}",
+                        "user_id": request.user_id,
+                    },
+                    callbacks=[langfuse_handler],
+                )
+                
+                # Invoke agent
+                logger.debug("Invoking agent with %s messages", len(lc_messages))
+                response = await agent.ainvoke({"messages": lc_messages}, config=config)
+                
+                span.update_trace(output={"response": response["messages"][-1].content if response["messages"] else ""})
+        else:
+            config = RunnableConfig(
+                configurable={
+                    "thread_id": f"{request.session_id}",
+                    "user_id": request.user_id,
+                },
+            )
+            
+            # Invoke agent
+            logger.debug("Invoking agent with %s messages", len(lc_messages))
+            response = await agent.ainvoke({"messages": lc_messages}, config=config)
+        logger.debug("Agent response contains %s messages", len(response["messages"]))
         # Extract response
         last_message = response["messages"][-1]
         tools_used = []
@@ -152,7 +188,7 @@ async def process_request(self, request: AgentRequest) -> AgentResponse:
                         tools_used.append(tool_call["name"])
         # Remove duplicates
         tools_used = list(set(tools_used))
-        logger.info(f"Agent completed. Tools used: {tools_used}")
+        logger.info("Agent completed. Tools used: %s", tools_used)
 
         # Check output guardrails if enabled
         if self._guardrail_service:
@@ -173,12 +209,12 @@ async def process_request(self, request: AgentRequest) -> AgentResponse:
             "model": request.model,
             "agent_type": request.agent_type.value,
             "trace_id": trace_id,
-            "debug_message_count": len(response["messages"]),
+            "debug_message_count": len(response["messages"]) if response else 0,
             "debug_message_types": message_types,
             "debug_tools_found": len(tools_used) > 0,
         }
 
-        logger.info(f"Returning response for session {request.session_id}")
+        logger.info("Returning response for session %s", request.session_id)
         return AgentResponse(
             content=last_message.content,
             agent_type=request.agent_type,
@@ -201,7 +237,7 @@ async def stream_response(self, request: AgentRequest):
         # Create config
         config = RunnableConfig(
             configurable={
-                "thread_id": f"{request.user_id}_{request.session_id}",
+                "thread_id": f"{request.session_id}",
                 "user_id": request.user_id,
             }
         )

cx-agent-backend/src/infrastructure/adapters/openai_llm_service.py

@@ -3,7 +3,7 @@
 from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
 from langchain_openai import ChatOpenAI
 
-from domain.entities.conversation import Message, MessageRole
+from domain.entities.conversation import MessageRole
 from domain.services.llm_service import LLMRequest, LLMResponse, LLMService