状态机模式 描述了智能体的行为随着任务在不同状态间转换而改变的工作流。本教程展示如何通过使用工具调用来动态更改单个智能体的配置,从而实现状态机——根据当前状态更新其可用工具和指令。状态可以从多个来源确定:智能体的过去操作(工具调用)、外部状态(例如 API 调用结果),甚至是初始用户输入(例如,通过运行分类器来确定用户意图)。在本教程中,您将构建一个执行以下操作的客户支持智能体:
from langchain.tools import tool, ToolRuntimefrom langchain.messages import ToolMessagefrom langgraph.types import Command@tooldef 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_step": "issue_classifier", } )@tooldef 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_step": "resolution_specialist", } )@tooldef escalate_to_human(reason: str) -> 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}"@tooldef provide_solution(solution: str) -> str: """Provide a solution to the customer's issue.""" return f"Solution provided: {solution}"
# Define prompts as constants for easy referenceWARRANTY_COLLECTOR_PROMPT = """You are a customer support agent helping with device issues.CURRENT STAGE: Warranty verificationAt this step, you need to:1. Greet the customer warmly2. Ask if their device is under warranty3. Use record_warranty_status to record their response and move to the next stepBe 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 classificationCUSTOMER INFO: Warranty status is {warranty_status}At this step, you need to:1. Ask the customer to describe their issue2. 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 stepIf unclear, ask clarifying questions before classifying."""RESOLUTION_SPECIALIST_PROMPT = """You are a customer support agent helping with device issues.CURRENT STAGE: ResolutionCUSTOMER INFO: Warranty status is {warranty_status}, issue type is {issue_type}At this step, you need to:1. For SOFTWARE issues: provide troubleshooting steps using provide_solution2. For HARDWARE issues: - If IN WARRANTY: explain warranty repair process using provide_solution - If OUT OF WARRANTY: escalate_to_human for paid repair optionsBe specific and helpful in your solutions."""
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponsefrom typing import Callable@wrap_model_calldef apply_step_config( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse],) -> ModelResponse: """Configure agent behavior based on the current step.""" # Get current step (defaults to warranty_collector for first interaction) current_step = request.state.get("current_step", "warranty_collector") # Look up step configuration stage_config = STEP_CONFIG[current_step] # 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_step}") # Format prompt with state values (supports {warranty_status}, {issue_type}, etc.) system_prompt = stage_config["prompt"].format(**request.state) # Inject system prompt and step-specific tools request = request.override( system_prompt=system_prompt, tools=stage_config["tools"], ) return handler(request)
@tooldef go_back_to_warranty() -> Command: """Go back to warranty verification step.""" return Command(update={"current_step": "warranty_collector"})@tooldef go_back_to_classification() -> Command: """Go back to issue classification step.""" return Command(update={"current_step": "issue_classifier"})# Update the resolution_specialist configuration to include these toolsSTEP_CONFIG["resolution_specialist"]["tools"].extend([ go_back_to_warranty, go_back_to_classification])
更新解决专家的提示以提及这些工具:
RESOLUTION_SPECIALIST_PROMPT = """You are a customer support agent helping with device issues.CURRENT STAGE: ResolutionCUSTOMER INFO: Warranty status is {warranty_status}, issue type is {issue_type}At this step, you need to:1. For SOFTWARE issues: provide troubleshooting steps using provide_solution2. For HARDWARE issues: - If IN WARRANTY: explain warranty repair process using provide_solution - If OUT OF WARRANTY: escalate_to_human for paid repair optionsIf the customer indicates any information was wrong, use:- go_back_to_warranty to correct warranty status- go_back_to_classification to correct issue typeBe specific and helpful in your solutions."""
现在智能体可以处理更正:
result = agent.invoke( {"messages": [HumanMessage("Actually, I made a mistake - my device is out of warranty")]}, config)# Agent will call go_back_to_warranty and restart the warranty verification step
"""Customer Support State Machine ExampleThis example demonstrates the state machine pattern.A single agent dynamically changes its behavior based on the current_step state,creating a state machine for sequential information collection."""from langchain_core.utils.uuid import uuid7from langgraph.checkpoint.memory import InMemorySaverfrom langgraph.types import Commandfrom typing import Callable, Literalfrom typing_extensions import NotRequiredfrom langchain.agents import AgentState, create_agentfrom langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse, SummarizationMiddlewarefrom langchain.chat_models import init_chat_modelfrom langchain.messages import HumanMessage, ToolMessagefrom langchain.tools import tool, ToolRuntimemodel = init_chat_model("google_genai:gemini-3.1-pro-preview")# Define the possible workflow stepsSupportStep = Literal["warranty_collector", "issue_classifier", "resolution_specialist"]class SupportState(AgentState): """State for customer support workflow.""" current_step: NotRequired[SupportStep] warranty_status: NotRequired[Literal["in_warranty", "out_of_warranty"]] issue_type: NotRequired[Literal["hardware", "software"]]@tooldef 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_step": "issue_classifier", } )@tooldef 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_step": "resolution_specialist", } )@tooldef escalate_to_human(reason: str) -> 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}"@tooldef provide_solution(solution: str) -> str: """Provide a solution to the customer's issue.""" return f"Solution provided: {solution}"# Define prompts as constantsWARRANTY_COLLECTOR_PROMPT = """You are a customer support agent helping with device issues.CURRENT STEP: Warranty verificationAt this step, you need to:1. Greet the customer warmly2. Ask if their device is under warranty3. Use record_warranty_status to record their response and move to the next stepBe conversational and friendly. Don't ask multiple questions at once."""ISSUE_CLASSIFIER_PROMPT = """You are a customer support agent helping with device issues.CURRENT STEP: Issue classificationCUSTOMER INFO: Warranty status is {warranty_status}At this step, you need to:1. Ask the customer to describe their issue2. 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 stepIf unclear, ask clarifying questions before classifying."""RESOLUTION_SPECIALIST_PROMPT = """You are a customer support agent helping with device issues.CURRENT STEP: ResolutionCUSTOMER INFO: Warranty status is {warranty_status}, issue type is {issue_type}At this step, you need to:1. For SOFTWARE issues: provide troubleshooting steps using provide_solution2. For HARDWARE issues: - If IN WARRANTY: explain warranty repair process using provide_solution - If OUT OF WARRANTY: escalate_to_human for paid repair optionsBe specific and helpful in your solutions."""# Step configuration: maps step name to (prompt, tools, required_state)STEP_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_calldef apply_step_config( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse],) -> ModelResponse: """Configure agent behavior based on the current step.""" # Get current step (defaults to warranty_collector for first interaction) current_step = request.state.get("current_step", "warranty_collector") # Look up step configuration step_config = STEP_CONFIG[current_step] # Validate required state exists for key in step_config["requires"]: if request.state.get(key) is None: raise ValueError(f"{key} must be set before reaching {current_step}") # Format prompt with state values system_prompt = step_config["prompt"].format(**request.state) # Inject system prompt and step-specific tools request = request.override( system_prompt=system_prompt, tools=step_config["tools"], ) return handler(request)# Collect all tools from all step configurationsall_tools = [ record_warranty_status, record_issue_type, provide_solution, escalate_to_human,]# Create the agent with step-based configuration and summarizationagent = create_agent( model, tools=all_tools, state_schema=SupportState, middleware=[ apply_step_config, SummarizationMiddleware( model="gpt-5.4-mini", trigger=("tokens", 4000), keep=("messages", 10) ) ], checkpointer=InMemorySaver(),)# ============================================================================# Test the workflow# ============================================================================if __name__ == "__main__": thread_id = str(uuid7()) config = {"configurable": {"thread_id": thread_id}} result = agent.invoke( {"messages": [HumanMessage("Hi, my phone screen is cracked")]}, config ) result = agent.invoke( {"messages": [HumanMessage("Yes, it's still under warranty")]}, config ) result = agent.invoke( {"messages": [HumanMessage("The screen is physically cracked from dropping it")]}, config ) result = agent.invoke( {"messages": [HumanMessage("What should I do?")]}, config ) for msg in result['messages']: msg.pretty_print()