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"Processing email: {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"Search temporarily unavailable: {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"Bug ticket {ticket_id} created"],
            "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"Relevant documentation:\n{formatted_docs}")

    if state.get('customer_history'):
        # 为提示格式化客户数据
        context_sections.append(f"Customer tier: {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]]:
    """使用中断暂停以进行人工审核，并根据决策进行路由"""

    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": "Please review and approve/edit this response"
    })

    # 现在处理人工决策
    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"Sending reply: {state['draft_response'][:100]}...")
    return {}