使用交接构建客户支持

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):
    """客户支持工作流的状态。"""
    current_step: NotRequired[SupportStep]
    warranty_status: NotRequired[Literal["in_warranty", "out_of_warranty"]]
    issue_type: NotRequired[Literal["hardware", "software"]]

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:
    """记录客户的保修状态并过渡到问题分类。"""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"保修状态已记录为: {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:
    """记录问题类型并过渡到解决方案专家。"""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"问题类型已记录为: {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:
    """将案例升级到人工支持专家。"""
    # 在实际系统中，这将创建工单、通知员工等。
    return f"正在升级到人工支持。原因: {reason}"


@tool
def provide_solution(solution: str) -> str:
    """为客户提供问题的解决方案。"""
    return f"已提供解决方案: {solution}"

# 步骤配置：将步骤名称映射到（提示词、工具、必需状态）
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"],
    },
}

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:
    """根据当前步骤配置智能体行为。"""
    # 获取当前步骤（首次交互默认为 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"在达到 {current_step} 之前必须设置 {key}")

    # 用状态值格式化提示词（支持 {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)

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(),
)

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("=== 轮次 1: 保修信息收集 ===")
result = agent.invoke(
    {"messages": [HumanMessage("你好，我的手机屏幕碎了")]},
    config
)
for msg in result['messages']:
    msg.pretty_print()

# 轮次 2: 用户回应保修信息
print("\n=== 轮次 2: 保修回应 ===")
result = agent.invoke(
    {"messages": [HumanMessage("是的，还在保修期内")]},
    config
)
for msg in result['messages']:
    msg.pretty_print()
print(f"当前步骤: {result.get('current_step')}")

# 轮次 3: 用户描述问题
print("\n=== 轮次 3: 问题描述 ===")
result = agent.invoke(
    {"messages": [HumanMessage("屏幕因为摔落而物理性碎裂")]},
    config
)
for msg in result['messages']:
    msg.pretty_print()
print(f"当前步骤: {result.get('current_step')}")

# 轮次 4: 解决方案
print("\n=== 轮次 4: 解决方案 ===")
result = agent.invoke(
    {"messages": [HumanMessage("我该怎么办？")]},
    config
)
for msg in result['messages']:
    msg.pretty_print()

{
    "messages": [HumanMessage("你好，我的手机屏幕碎了")],
    "current_step": "warranty_collector"  # 默认值
}

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

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

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(),
)

@tool
def go_back_to_warranty() -> Command:
    """返回保修验证步骤。"""
    return Command(update={"current_step": "warranty_collector"})


@tool
def go_back_to_classification() -> Command:
    """返回问题分类步骤。"""
    return Command(update={"current_step": "issue_classifier"})


# 更新 resolution_specialist 配置以包含这些工具
STEP_CONFIG["resolution_specialist"]["tools"].extend([
    go_back_to_warranty,
    go_back_to_classification
])

RESOLUTION_SPECIALIST_PROMPT = """你是一位帮助解决设备问题的客户支持代理。

当前阶段: 解决方案
客户信息: 保修状态为 {warranty_status}，问题类型为 {issue_type}

在此步骤中，你需要：
1. 对于软件问题：使用 provide_solution 提供故障排除步骤
2. 对于硬件问题：
   - 如果在保修期内：使用 provide_solution 解释保修维修流程
   - 如果超出保修期：escalate_to_human 处理付费维修选项

如果客户表示任何信息有误，请使用：
- go_back_to_warranty 更正保修状态
- go_back_to_classification 更正问题类型

提供具体且有帮助的解决方案。"""

result = agent.invoke(
    {"messages": [HumanMessage("实际上，我搞错了——我的设备超出保修期了")]},
    config
)
# 智能体将调用 go_back_to_warranty 并重新开始保修验证步骤