[feature]: Add ReAct pattern to DevRel agent

Author: smokeyScraper

backend/app/agents/devrel/agent.py

@@ -5,17 +5,15 @@
 from langchain_google_genai import ChatGoogleGenerativeAI
 from langgraph.checkpoint.memory import InMemorySaver
 from ..base_agent import BaseAgent, AgentState
-from ..classification_router import MessageCategory
 from .tools.search_tool import TavilySearchTool
 from .tools.faq_tool import FAQTool
+from .github.github_toolkit import GitHubToolkit
 from app.core.config import settings
 from .nodes.gather_context import gather_context_node
-from .nodes.handlers.faq import handle_faq_node
-from .nodes.handlers.web_search import handle_web_search_node
-from .nodes.handlers.technical_support import handle_technical_support_node
-from .nodes.handlers.onboarding import handle_onboarding_node
-from .generate_response_node import generate_response_node
 from .nodes.summarization import check_summarization_needed, summarize_conversation_node, store_summary_to_database
+from .nodes.react_supervisor import react_supervisor_node, supervisor_decision_router
+from .tool_wrappers import web_search_tool_node, faq_handler_tool_node, onboarding_tool_node, github_toolkit_tool_node
+from .nodes.generate_response import generate_response_node
 
 logger = logging.getLogger(__name__)
 
@@ -31,48 +29,55 @@ def __init__(self, config: Dict[str, Any] = None):
         )
         self.search_tool = TavilySearchTool()
         self.faq_tool = FAQTool()
+        self.github_toolkit = GitHubToolkit()
         self.checkpointer = InMemorySaver()
         super().__init__("DevRelAgent", self.config)
 
     def _build_graph(self):
         """Build the DevRel agent workflow graph"""
         workflow = StateGraph(AgentState)
 
-        # Add nodes
+        # Phase 1: Gather Context
         workflow.add_node("gather_context", gather_context_node)
-        workflow.add_node("handle_faq", partial(handle_faq_node, faq_tool=self.faq_tool))
-        workflow.add_node("handle_web_search", partial(
-            handle_web_search_node, search_tool=self.search_tool, llm=self.llm))
-        workflow.add_node("handle_technical_support", handle_technical_support_node)
-        workflow.add_node("handle_onboarding", handle_onboarding_node)
+
+        # Phase 2: ReAct Supervisor - Decide what to do next
+        workflow.add_node("react_supervisor", partial(react_supervisor_node, llm=self.llm))
+        workflow.add_node("web_search_tool", partial(web_search_tool_node, search_tool=self.search_tool, llm=self.llm))
+        workflow.add_node("faq_handler_tool", partial(faq_handler_tool_node, faq_tool=self.faq_tool))
+        workflow.add_node("onboarding_tool", onboarding_tool_node)
+        workflow.add_node("github_toolkit_tool", partial(github_toolkit_tool_node, github_toolkit=self.github_toolkit))
+
+        # Phase 3: Generate Response
         workflow.add_node("generate_response", partial(generate_response_node, llm=self.llm))
+
+        # Phase 4: Summarization
         workflow.add_node("check_summarization", check_summarization_needed)
         workflow.add_node("summarize_conversation", partial(summarize_conversation_node, llm=self.llm))
 
-        # Add edges
+        # Entry point
+        workflow.set_entry_point("gather_context")
+        workflow.add_edge("gather_context", "react_supervisor")
+
+        # ReAct supervisor routing
         workflow.add_conditional_edges(
-            "gather_context",
-            self._route_to_handler,
+            "react_supervisor",
+            supervisor_decision_router,
             {
-                MessageCategory.FAQ: "handle_faq",
-                MessageCategory.WEB_SEARCH: "handle_web_search",
-                MessageCategory.ONBOARDING: "handle_onboarding",
-                MessageCategory.TECHNICAL_SUPPORT: "handle_technical_support",
-                MessageCategory.COMMUNITY_ENGAGEMENT: "handle_technical_support",
-                MessageCategory.DOCUMENTATION: "handle_technical_support",
-                MessageCategory.BUG_REPORT: "handle_technical_support",
-                MessageCategory.FEATURE_REQUEST: "handle_technical_support",
-                MessageCategory.NOT_DEVREL: "handle_technical_support"
+                "web_search": "web_search_tool",
+                "faq_handler": "faq_handler_tool",
+                "onboarding": "onboarding_tool",
+                "github_toolkit": "github_toolkit_tool",
+                "complete": "generate_response"
             }
         )
 
-        # All handlers lead to response generation
-        for node in ["handle_faq", "handle_web_search", "handle_technical_support", "handle_onboarding"]:
-            workflow.add_edge(node, "generate_response")
+        # All tools return to supervisor
+        for tool in ["web_search_tool", "faq_handler_tool", "onboarding_tool", "github_toolkit_tool"]:
+            workflow.add_edge(tool, "react_supervisor")
 
         workflow.add_edge("generate_response", "check_summarization")
 
-        # Conditional edge for summarization
+        # Summarization routing
         workflow.add_conditional_edges(
             "check_summarization",
             self._should_summarize,
@@ -82,42 +87,11 @@ def _build_graph(self):
             }
         )
 
-        # End after summarization
         workflow.add_edge("summarize_conversation", END)
 
-        # Set entry point
-        workflow.set_entry_point("gather_context")
-
-        # Compile with InMemorySaver checkpointer
+        # Compile with checkpointer
         self.graph = workflow.compile(checkpointer=self.checkpointer)
 
-    def _route_to_handler(self, state: AgentState) -> str:
-        """Route to the appropriate handler based on intent"""
-        classification = state.context.get("classification", {})
-        intent = classification.get("category")
-
-        if isinstance(intent, str):
-            try:
-                intent = MessageCategory(intent.lower())
-            except ValueError:
-                logger.warning(f"Unknown intent string '{intent}', defaulting to TECHNICAL_SUPPORT")
-                intent = MessageCategory.TECHNICAL_SUPPORT
-
-        logger.info(f"Routing based on intent: {intent} for session {state.session_id}")
-
-        # Mapping from MessageCategory enum to string keys used in add_conditional_edges
-        if intent in [MessageCategory.FAQ, MessageCategory.WEB_SEARCH,
-                      MessageCategory.ONBOARDING, MessageCategory.TECHNICAL_SUPPORT,
-                      MessageCategory.COMMUNITY_ENGAGEMENT, MessageCategory.DOCUMENTATION,
-                      MessageCategory.BUG_REPORT, MessageCategory.FEATURE_REQUEST,
-                      MessageCategory.NOT_DEVREL]:
-            logger.info(f"Routing to handler for: {intent}")
-            return intent
-
-        # Later to be changed to handle anomalies
-        logger.info(f"Unknown intent '{intent}', routing to technical support")
-        return MessageCategory.TECHNICAL_SUPPORT
-
     def _should_summarize(self, state: AgentState) -> str:
         """Determine if conversation should be summarized"""
         if state.summarization_needed:

backend/app/agents/devrel/generate_response_node.py renamed to backend/app/agents/devrel/generate_response.py

@@ -2,7 +2,7 @@
 from typing import Dict, Any
 from app.agents.state import AgentState
 from langchain_core.messages import HumanMessage
-from .prompts.base_prompt import GENERAL_LLM_RESPONSE_PROMPT
+from .prompts.response_prompt import RESPONSE_PROMPT
 from .nodes.handlers.web_search import create_search_response
 
 logger = logging.getLogger(__name__)
@@ -46,7 +46,7 @@ async def _create_llm_response(state: AgentState, task_result: Dict[str, Any], l
     current_context_str = "\n".join(context_parts)
 
     try:
-        prompt = GENERAL_LLM_RESPONSE_PROMPT.format(
+        prompt = RESPONSE_PROMPT.format(
             conversation_summary=conversation_summary,
             latest_message=latest_message,
             conversation_history=conversation_history_str,

backend/app/agents/devrel/nodes/gather_context.py

@@ -1,10 +1,11 @@
 import logging
 from datetime import datetime
+from typing import Dict, Any
 from app.agents.state import AgentState
 
 logger = logging.getLogger(__name__)
 
-async def gather_context_node(state: AgentState) -> AgentState:
+async def gather_context_node(state: AgentState) -> Dict[str, Any]:
     """Gather additional context for the user and their request"""
     logger.info(f"Gathering context for session {state.session_id}")
 
@@ -31,5 +32,6 @@ async def gather_context_node(state: AgentState) -> AgentState:
     return {
         "messages": [new_message],
         "context": updated_context,
-        "current_task": "context_gathered"
+        "current_task": "context_gathered",
+        "last_interaction_time": datetime.now()
     }

backend/app/agents/devrel/nodes/generate_response.py

@@ -0,0 +1,101 @@
+import logging
+import json
+from typing import Dict, Any
+from app.agents.state import AgentState
+from langchain_core.messages import HumanMessage
+from ..prompts.response_prompt import RESPONSE_PROMPT
+
+logger = logging.getLogger(__name__)
+
+async def generate_response_node(state: AgentState, llm) -> Dict[str, Any]:
+    """
+    Final Response Generation Node
+    """
+    logger.info(f"Generating response for session {state.session_id}")
+
+    try:
+        final_response = await _create_response(state, llm)
+
+        return {
+            "final_response": final_response,
+            "current_task": "response_generated"
+        }
+
+    except Exception as e:
+        logger.error(f"Error generating response: {str(e)}")
+        return {
+            "final_response": "I apologize, but I encountered an error while generating my response. Please try asking your question again.",
+            "errors": state.errors + [str(e)],
+            "current_task": "response_error"
+        }
+
+async def _create_response(state: AgentState, llm) -> str:
+    """
+    Response Generation and LLM synthesis
+    """
+    logger.info(f"Creating response for session {state.session_id}")
+
+    latest_message = _get_latest_message(state)
+
+    conversation_summary = state.conversation_summary or "This is the beginning of our conversation."
+
+    recent_messages_count = min(10, len(state.messages))
+    conversation_history = ""
+    if state.messages:
+        conversation_history = "\n".join([
+            f"{msg.get('role', 'user')}: {msg.get('content', '')}"
+            for msg in state.messages[-recent_messages_count:]
+        ])
+
+        if len(state.messages) > recent_messages_count:
+            conversation_history = f"[Showing last {recent_messages_count} of {len(state.messages)} messages]\n" + \
+                conversation_history
+    else:
+        conversation_history = "No previous conversation"
+
+    context_parts = [
+        f"Platform: {state.platform}",
+        f"Total interactions: {state.interaction_count}",
+        f"Session duration: {(state.last_interaction_time - state.session_start_time).total_seconds() / 60:.1f} minutes"
+    ]
+
+    if state.key_topics:
+        context_parts.append(f"Key topics discussed: {', '.join(state.key_topics)}")
+    if state.user_profile:
+        context_parts.append(f"User profile: {state.user_profile}")
+
+    current_context = "\n".join(context_parts)
+
+    supervisor_thinking = state.context.get("supervisor_thinking", "No reasoning process available")
+
+    tool_results = state.context.get("tool_results", [])
+    tool_results_str = json.dumps(tool_results, indent=2) if tool_results else "No tool results"
+
+    task_result = state.task_result or {}
+    task_result_str = json.dumps(task_result, indent=2) if task_result else "No task result"
+
+    try:
+        prompt = RESPONSE_PROMPT.format(
+            latest_message=latest_message,
+            conversation_summary=conversation_summary,
+            conversation_history=conversation_history,
+            current_context=current_context,
+            supervisor_thinking=supervisor_thinking,
+            tool_results=tool_results_str,
+            task_result=task_result_str
+        )
+
+        logger.info(f"Generated response prompt using existing RESPONSE_PROMPT")
+
+    except KeyError as e:
+        logger.error(f"Missing key in RESPONSE_PROMPT: {e}")
+        return f"Error: Response template formatting error - {str(e)}"
+
+    response = await llm.ainvoke([HumanMessage(content=prompt)])
+    return response.content.strip()
+
+def _get_latest_message(state: AgentState) -> str:
+    """Extract the latest message from state"""
+    if state.messages:
+        return state.messages[-1].get("content", "")
+    return state.context.get("original_message", "")

backend/app/agents/devrel/nodes/handlers/user_support.py

@@ -0,0 +1,112 @@
+import logging
+import json
+from typing import Dict, Any, Literal
+from app.agents.state import AgentState
+from langchain_core.messages import HumanMessage
+from ..prompts.react_prompt import REACT_SUPERVISOR_PROMPT
+
+logger = logging.getLogger(__name__)
+
+async def react_supervisor_node(state: AgentState, llm) -> Dict[str, Any]:
+    """ReAct Supervisor: Think -> Act -> Observe"""
+    logger.info(f"ReAct Supervisor thinking for session {state.session_id}")
+
+    # Get current context
+    latest_message = _get_latest_message(state)
+    conversation_history = _get_conversation_history(state)
+    tool_results = state.context.get("tool_results", [])
+    iteration_count = state.context.get("iteration_count", 0)
+
+    prompt = REACT_SUPERVISOR_PROMPT.format(
+        latest_message=latest_message,
+        platform=state.platform,
+        interaction_count=state.interaction_count,
+        iteration_count=iteration_count,
+        conversation_history=conversation_history,
+        tool_results=json.dumps(tool_results, indent=2) if tool_results else "No previous tool results"
+    )
+
+    response = await llm.ainvoke([HumanMessage(content=prompt)])
+    decision = _parse_supervisor_decision(response.content)
+
+    logger.info(f"ReAct Supervisor decision: {decision['action']}")
+
+    # Update state with supervisor's thinking
+    return {
+        "context": {
+            **state.context,
+            "supervisor_thinking": response.content,
+            "supervisor_decision": decision,
+            "iteration_count": iteration_count + 1
+        },
+        "current_task": f"supervisor_decided_{decision['action']}"
+    }
+
+def _parse_supervisor_decision(response: str) -> Dict[str, Any]:
+    """Parse the supervisor's decision from LLM response"""
+    try:
+        lines = response.strip().split('\n')
+        decision = {"action": "complete", "reasoning": "", "thinking": ""}
+
+        for line in lines:
+            if line.startswith("THINK:"):
+                decision["thinking"] = line.replace("THINK:", "").strip()
+            elif line.startswith("ACT:"):
+                action = line.replace("ACT:", "").strip().lower()
+                if action in ["web_search", "faq_handler", "onboarding", "github_toolkit", "complete"]:
+                    decision["action"] = action
+            elif line.startswith("REASON:"):
+                decision["reasoning"] = line.replace("REASON:", "").strip()
+
+        return decision
+    except Exception as e:
+        logger.error(f"Error parsing supervisor decision: {e}")
+        return {"action": "complete", "reasoning": "Error in decision parsing", "thinking": ""}
+
+def supervisor_decision_router(state: AgentState) -> Literal["web_search", "faq_handler", "onboarding", "github_toolkit", "complete"]:
+    """Route based on supervisor's decision"""
+    decision = state.context.get("supervisor_decision", {})
+    action = decision.get("action", "complete")
+
+    # Safety check for infinite loops
+    iteration_count = state.context.get("iteration_count", 0)
+    if iteration_count > 10:
+        logger.warning(f"Max iterations reached for session {state.session_id}")
+        return "complete"
+
+    return action
+
+def add_tool_result(state: AgentState, tool_name: str, result: Dict[str, Any]) -> Dict[str, Any]:
+    """Add tool result to state context"""
+    tool_results = state.context.get("tool_results", [])
+    tool_results.append({
+        "tool": tool_name,
+        "result": result,
+        "iteration": state.context.get("iteration_count", 0)
+    })
+
+    return {
+        "context": {
+            **state.context,
+            "tool_results": tool_results
+        },
+        "tools_used": state.tools_used + [tool_name],
+        "current_task": f"completed_{tool_name}"
+    }
+
+def _get_latest_message(state: AgentState) -> str:
+    """Extract the latest message from state"""
+    if state.messages:
+        return state.messages[-1].get("content", "")
+    return state.context.get("original_message", "")
+
+def _get_conversation_history(state: AgentState, max_messages: int = 5) -> str:
+    """Get formatted conversation history"""
+    if not state.messages:
+        return "No previous conversation"
+
+    recent_messages = state.messages[-max_messages:]
+    return "\n".join([
+        f"{msg.get('role', 'user')}: {msg.get('content', '')}"
+        for msg in recent_messages
+    ])