使用 LangGraph 进行思考

from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command, RetryPolicy
from langchain_openai import ChatOpenAI
from langchain.messages import HumanMessage

llm = ChatOpenAI(model="gpt-5-nano")

def read_email(state: EmailAgentState) -> dict:
    """提取并解析邮件内容"""
    # 在生产环境中，这里会连接到您的邮件服务
    return {
        "messages": [HumanMessage(content=f"正在处理邮件: {state['email_content']}")]
    }

def classify_intent(state: EmailAgentState) -> Command[Literal["search_documentation", "human_review", "draft_response", "bug_tracking"]]:
    """使用 LLM 分类邮件意图和紧急程度，然后相应路由"""

    # 创建返回 EmailClassification 字典的结构化 LLM
    structured_llm = llm.with_structured_output(EmailClassification)

    # 按需格式化提示，而不是存储在状态中
    classification_prompt = f"""
    分析此客户邮件并进行分类：

    邮件: {state['email_content']}
    来自: {state['sender_email']}

    提供分类，包括意图、紧急程度、主题和摘要。
    """

    # 直接获取结构化响应作为字典
    classification = structured_llm.invoke(classification_prompt)

    # 根据分类确定下一个节点
    if classification['intent'] == 'billing' or classification['urgency'] == 'critical':
        goto = "human_review"
    elif classification['intent'] in ['question', 'feature']:
        goto = "search_documentation"
    elif classification['intent'] == 'bug':
        goto = "bug_tracking"
    else:
        goto = "draft_response"

    # 将分类作为单个字典存储在状态中
    return Command(
        update={"classification": classification},
        goto=goto
    )

def search_documentation(state: EmailAgentState) -> Command[Literal["draft_response"]]:
    """在知识库中搜索相关信息"""

    # 根据分类构建搜索查询
    classification = state.get('classification', {})
    query = f"{classification.get('intent', '')} {classification.get('topic', '')}"

    try:
        # 在此实现您的搜索逻辑
        # 存储原始搜索结果，而不是格式化后的文本
        search_results = [
            "通过 设置 > 安全 > 更改密码 重置密码",
            "密码必须至少包含 12 个字符",
            "包含大写字母、小写字母、数字和符号"
        ]
    except SearchAPIError as e:
        # 对于可恢复的搜索错误，存储错误并继续
        search_results = [f"搜索暂时不可用: {str(e)}"]

    return Command(
        update={"search_results": search_results},  # 存储原始结果或错误
        goto="draft_response"
    )

def bug_tracking(state: EmailAgentState) -> Command[Literal["draft_response"]]:
    """创建或更新缺陷跟踪工单"""

    # 在您的缺陷跟踪系统中创建工单
    ticket_id = "BUG-12345"  # 将通过 API 创建

    return Command(
        update={
            "search_results": [f"缺陷工单 {ticket_id} 已创建"],
            "current_step": "bug_tracked"
        },
        goto="draft_response"
    )

def draft_response(state: EmailAgentState) -> Command[Literal["human_review", "send_reply"]]:
    """使用上下文生成响应并根据质量路由"""

    classification = state.get('classification', {})

    # 根据状态中的原始数据按需格式化上下文
    context_sections = []

    if state.get('search_results'):
        # 为提示格式化搜索结果
        formatted_docs = "\n".join([f"- {doc}" for doc in state['search_results']])
        context_sections.append(f"相关文档:\n{formatted_docs}")

    if state.get('customer_history'):
        # 为提示格式化客户数据
        context_sections.append(f"客户层级: {state['customer_history'].get('tier', 'standard')}")

    # 使用格式化后的上下文构建提示
    draft_prompt = f"""
    起草对此客户邮件的回复:
    {state['email_content']}

    邮件意图: {classification.get('intent', 'unknown')}
    紧急程度: {classification.get('urgency', 'medium')}

    {chr(10).join(context_sections)}

    指南:
    - 保持专业和乐于助人
    - 针对他们的具体关切
    - 在相关时使用提供的文档
    """

    response = llm.invoke(draft_prompt)

    # 根据紧急程度和意图确定是否需要人工审核
    needs_review = (
        classification.get('urgency') in ['high', 'critical'] or
        classification.get('intent') == 'complex'
    )

    # 路由到适当的下一个节点
    goto = "human_review" if needs_review else "send_reply"

    return Command(
        update={"draft_response": response.content},  # 仅存储原始响应
        goto=goto
    )

def human_review(state: EmailAgentState) -> Command[Literal["send_reply", END]]:
    """使用 interrupt 暂停以供人工审核，并根据决策路由"""

    classification = state.get('classification', {})

    # interrupt() 必须在前 - 之前的代码在恢复时会重新运行
    human_decision = interrupt({
        "email_id": state.get('email_id',''),
        "original_email": state.get('email_content',''),
        "draft_response": state.get('draft_response',''),
        "urgency": classification.get('urgency'),
        "intent": classification.get('intent'),
        "action": "请审核并批准/编辑此回复"
    })

    # 现在处理人工的决策
    if human_decision.get("approved"):
        return Command(
            update={"draft_response": human_decision.get("edited_response", state.get('draft_response',''))},
            goto="send_reply"
        )
    else:
        # 拒绝意味着人工将直接处理
        return Command(update={}, goto=END)

def send_reply(state: EmailAgentState) -> dict:
    """发送邮件回复"""
    # 与邮件服务集成
    print(f"发送回复: {state['draft_response'][:100]}...")
    return {}