用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"相关文档：\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"Sending reply: {state['draft_response'][:100]}...")
    return {}