docs: handoffs conceptual part

handoffs.py

@@ -0,0 +1,187 @@
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.types import Command
+from typing import Callable, Any
+from typing import Literal
+from typing_extensions import NotRequired
+from dataclasses import dataclass
+
+from langchain.agents import AgentState
+from langchain.agents import create_agent
+from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
+from langchain.chat_models import init_chat_model
+from langchain.messages import HumanMessage, SystemMessage
+from langchain.tools import tool, ToolRuntime
+from langchain_core.messages import ToolMessage
+
+model = init_chat_model("anthropic:claude-3-5-sonnet-latest")
+
+
+# Define the possible workflow stages
+SupportStage = Literal["warranty_collector", "issue_classifier", "resolution_specialist"]
+
+
+class SupportState(AgentState):
+    """State for customer support workflow with handoffs."""
+
+    current_stage: NotRequired[SupportStage]
+    warranty_status: NotRequired[Literal["in_warranty", "out_of_warranty"]]
+    issue_type: NotRequired[Literal["hardware", "software"]]
+
+
+
+@tool
+def record_warranty_status(
+    status: Literal["in_warranty", "out_of_warranty"],
+    runtime: ToolRuntime[None, SupportState],
+) -> Command:
+    """Record the customer's warranty status and transition to issue classification."""
+    return Command(
+        update={
+            "messages": [
+                ToolMessage(
+                    content=f"Warranty status recorded as: {status}",
+                    tool_call_id=runtime.tool_call_id,
+                )
+            ],
+            "warranty_status": status,
+            "current_stage": "issue_classifier",
+        }
+    )
+
+
+@tool
+def record_issue_type(
+    issue_type: Literal["hardware", "software"],
+    runtime: ToolRuntime[None, SupportState],
+) -> Command:
+    """Record the type of issue and transition to resolution specialist."""
+    return Command(
+        update={
+            "messages": [
+                ToolMessage(
+                    content=f"Issue type recorded as: {issue_type}",
+                    tool_call_id=runtime.tool_call_id,
+                )
+            ],
+            "issue_type": issue_type,
+            "current_stage": "resolution_specialist",
+        }
+    )
+
+
+@tool
+def escalate_to_human(reason: str, runtime: ToolRuntime[None, SupportState]) -> str:
+    """Escalate the case to a human support specialist."""
+    # In a real system, this would create a ticket, notify staff, etc.
+    return f"Escalating to human support. Reason: {reason}"
+
+
+@tool
+def provide_solution(solution: str, runtime: ToolRuntime[None, SupportState]) -> str:
+    """Provide a solution to the customer's issue."""
+    return f"Solution provided: {solution}"
+
+
+# Define prompts as constants for lazy interpolation
+WARRANTY_COLLECTOR_PROMPT = """You are a customer support agent helping with device issues.
+
+CURRENT STAGE: Warranty verification
+
+At this stage, you need to:
+1. Greet the customer warmly
+2. Ask if their device is under warranty
+3. Use record_warranty_status to record their response and move to the next stage
+
+Be conversational and friendly. Don't ask multiple questions at once."""
+
+ISSUE_CLASSIFIER_PROMPT = """You are a customer support agent helping with device issues.
+
+CURRENT STAGE: Issue classification
+CUSTOMER INFO: Warranty status is {warranty_status}
+
+At this stage, you need to:
+1. Ask the customer to describe their issue
+2. Determine if it's a hardware issue (physical damage, broken parts) or software issue (app crashes, performance)
+3. Use record_issue_type to record the classification and move to the next stage
+
+If unclear, ask clarifying questions before classifying."""
+
+RESOLUTION_SPECIALIST_PROMPT = """You are a customer support agent helping with device issues.
+
+CURRENT STAGE: Resolution
+CUSTOMER INFO: Warranty status is {warranty_status}, issue type is {issue_type}
+
+At this stage, you need to:
+1. For SOFTWARE issues: provide troubleshooting steps using provide_solution
+2. For HARDWARE issues:
+   - If IN WARRANTY: explain warranty repair process using provide_solution
+   - If OUT OF WARRANTY: escalate_to_human for paid repair options
+
+Be specific and helpful in your solutions."""
+
+
+# Stage configuration: maps stage name to (prompt_template, tools, required_state)
+STAGE_CONFIG = {
+    "warranty_collector": {
+        "prompt": WARRANTY_COLLECTOR_PROMPT,
+        "tools": [record_warranty_status],
+        "requires": [],
+    },
+    "issue_classifier": {
+        "prompt": ISSUE_CLASSIFIER_PROMPT,
+        "tools": [record_issue_type],
+        "requires": ["warranty_status"],
+    },
+    "resolution_specialist": {
+        "prompt": RESOLUTION_SPECIALIST_PROMPT,
+        "tools": [provide_solution, escalate_to_human],
+        "requires": ["warranty_status", "issue_type"],
+    },
+}
+
+
+@wrap_model_call
+async def apply_stage_config(
+    request: ModelRequest,
+    handler: Callable[[ModelRequest], ModelResponse],
+) -> ModelResponse:
+    """Configure agent behavior based on the current stage."""
+    # Get current stage (defaults to warranty_collector for first interaction)
+    current_stage = request.state.get("current_stage", "warranty_collector")
+
+    # Look up stage configuration
+    stage_config = STAGE_CONFIG[current_stage]
+
+    # Validate required state exists
+    for key in stage_config["requires"]:
+        if request.state.get(key) is None:
+            raise ValueError(f"{key} must be set before reaching {current_stage}")
+
+    # Format prompt with state values (supports {warranty_status}, {issue_type}, etc.)
+    system_prompt = stage_config["prompt"].format(**request.state)
+
+    # Inject system prompt and stage-specific tools
+    request = request.override(
+        system_prompt=system_prompt,
+        tools=stage_config["tools"],
+    )
+
+    return await handler(request)
+
+# Collect all tools from all agent configurations
+all_tools = [
+    record_warranty_status,
+    record_issue_type,
+    provide_solution,
+    escalate_to_human,
+]
+
+# Create the agent with stage-based configuration
+agent = create_agent(
+    model,
+    tools=all_tools,
+    state_schema=SupportState,
+    middleware=[apply_stage_config],  # Single middleware handles all stages
+    checkpointer=InMemorySaver(),  # Required for state persistence across turns
+)
+

pyproject.toml

@@ -10,6 +10,8 @@ dependencies = [
     "nbformat>=5.0.0",
     "pyyaml>=6.0.2",
     "nbconvert>=7.16.6",
+    "langchain>=1.0.1",
+    "langchain-anthropic>=1.0.0",
 ]
 
 

src/docs.json

@@ -330,7 +330,8 @@
                           "oss/python/langchain/knowledge-base",
                           "oss/python/langchain/rag",
                           "oss/python/langchain/sql-agent",
-                          "oss/python/langchain/supervisor"
+                          "oss/python/langchain/supervisor",
+                          "oss/python/langchain/customer-support-handoffs"
                         ]
                       },
                       {