implement task decomposer agent

Author: ks6088ts

Makefile

@@ -114,4 +114,4 @@ ci-test-docs: install-deps-docs docs ## run CI test for documentation
 
 .PHONY: langgraph-studio
 langgraph-studio: ## run LangGraph Studio
-	uv run langgraph dev --no-reload
+	uv run langgraph dev

docs/index.md

@@ -42,6 +42,7 @@ uv run python -m template_langgraph.tasks.run_issue_formatter_agent
 ### LangGraph
 
 - [Build a custom workflow](https://langchain-ai.github.io/langgraph/concepts/why-langgraph/)
+- [LangGraphの(LLMなし)Human-in-the-loopを試してみた](https://qiita.com/te_yama/items/db38201af60dec76384d)
 
 ### Sample Codes
 

template_langgraph/agents/task_decomposer_agent/agent.py

@@ -1,6 +1,7 @@
 from langgraph.graph import END, StateGraph
+from langgraph.types import interrupt
 
-from template_langgraph.agents.chat_with_tools_agent.models import AgentState
+from template_langgraph.agents.task_decomposer_agent.models import AgentState, TaskList
 from template_langgraph.llms.azure_openais import AzureOpenAiWrapper
 from template_langgraph.loggers import get_logger
 
@@ -18,22 +19,58 @@ def create_graph(self):
 
         # Create nodes
         workflow.add_node("chat", self.chat)
+        workflow.add_node("human_feedback", self.human_feedback)
 
         # Create edges
         workflow.set_entry_point("chat")
-        workflow.add_edge("chat", END)
-
-        # Compile the graph
+        workflow.add_edge("chat", "human_feedback")
+        workflow.add_conditional_edges(
+            source="human_feedback",
+            path=self.route_human_feedback,
+            path_map={
+                "loopback": "chat",
+                "end": END,
+            },
+        )
         return workflow.compile()
 
     def chat(self, state: AgentState) -> AgentState:
         """Chat with tools using the state."""
         logger.info(f"Chatting with tools using state: {state}")
-        return {
-            "messages": [
-                self.llm.invoke(state["messages"]),
-            ]
-        }
+
+        task_list = self.llm.with_structured_output(TaskList).invoke(
+            input=state["messages"],
+        )
+        state["task_list"] = task_list
+        logger.info(f"Decomposed tasks: {task_list}")
+        return state
+
+    def human_feedback(self, state: AgentState) -> AgentState:
+        """Handle human feedback."""
+        logger.info(f"Handling human feedback with state: {state}")
+        feedback = interrupt("Type your feedback. If you want to end the conversation, type 'end'.")
+        state["messages"].append(
+            {
+                "content": feedback,
+                "role": "user",
+            }
+        )
+        return state
+
+    def route_human_feedback(
+        self,
+        state: AgentState,
+    ):
+        """
+        Use in the conditional_edge to route to the HumanFeedbackNode if the last message
+        has human feedback. Otherwise, route to the end.
+        """
+        human_feedback = state["messages"][-1].content.strip().lower()
+        if human_feedback == "end":
+            logger.info("Ending the conversation as per user request.")
+            return "end"
+        logger.info("Looping back to chat for further processing.")
+        return "loopback"
 
     def draw_mermaid_png(self) -> bytes:
         """Draw the graph in Mermaid format."""

template_langgraph/agents/task_decomposer_agent/models.py

@@ -19,6 +19,10 @@ class Task(BaseModel):
     assigned_to: str | None = Field(None, description="Name of the agent assigned to the task")
 
 
+class TaskList(BaseModel):
+    tasks: Sequence[Task] = Field(..., description="List of tasks to be decomposed")
+
+
 class AgentInput(BaseModel):
     request: str = Field(..., description="Request from the user")
 
@@ -29,4 +33,4 @@ class AgentOutput(BaseModel):
 
 class AgentState(TypedDict):
     messages: Annotated[Sequence[BaseMessage], add_messages]
-    decomposed_tasks: Sequence[Task]
+    task_list: TaskList