LangChain 中文文档

外观

状态机模式描述了一类工作流：智能体的行为会随着它在任务的不同状态之间推进而发生变化。本教程演示如何通过工具调用动态改变单个智能体的配置来实现状态机——根据当前状态更新其可用工具和指令。状态可以由多个来源确定：智能体过去的动作（工具调用）、外部状态（例如 API 调用结果），甚至最初的用户输入（例如，通过运行分类器来确定用户意图）。

在本教程中，你将构建一个客户支持智能体，实现以下功能：

  • 在继续处理前收集保修信息。
  • 将问题分类为硬件或软件。
  • 提供解决方案或将问题升级给人工支持。
  • 在多个对话轮次之间维护会话状态。

与子智能体模式（子智能体作为工具被调用）不同，状态机模式使用单个智能体，其配置根据工作流进度而改变。每个"步骤"都只是同一个底层智能体的不同配置（系统提示词 + 工具），根据状态动态选择。

以下是我们将要构建的工作流：

设置 ​

安装 ​

本教程需要 langchain 包：

bash
pip install langchain
bash
uv add langchain
bash
conda install langchain -c conda-forge
bash
npm install langchain
bash
yarn add langchain
bash
pnpm add langchain

更多详情，请参阅我们的安装指南。

LangSmith ​

设置 LangSmith 以检查智能体内部发生的情况。然后设置以下环境变量：

bash
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."
python
import getpass
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()
bash
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."
typescript
process.env.LANGSMITH_TRACING = "true";
process.env.LANGSMITH_API_KEY = "...";

选择 LLM ​

从 LangChain 的集成套件中选择一个对话模型：

OpenAI ​

    👉 Read the [OpenAI chat model integration docs](/oss/python/integrations/chat/openai)
bash
pip install -U "langchain[openai]"
python
import os
from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "sk-..."

model = init_chat_model("gpt-5.5")
python
import os
from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "sk-..."

model = ChatOpenAI(model="gpt-5.5")

Anthropic ​

    👉 Read the [Anthropic chat model integration docs](/oss/python/integrations/chat/anthropic)
bash
pip install -U "langchain[anthropic]"
python
import os
from langchain.chat_models import init_chat_model

os.environ["ANTHROPIC_API_KEY"] = "sk-..."

model = init_chat_model("claude-sonnet-4-6")
python
import os
from langchain_anthropic import ChatAnthropic

os.environ["ANTHROPIC_API_KEY"] = "sk-..."

model = ChatAnthropic(model="claude-sonnet-4-6")

Azure ​

    👉 Read the [Azure chat model integration docs](/oss/python/integrations/chat/azure_chat_openai)
bash
pip install -U "langchain[openai]"
python
import os
from langchain.chat_models import init_chat_model

os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"

model = init_chat_model(
    "azure_openai:gpt-5.5",
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)
python
import os
from langchain_openai import AzureChatOpenAI

os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"

model = AzureChatOpenAI(
    model="gpt-5.5",
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"]
)

Google Gemini ​

    👉 Read the [Google GenAI chat model integration docs](/oss/python/integrations/chat/google_generative_ai)
bash
pip install -U "langchain[google-genai]"
python
import os
from langchain.chat_models import init_chat_model

os.environ["GOOGLE_API_KEY"] = "..."

model = init_chat_model("google_genai:gemini-2.5-flash-lite")
python
import os
from langchain_google_genai import ChatGoogleGenerativeAI

os.environ["GOOGLE_API_KEY"] = "..."

model = ChatGoogleGenerativeAI(model="gemini-2.5-flash-lite")

AWS Bedrock ​

    👉 Read the [AWS Bedrock chat model integration docs](/oss/python/integrations/chat/bedrock)
bash
pip install -U "langchain[aws]"
python
from langchain.chat_models import init_chat_model

# Follow the steps here to configure your credentials:
# https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html

model = init_chat_model(
    "us.anthropic.claude-sonnet-4-6",
    model_provider="bedrock_converse",
)
python
from langchain_aws import ChatBedrock

model = ChatBedrock(model="us.anthropic.claude-sonnet-4-6")

HuggingFace ​

    👉 Read the [HuggingFace chat model integration docs](/oss/python/integrations/chat/huggingface)
bash
pip install -U "langchain[huggingface]"
python
import os
from langchain.chat_models import init_chat_model

os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."

model = init_chat_model(
    "microsoft/Phi-3-mini-4k-instruct",
    model_provider="huggingface",
    temperature=0.7,
    max_tokens=1024,
)
python
import os
from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint

os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."

llm = HuggingFaceEndpoint(
    repo_id="microsoft/Phi-3-mini-4k-instruct",
    temperature=0.7,
    max_length=1024,
)
model = ChatHuggingFace(llm=llm)

OpenRouter ​

    👉 Read the [OpenRouter chat model integration docs](/oss/python/integrations/chat/openrouter)
bash
pip install -U "langchain-openrouter"
python
import os
from langchain.chat_models import init_chat_model

os.environ["OPENROUTER_API_KEY"] = "sk-..."

model = init_chat_model(
    "auto",
    model_provider="openrouter",
)
python
import os
from langchain_openrouter import ChatOpenRouter

os.environ["OPENROUTER_API_KEY"] = "sk-..."

model = ChatOpenRouter(model="auto")

OpenAI ​

    👉 Read the [OpenAI chat model integration docs](/oss/javascript/integrations/chat/openai)
bash
npm install @langchain/openai
bash
pnpm install @langchain/openai
bash
yarn add @langchain/openai
bash
bun add @langchain/openai
typescript
import { initChatModel } from "langchain";

process.env.OPENAI_API_KEY = "your-api-key";

const model = await initChatModel("gpt-5.5");
typescript
import { ChatOpenAI } from "@langchain/openai";

const model = new ChatOpenAI({
  model: "gpt-5.5",
  apiKey: "your-api-key"
});

Anthropic ​

    👉 Read the [Anthropic chat model integration docs](/oss/javascript/integrations/chat/anthropic)
bash
npm install @langchain/anthropic
bash
pnpm install @langchain/anthropic
bash
yarn add @langchain/anthropic
bash
pnpm add @langchain/anthropic
typescript
import { initChatModel } from "langchain";

process.env.ANTHROPIC_API_KEY = "your-api-key";

const model = await initChatModel("claude-sonnet-4-6");
typescript
import { ChatAnthropic } from "@langchain/anthropic";

const model = new ChatAnthropic({
  model: "claude-sonnet-4-6",
  apiKey: "your-api-key"
});

Azure ​

    👉 Read the [Azure chat model integration docs](/oss/javascript/integrations/chat/azure)
bash
npm install @langchain/azure
bash
pnpm install @langchain/azure
bash
yarn add @langchain/azure
bash
bun add @langchain/azure
typescript
import { initChatModel } from "langchain";

process.env.AZURE_OPENAI_API_KEY = "your-api-key";
process.env.AZURE_OPENAI_ENDPOINT = "your-endpoint";
process.env.OPENAI_API_VERSION = "your-api-version";

const model = await initChatModel("azure_openai:gpt-5.5");
typescript
import { AzureChatOpenAI } from "@langchain/openai";

const model = new AzureChatOpenAI({
  model: "gpt-5.5",
  azureOpenAIApiKey: "your-api-key",
  azureOpenAIApiEndpoint: "your-endpoint",
  azureOpenAIApiVersion: "your-api-version"
});

Google Gemini ​

    👉 Read the [Google GenAI chat model integration docs](/oss/javascript/integrations/chat/google_generative_ai)
bash
npm install @langchain/google-genai
bash
pnpm install @langchain/google-genai
bash
yarn add @langchain/google-genai
bash
bun add @langchain/google-genai
typescript
import { initChatModel } from "langchain";

process.env.GOOGLE_API_KEY = "your-api-key";

const model = await initChatModel("google-genai:gemini-2.5-flash-lite");
typescript
import { ChatGoogleGenerativeAI } from "@langchain/google-genai";

const model = new ChatGoogleGenerativeAI({
  model: "gemini-2.5-flash-lite",
  apiKey: "your-api-key"
});

Bedrock Converse ​

    👉 Read the [AWS Bedrock chat model integration docs](/oss/javascript/integrations/chat/bedrock_converse)
bash
npm install @langchain/aws
bash
pnpm install @langchain/aws
bash
yarn add @langchain/aws
bash
bun add @langchain/aws
typescript
import { initChatModel } from "langchain";

// Follow the steps here to configure your credentials:
// https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html

const model = await initChatModel("bedrock:gpt-5.5");
typescript
import { ChatBedrockConverse } from "@langchain/aws";

// Follow the steps here to configure your credentials:
// https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html

const model = new ChatBedrockConverse({
  model: "gpt-5.5",
  region: "us-east-2"
});

1. 定义自定义状态 ​

首先，定义一个自定义状态模式，用于跟踪当前处于哪个步骤：

python
from langchain.agents import AgentState
from typing_extensions import NotRequired
from typing import Literal

# 定义可能的工作流步骤
SupportStep = 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"]]
typescript
import { StateSchema } from "@langchain/langgraph";
import { z } from "zod";

// 定义可能的工作流步骤
const SupportStepSchema = z.enum(["warranty_collector", "issue_classifier", "resolution_specialist"]);  
const WarrantyStatusSchema = z.enum(["in_warranty", "out_of_warranty"]);
const IssueTypeSchema = z.enum(["hardware", "software"]);

// 客户支持工作流的状态
const SupportState = new StateSchema({  
  currentStep: SupportStepSchema.optional(),  
  warrantyStatus: WarrantyStatusSchema.optional(),
  issueType: IssueTypeSchema.optional(),
});

current_step 字段是状态机模式的核心——它决定每一轮加载哪个配置（提示词 + 工具）。

2. 创建管理工作流状态的工具 ​

创建更新工作流状态的工具。这些工具允许智能体记录信息并进入下一步。

关键是使用 Command 更新状态，包括 current_step 字段：

python
from langchain.tools import tool, ToolRuntime
from langchain.messages import ToolMessage
from langgraph.types import Command

@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_step": "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_step": "resolution_specialist",  
        }
    )

@tool
def escalate_to_human(reason: str) -> str:
    """Escalate the case to a human support specialist."""
    # 在真实系统中，这会创建工单、通知工作人员等。
    return f"Escalating to human support. Reason: {reason}"

@tool
def provide_solution(solution: str) -> str:
    """Provide a solution to the customer's issue."""
    return f"Solution provided: {solution}"
typescript
import { z } from "zod";
import { tool, ToolMessage, type ToolRuntime } from "langchain";
import { Command } from "@langchain/langgraph";

const recordWarrantyStatus = tool(
  async (input, config: ToolRuntime<typeof SupportState.State>) => {
    return new Command({ 
      update: { 
        messages: [
          new ToolMessage({
            content: `Warranty status recorded as: ${input.status}`,
            tool_call_id: config.toolCallId,
          }),
        ],
        warrantyStatus: input.status,
        currentStep: "issue_classifier", 
      },
    });
  },
  {
    name: "record_warranty_status",
    description:
      "Record the customer's warranty status and transition to issue classification.",
    schema: z.object({
      status: WarrantyStatusSchema,
    }),
  }
);

const recordIssueType = tool(
  async (input, config: ToolRuntime<typeof SupportState.State>) => {
    return new Command({ 
      update: { 
        messages: [
          new ToolMessage({
            content: `Issue type recorded as: ${input.issueType}`,
            tool_call_id: config.toolCallId,
          }),
        ],
        issueType: input.issueType,
        currentStep: "resolution_specialist", 
      },
    });
  },
  {
    name: "record_issue_type",
    description:
      "Record the type of issue and transition to resolution specialist.",
    schema: z.object({
      issueType: IssueTypeSchema,
    }),
  }
);

const escalateToHuman = tool(
  async (input) => {
    // 在真实系统中，这会创建工单、通知工作人员等。
    return `Escalating to human support. Reason: ${input.reason}`;
  },
  {
    name: "escalate_to_human",
    description: "Escalate the case to a human support specialist.",
    schema: z.object({
      reason: z.string(),
    }),
  }
);

const provideSolution = tool(
  async (input) => {
    return `Solution provided: ${input.solution}`;
  },
  {
    name: "provide_solution",
    description: "Provide a solution to the customer's issue.",
    schema: z.object({
      solution: z.string(),
    }),
  }
);

注意 record_warranty_status 和 record_issue_type 如何返回 Command 对象，这些对象同时更新数据（warranty_status、issue_type）和 current_step。这就是状态机的工作原理——工具控制工作流的推进。

3. 定义步骤配置 ​

为每个步骤定义提示词和工具。首先，为每个步骤定义提示词：

查看完整的提示词定义 ​

python
# 将提示词定义为常量，便于引用
WARRANTY_COLLECTOR_PROMPT = """You are a customer support agent helping with device issues.

CURRENT STAGE: Warranty verification

At this step, 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 step

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 step, 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 step

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 step, 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."""
typescript
// 将提示词定义为常量，便于引用
const WARRANTY_COLLECTOR_PROMPT = `You are a customer support agent helping with device issues.

CURRENT STAGE: Warranty verification

At this step, 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 step

Be conversational and friendly. Don't ask multiple questions at once.`;

const 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 step, 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 step

If unclear, ask clarifying questions before classifying.`;

const 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 step, 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.`;

然后使用字典将步骤名称映射到其配置：

python
# 步骤配置：将步骤名称映射到（提示词、工具、所需状态）
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"],
    },
}
typescript
// 步骤配置：将步骤名称映射到（提示词、工具、所需状态）
const STEP_CONFIG = {
  warranty_collector: {
    prompt: WARRANTY_COLLECTOR_PROMPT,
    tools: [recordWarrantyStatus],
    requires: [],
  },
  issue_classifier: {
    prompt: ISSUE_CLASSIFIER_PROMPT,
    tools: [recordIssueType],
    requires: ["warrantyStatus"],
  },
  resolution_specialist: {
    prompt: RESOLUTION_SPECIALIST_PROMPT,
    tools: [provideSolution, escalateToHuman],
    requires: ["warrantyStatus", "issueType"],
  },
} as const;

这种基于字典的配置便于：

  • 一眼查看所有步骤
  • 添加新步骤（只需再添加一条）
  • 了解工作流依赖关系（requires 字段）
  • 使用带状态变量的提示词模板（例如 {warranty_status}）

4. 创建基于步骤的中间件 ​

创建从状态中读取 current_step 并应用相应配置的中间件。我们将使用 @wrap_model_call 装饰器来实现简洁的实现：

python
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable

@wrap_model_call
def apply_step_config(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    """Configure agent behavior based on the current step."""
    # 获取当前步骤（首次交互时默认为 warranty_collector）
    current_step = request.state.get("current_step", "warranty_collector")  

    # 查找步骤配置
    stage_config = STEP_CONFIG[current_step]  

    # 验证所需的状态是否存在
    for key in stage_config["requires"]:
        if request.state.get(key) is None:
            raise ValueError(f"{key} must be set before reaching {current_step}")

    # 使用状态值格式化提示词（支持 {warranty_status}、{issue_type} 等）
    system_prompt = stage_config["prompt"].format(**request.state)

    # 注入系统提示词和步骤专属工具
    request = request.override(  
        system_prompt=system_prompt,  
        tools=stage_config["tools"],  
    )

    return handler(request)
typescript
import { createMiddleware } from "langchain";

const applyStepMiddleware = createMiddleware({
  name: "applyStep",
  stateSchema: SupportState,
  wrapModelCall: async (request, handler) => {
    // 获取当前步骤（首次交互时默认为 warranty_collector）
    const currentStep = request.state.currentStep ?? "warranty_collector"; 

    // 查找步骤配置
    const stepConfig = STEP_CONFIG[currentStep]; 

    // 验证所需的状态是否存在
    for (const key of stepConfig.requires) {
      if (request.state[key] === undefined) {
        throw new Error(`${key} must be set before reaching ${currentStep}`);
      }
    }

    // 使用状态值格式化提示词（支持 {warrantyStatus}、{issueType} 等）
    let systemPrompt: string = stepConfig.prompt;
    for (const [key, value] of Object.entries(request.state)) {
      systemPrompt = systemPrompt.replace(`{${key}}`, String(value ?? ""));
    }

    // 注入系统提示词和步骤专属工具
    return handler({
      ...request, 
      systemPrompt, 
      tools: [...stepConfig.tools], 
    });
  },
});

该中间件：

  1. 读取当前步骤：从状态中获取 current_step（默认为 "warranty_collector"）。
  2. 查找配置：在 STEP_CONFIG 中找到匹配的条目。
  3. 验证依赖：确保所需的状态字段存在。
  4. 格式化提示词：将状态值注入提示词模板。
  5. 应用配置：覆盖系统提示词和可用工具。

request.override() 方法是关键——它允许我们根据状态动态改变智能体的行为，而无需创建单独的智能体实例。

5. 创建智能体 ​

现在创建带有基于步骤的中间件和用于状态持久化的检查点的智能体：

python
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

# 收集所有步骤配置中的全部工具
all_tools = [
    record_warranty_status,
    record_issue_type,
    provide_solution,
    escalate_to_human,
]

# 使用基于步骤的配置创建智能体
agent = create_agent(
    model,
    tools=all_tools,
    state_schema=SupportState,  
    middleware=[apply_step_config],  
    checkpointer=InMemorySaver(),  
)
typescript
import { createAgent } from "langchain";
import { MemorySaver } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";

// 收集所有步骤配置中的全部工具
const allTools = [
  recordWarrantyStatus,
  recordIssueType,
  provideSolution,
  escalateToHuman,
];

// 初始化模型
const model = new ChatOpenAI({
  model: "gpt-5.4-mini",
  temperature: 0.7,
});

// 使用基于步骤的配置创建智能体
const agent = createAgent({
  model,
  tools: allTools,
  stateSchema: SupportState,  
  middleware: [applyStepMiddleware],  
  checkpointer: new MemorySaver(),  
});

INFO

为什么要用检查点？ 检查点可在多个对话轮次之间维护状态。没有它，current_step 状态会在用户消息之间丢失，从而破坏工作流。

6. 测试工作流 ​

测试完整的工作流：

python
from langchain.messages import HumanMessage
from langchain_core.utils.uuid import uuid7

# 此对话线程的配置
thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}

# 第 1 轮：初始消息 - 从 warranty_collector 步骤开始
print("=== Turn 1: Warranty Collection ===")
result = agent.invoke(
    {"messages": [HumanMessage("Hi, my phone screen is cracked")]},
    config
)
for msg in result['messages']:
    msg.pretty_print()

# 第 2 轮：用户回复保修相关信息
print("\n=== Turn 2: Warranty Response ===")
result = agent.invoke(
    {"messages": [HumanMessage("Yes, it's still under warranty")]},
    config
)
for msg in result['messages']:
    msg.pretty_print()
print(f"Current step: {result.get('current_step')}")

# 第 3 轮：用户描述问题
print("\n=== Turn 3: Issue Description ===")
result = agent.invoke(
    {"messages": [HumanMessage("The screen is physically cracked from dropping it")]},
    config
)
for msg in result['messages']:
    msg.pretty_print()
print(f"Current step: {result.get('current_step')}")

# 第 4 轮：解决方案
print("\n=== Turn 4: Resolution ===")
result = agent.invoke(
    {"messages": [HumanMessage("What should I do?")]},
    config
)
for msg in result['messages']:
    msg.pretty_print()
typescript
import { HumanMessage } from "@langchain/core/messages";
import { v7 as uuid7 } from "uuid";

// 此对话线程的配置
const threadId = uuid7();
const config = { configurable: { thread_id: threadId } };

// 第 1 轮：初始消息 - 从 warranty_collector 步骤开始
console.log("=== Turn 1: Warranty Collection ===");
let result = await agent.invoke(
  { messages: [new HumanMessage("Hi, my phone screen is cracked")] },
  config
);
for (const msg of result.messages) {
  console.log(msg.content);
}

// 第 2 轮：用户回复保修相关信息
console.log("\n=== Turn 2: Warranty Response ===");
result = await agent.invoke(
  { messages: [new HumanMessage("Yes, it's still under warranty")] },
  config
);
for (const msg of result.messages) {
  console.log(msg.content);
}
console.log(`Current step: ${result.currentStep}`);

// 第 3 轮：用户描述问题
console.log("\n=== Turn 3: Issue Description ===");
result = await agent.invoke(
  { messages: [new HumanMessage("The screen is physically cracked from dropping it")] },
  config
);
for (const msg of result.messages) {
  console.log(msg.content);
}
console.log(`Current step: ${result.currentStep}`);

// 第 4 轮：解决方案
console.log("\n=== Turn 4: Resolution ===");
result = await agent.invoke(
  { messages: [new HumanMessage("What should I do?")] },
  config
);
for (const msg of result.messages) {
  console.log(msg.content);
}

预期流程：

  1. 保修验证步骤：询问保修状态
  2. 问题分类步骤：询问问题详情，判断是否为硬件问题
  3. 解决方案步骤：提供保修维修说明

7. 理解状态转换 ​

让我们追踪每一轮发生的情况：

第 1 轮：初始消息 ​

python
{
    "messages": [HumanMessage("Hi, my phone screen is cracked")],
    "current_step": "warranty_collector"  # 默认值
}
typescript
{
  messages: [new HumanMessage("Hi, my phone screen is cracked")],
  currentStep: "warranty_collector"  // 默认值
}

中间件应用：

  • 系统提示词：WARRANTY_COLLECTOR_PROMPT
  • 工具：[record_warranty_status]

第 2 轮：保修状态记录后 ​

Tool call: record_warranty_status("in_warranty") returns:

python
Command(update={
    "warranty_status": "in_warranty",
    "current_step": "issue_classifier"  # 状态转换！
})

Tool call: recordWarrantyStatus("in_warranty") returns:

typescript
new Command({
  update: {
    warrantyStatus: "in_warranty",
    currentStep: "issue_classifier"  // 状态转换！
  }
})

下一轮，中间件应用：

  • 系统提示词：ISSUE_CLASSIFIER_PROMPT（使用 warranty_status="in_warranty" 格式化）
  • 工具：[record_issue_type]

第 3 轮：问题分类后 ​

Tool call: record_issue_type("hardware") returns:

python
Command(update={
    "issue_type": "hardware",
    "current_step": "resolution_specialist"  # 状态转换！
})

Tool call: recordIssueType("hardware") returns:

typescript
new Command({
  update: {
    issueType: "hardware",
    currentStep: "resolution_specialist"  // 状态转换！
  }
})

下一轮，中间件应用：

  • 系统提示词：RESOLUTION_SPECIALIST_PROMPT（使用 warranty_status 和 issue_type 格式化）
  • 工具：[provide_solution, escalate_to_human]

关键要点：工具通过更新 current_step 驱动工作流，而中间件通过在下轮应用相应的配置做出响应。

8. 管理消息历史 ​

随着智能体逐步推进，消息历史会不断增长。使用摘要中间件在保留对话上下文的同时压缩较早的消息：

python
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware  
from langgraph.checkpoint.memory import InMemorySaver

agent = 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(),
)
typescript
import { createAgent, SummarizationMiddleware } from "langchain";  
import { MemorySaver } from "@langchain/langgraph";

const agent = createAgent({
  model,
  tools: allTools,
  stateSchema: SupportState,
  middleware: [
    applyStepMiddleware,
    new SummarizationMiddleware({  
      model: "gpt-5.4-mini",
      trigger: { tokens: 4000 },
      keep: { messages: 10 },
    }),
  ],
  checkpointer: new MemorySaver(),
});

其他记忆管理技术请参阅短期记忆指南。

9. 增加灵活性：返回上一步 ​

某些工作流需要允许用户返回上一步以更正信息（例如更改保修状态或问题分类）。然而，并非所有转换都有意义——例如，退款处理完毕后通常就无法返回。对于此支持工作流，我们将添加工具以返回到保修验证和问题分类步骤。

TIP

如果你的工作流需要在大多数步骤之间进行任意转换，请考虑是否真的需要结构化工作流。此模式最适合步骤遵循清晰顺序推进、偶尔为更正信息而向后退的情况。

为解决方案步骤添加"返回"工具：

python
@tool
def go_back_to_warranty() -> Command:  
    """Go back to warranty verification step."""
    return Command(update={"current_step": "warranty_collector"})  

@tool
def go_back_to_classification() -> Command:  
    """Go back to issue classification step."""
    return Command(update={"current_step": "issue_classifier"})  

# 更新 resolution_specialist 配置以包含这些工具
STEP_CONFIG["resolution_specialist"]["tools"].extend([
    go_back_to_warranty,
    go_back_to_classification
])
typescript
import { tool } from "langchain";
import { Command } from "@langchain/langgraph";
import { z } from "zod";

const goBackToWarranty = tool(  
  async () => {
    return new Command({ update: { currentStep: "warranty_collector" } });  
  },
  {
    name: "go_back_to_warranty",
    description: "Go back to warranty verification step.",
    schema: z.object({}),
  }
);

const goBackToClassification = tool(  
  async () => {
    return new Command({ update: { currentStep: "issue_classifier" } });  
  },
  {
    name: "go_back_to_classification",
    description: "Go back to issue classification step.",
    schema: z.object({}),
  }
);

// 更新 resolution_specialist 配置以包含这些工具
STEP_CONFIG.resolution_specialist.tools.push(
  goBackToWarranty,
  goBackToClassification
);

更新解决方案专家的提示词以提及这些工具：

python
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 step, 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

If the customer indicates any information was wrong, use:
- go_back_to_warranty to correct warranty status
- go_back_to_classification to correct issue type

Be specific and helpful in your solutions."""
typescript
const RESOLUTION_SPECIALIST_PROMPT = `You are a customer support agent helping with device issues.

CURRENT STAGE: Resolution
CUSTOMER INFO: Warranty status is {warrantyStatus}, issue type is {issueType}

At this step, 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

If the customer indicates any information was wrong, use:
- go_back_to_warranty to correct warranty status
- go_back_to_classification to correct issue type

Be specific and helpful in your solutions.`;

现在智能体可以处理更正信息：

python
result = agent.invoke(
    {"messages": [HumanMessage("Actually, I made a mistake - my device is out of warranty")]},
    config
)
# 智能体将调用 go_back_to_warranty 并重新开始保修验证步骤
typescript
const result = await agent.invoke(
  { messages: [new HumanMessage("Actually, I made a mistake - my device is out of warranty")] },
  config
);
// 智能体将调用 go_back_to_warranty 并重新开始保修验证步骤

完整示例 ​

以下是所有内容整合到可运行脚本中的完整示例：

python
"""
Customer Support State Machine Example

This 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 uuid7

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from typing import Callable, Literal
from typing_extensions import NotRequired

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse, SummarizationMiddleware
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool, ToolRuntime

model = init_chat_model("google_genai:gemini-3.6-flash")

# 定义可能的工作流步骤
SupportStep = 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"]]

@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_step": "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_step": "resolution_specialist",
        }
    )

@tool
def escalate_to_human(reason: str) -> str:
    """Escalate the case to a human support specialist."""
    # 在真实系统中，这会创建工单、通知工作人员等。
    return f"Escalating to human support. Reason: {reason}"

@tool
def provide_solution(solution: str) -> str:
    """Provide a solution to the customer's issue."""
    return f"Solution provided: {solution}"

# 将提示词定义为常量
WARRANTY_COLLECTOR_PROMPT = """You are a customer support agent helping with device issues.

CURRENT STEP: Warranty verification

At this step, 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 step

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 STEP: Issue classification
CUSTOMER INFO: Warranty status is {warranty_status}

At this step, 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 step

If unclear, ask clarifying questions before classifying."""

RESOLUTION_SPECIALIST_PROMPT = """You are a customer support agent helping with device issues.

CURRENT STEP: Resolution
CUSTOMER 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_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."""

# 步骤配置：将步骤名称映射到（提示词、工具、所需状态）
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_call
def apply_step_config(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    """Configure agent behavior based on the current step."""
    # 获取当前步骤（首次交互时默认为 warranty_collector）
    current_step = request.state.get("current_step", "warranty_collector")

    # 查找步骤配置
    step_config = STEP_CONFIG[current_step]

    # 验证所需的状态是否存在
    for key in step_config["requires"]:
        if request.state.get(key) is None:
            raise ValueError(f"{key} must be set before reaching {current_step}")

    # 使用状态值格式化提示词
    system_prompt = step_config["prompt"].format(**request.state)

    # 注入系统提示词和步骤专属工具
    request = request.override(
        system_prompt=system_prompt,
        tools=step_config["tools"],
    )

    return handler(request)

# 收集所有步骤配置中的全部工具
all_tools = [
    record_warranty_status,
    record_issue_type,
    provide_solution,
    escalate_to_human,
]

# 使用基于步骤的配置和摘要功能创建智能体
agent = 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(),
)

# ============================================================================
# 测试工作流
# ============================================================================

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()
typescript
import { createMiddleware, createAgent } from "langchain";

import { z } from "zod";
import { tool, ToolMessage, type ToolRuntime, HumanMessage } from "langchain";
import { Command, MemorySaver, StateSchema } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";

// 定义可能的工作流步骤
const SupportStepSchema = z.enum([
  "warranty_collector",
  "issue_classifier",
  "resolution_specialist",
]);
const WarrantyStatusSchema = z.enum(["in_warranty", "out_of_warranty"]);
const IssueTypeSchema = z.enum(["hardware", "software"]);

// 客户支持工作流的状态
const SupportState = new StateSchema({
  currentStep: SupportStepSchema.optional(),
  warrantyStatus: WarrantyStatusSchema.optional(),
  issueType: IssueTypeSchema.optional(),
});

const recordWarrantyStatus = tool(
  async (input, config: ToolRuntime<typeof SupportState.State>) => {
    return new Command({

      update: {

        messages: [
          new ToolMessage({
            content: `Warranty status recorded as: ${input.status}`,
            tool_call_id: config.toolCallId,
          }),
        ],
        warrantyStatus: input.status,
        currentStep: "issue_classifier",
      },
    });
  },
  {
    name: "record_warranty_status",
    description:
      "Record the customer's warranty status and transition to issue classification.",
    schema: z.object({
      status: WarrantyStatusSchema,
    }),
  }
);

const recordIssueType = tool(
  async (input, config: ToolRuntime<typeof SupportState.State>) => {
    return new Command({

      update: {

        messages: [
          new ToolMessage({
            content: `Issue type recorded as: ${input.issueType}`,
            tool_call_id: config.toolCallId,
          }),
        ],
        issueType: input.issueType,
        currentStep: "resolution_specialist",
      },
    });
  },
  {
    name: "record_issue_type",
    description:
      "Record the type of issue and transition to resolution specialist.",
    schema: z.object({
      issueType: IssueTypeSchema,
    }),
  }
);

const escalateToHuman = tool(
  async (input) => {
    // 在真实系统中，这会创建工单、通知工作人员等。
    return `Escalating to human support. Reason: ${input.reason}`;
  },
  {
    name: "escalate_to_human",
    description: "Escalate the case to a human support specialist.",
    schema: z.object({
      reason: z.string(),
    }),
  }
);

const provideSolution = tool(
  async (input) => {
    return `Solution provided: ${input.solution}`;
  },
  {
    name: "provide_solution",
    description: "Provide a solution to the customer's issue.",
    schema: z.object({
      solution: z.string(),
    }),
  }
);

// 将提示词定义为常量，便于引用
const WARRANTY_COLLECTOR_PROMPT = `You are a customer support agent helping with device issues.

CURRENT STAGE: Warranty verification

At this step, 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 step

Be conversational and friendly. Don't ask multiple questions at once.`;

const 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 step, 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 step

If unclear, ask clarifying questions before classifying.`;

const 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 step, 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.`;

// 步骤配置：将步骤名称映射到（提示词、工具、所需状态）
const STEP_CONFIG = {
  warranty_collector: {
    prompt: WARRANTY_COLLECTOR_PROMPT,
    tools: [recordWarrantyStatus],
    requires: [],
  },
  issue_classifier: {
    prompt: ISSUE_CLASSIFIER_PROMPT,
    tools: [recordIssueType],
    requires: ["warrantyStatus"],
  },
  resolution_specialist: {
    prompt: RESOLUTION_SPECIALIST_PROMPT,
    tools: [provideSolution, escalateToHuman],
    requires: ["warrantyStatus", "issueType"],
  },
} as const;

const applyStepMiddleware = createMiddleware({
  name: "applyStep",
  stateSchema: SupportState,
  wrapModelCall: async (request, handler) => {
    // 获取当前步骤（首次交互时默认为 warranty_collector）
    const currentStep = request.state.currentStep ?? "warranty_collector";

    // 查找步骤配置
    const stepConfig = STEP_CONFIG[currentStep];

    // 验证所需的状态是否存在
    for (const key of stepConfig.requires) {
      if (request.state[key] === undefined) {
        throw new Error(`${key} must be set before reaching ${currentStep}`);
      }
    }

    // 使用状态值格式化提示词（支持 {warrantyStatus}、{issueType} 等）
    let systemPrompt: string = stepConfig.prompt;
    for (const [key, value] of Object.entries(request.state)) {
      systemPrompt = systemPrompt.replace(`{${key}}`, String(value ?? ""));
    }

    // 注入系统提示词和步骤专属工具
    return handler({
      ...request,
      systemPrompt,
      tools: [...stepConfig.tools],
    });
  },
});

// 收集所有步骤配置中的全部工具
const allTools = [
  recordWarrantyStatus,
  recordIssueType,
  provideSolution,
  escalateToHuman,
];

const model = new ChatOpenAI({
  model: "gpt-5.4-mini",
});

// 使用基于步骤的配置创建智能体
const agent = createAgent({
  model,
  tools: allTools,
  middleware: [applyStepMiddleware],
  checkpointer: new MemorySaver(),
});

// 此对话线程的配置
const threadId = crypto.randomUUID();
const config = { configurable: { thread_id: threadId } };

// 第 1 轮：初始消息 - 从 warranty_collector 步骤开始
console.log("=== Turn 1: Warranty Collection ===");
let result = await agent.invoke(
  { messages: [new HumanMessage("Hi, my phone screen is cracked")] },
  config
);
for (const msg of result.messages) {
  console.log(msg.content);
}

// 第 2 轮：用户回复保修相关信息
console.log("\n=== Turn 2: Warranty Response ===");
result = await agent.invoke(
  { messages: [new HumanMessage("Yes, it's still under warranty")] },
  config
);
for (const msg of result.messages) {
  console.log(msg.content);
}
console.log(`Current step: ${result.currentStep}`);

// 第 3 轮：用户描述问题
console.log("\n=== Turn 3: Issue Description ===");
result = await agent.invoke(
  {
    messages: [
      new HumanMessage("The screen is physically cracked from dropping it"),
    ],
  },
  config
);
for (const msg of result.messages) {
  console.log(msg.content);
}
console.log(`Current step: ${result.currentStep}`);

// 第 4 轮：解决方案
console.log("\n=== Turn 4: Resolution ===");
result = await agent.invoke(
  { messages: [new HumanMessage("What should I do?")] },
  config
);
for (const msg of result.messages) {
  console.log(msg.content);
}

后续步骤 ​