一、目的

从最简单的调用api开始，不断加入中断执行、审批流程、输入验证、交互对话、反馈循环，实现良好人际交互

二、对话

2.1 调用API

客户端对话补全创造

注：1.是创造本次对话的完整回复，即只有一个回复，choices[0]；2.下面的n用不了

| 对象           | 常用参数          | 说明                     |
|----------------|-------------------|--------------------------|
| create()       | model, messages   | 模型，消息列表           |
|                | n                 | 生成几个回复选项         |
|                | temperature       | 随机性 (0-2)            |
|                | max_tokens        | 最大输出长度             |
|                | stream            | 是否流式输出             |
| response       | id, model         | 请求ID，模型名           |
|                | choices           | 回复选项列表             |
|                | usage             | token使用统计           |
| choices[]      | message           | 消息对象                 |
|                | finish_reason     | 结束原因                 |
| message        | role, content     | 角色，内容               |
| usage          | prompt_tokens     | 输入token数             |
|                | completion_tokens | 输出token数             |

#得充钱才行
os.environ["DEEPSEEK_API_KEY"] = "your api_key"


# Please install OpenAI SDK first: `pip3 install openai`
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get('DEEPSEEK_API_KEY'),
    base_url="https://api.deepseek.com")

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "You are a helpful assistant"},
        {"role": "user", "content": "Hello"},
    ],
    stream=False
)

print(response.choices[0].message.content)

2.2 LangGraph+api

LangGraph

状态：信息列表； 节点：调用api，将角色和信息返回状态； 图：api---end

api：客户端对话补全创造

import os
from typing import TypedDict
from openai import OpenAI
from langgraph.graph import StateGraph, END

# 初始化 DeepSeek 客户端
client = OpenAI(
    api_key=os.environ.get('DEEPSEEK_API_KEY'),
    base_url="https://api.deepseek.com"
)

# 定义状态
class GraphState(TypedDict):
    messages: list

# 节点函数 - 调用 DeepSeek API
def call_api_node(state: GraphState):
    """调用 DeepSeek API 的节点"""
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=state["messages"],
        stream=False,
    )
    
    # 将回复添加到消息列表
    state["messages"].append({
        "role": "assistant", 
        "content": response.choices[0].message.content
    })
    return state

# 创建图
def create_simple_graph():
    graph = StateGraph(GraphState)
    
    # 添加节点
    graph.add_node("call_api", call_api_node)
    
    # 设置入口点
    graph.set_entry_point("call_api")
    
    # 设置结束边
    graph.add_edge("call_api", END)
    
    return graph.compile()

# 使用图
if __name__ == "__main__":
    # 创建图
    app = create_simple_graph()
    
    # 初始状态
    initial_state = {
        "messages": [
            {"role": "system", "content": "You are a helpful assistant"},
            {"role": "user", "content": "Hello"},
        ]
    }
    
    # 执行图
    result = app.invoke(initial_state)
    
    # 打印结果
    print(result["messages"][-1]["content"])

2.3 多轮对话

中断和人工干预

状态：退出

节点

输入：查看是否有退出

回复：如果是退出，就返回

是否继续：看退出状态

做图：输入-回复-是否继续-end

import os
from typing import TypedDict
from openai import OpenAI
from langgraph.graph import StateGraph, END

client = OpenAI(
    api_key=os.environ.get('DEEPSEEK_API_KEY'),
    base_url="https://api.deepseek.com"
)

class ChatState(TypedDict):
    messages: list
    should_exit: bool  # 新增退出标志

def get_input_node(state: ChatState) -> ChatState:
    """获取用户输入"""
    user_input = input("👤 您: ")
    
    # 检查是否退出
    if user_input.lower() in ['退出', 'quit', 'exit', '结束']:
        return {
            "messages": state["messages"],
            "should_exit": True  # 设置退出标志
        }
    
    # 添加用户消息到历史
    new_messages = state["messages"] + [
        {"role": "user", "content": user_input}
    ]
    
    return {
        "messages": new_messages,
        "should_exit": False
    }

def ai_respond_node(state: ChatState) -> ChatState:
    """AI回复"""
    # 如果应该退出，直接返回
    if state.get("should_exit", False):
        return state
    
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=state["messages"],
        stream=False,
    )
    
    ai_reply = response.choices[0].message.content
    
    # 添加AI回复到历史
    new_messages = state["messages"] + [
        {"role": "assistant", "content": ai_reply}
    ]
    
    print(f"🤖 AI: {ai_reply}")
    
    return {
        "messages": new_messages,
        "should_exit": state.get("should_exit", False)
    }

def should_continue(state: ChatState) -> str:
    """决定是否继续对话"""
    # 检查退出标志
    if state.get("should_exit", False):
        return "end"
    else:
        return "continue"

# 创建多轮对话图
graph = StateGraph(ChatState)
graph.add_node("get_input", get_input_node)
graph.add_node("ai_respond", ai_respond_node)

graph.set_entry_point("get_input")
graph.add_edge("get_input", "ai_respond")

graph.add_conditional_edges(
    "ai_respond",
    should_continue,
    {
        "continue": "get_input",
        "end": END
    }
)

app = graph.compile()

# 使用
initial_state = {
    "messages": [
        {"role": "system", "content": "You are a helpful assistant"}
    ],
    "should_exit": False
}

print("=== 多轮对话开始 ===")
print("输入 '退出' 结束对话")

result = app.invoke(initial_state)
print("对话已结束！")

2.4 智能客服工单系统

# 工单系统状态
class TicketState(TypedDict):
    ticket_id: str
    customer_issue: str
    issue_category: str
    severity_level: str
    ai_solution: str
    requires_human_review: bool
    human_agent_assigned: Optional[str]
    resolution_status: str
    customer_feedback: Optional[str]
    processing_log: List[str]

def analyze_customer_issue(state: TicketState) -> TicketState:
    """分析客户问题"""
    issue = state.get("customer_issue", "")
    ticket_id = f"TK-{int(time.time())}"[-8:]
    
    # AI分析问题类别和严重程度
    if "无法登录" in issue or "密码" in issue:
        category = "账户问题"
        severity = "中等"
    elif "付款" in issue or "账单" in issue:
        category = "财务问题"
        severity = "高"
    elif "功能" in issue or "使用" in issue:
        category = "功能咨询"
        severity = "低"
    elif "故障" in issue or "错误" in issue:
        category = "技术故障"
        severity = "高"
    else:
        category = "一般咨询"
        severity = "低"
    
    print(f"📋 工单 {ticket_id}: 分析客户问题")
    print(f"   类别: {category}, 严重程度: {severity}")
    
    return {
        "ticket_id": ticket_id,
        "issue_category": category,
        "severity_level": severity,
        "resolution_status": "analyzing",
        "processing_log": ["问题分析完成"]
    }

def generate_ai_solution(state: TicketState) -> TicketState:
    """生成AI解决方案"""
    category = state.get("issue_category", "")
    severity = state.get("severity_level", "")
    
    # 根据类别生成解决方案
    solutions = {
        "账户问题": "请尝试以下步骤：1) 点击'忘记密码'重置密码 2) 清除浏览器缓存 3) 使用其他设备尝试登录",
        "财务问题": "建议：1) 检查账户余额和付款方式 2) 联系银行确认交易状态 3) 如需进一步帮助，将转接财务专员",
        "功能咨询": "使用指南：1) 查看帮助文档 2) 观看教学视频 3) 如需个性化指导，可安排专人协助",
        "技术故障": "故障处理：1) 重启应用程序 2) 检查网络连接 3) 更新到最新版本 4) 如问题持续，技术团队将介入",
        "一般咨询": "感谢您的咨询！我们的客服团队会尽快为您提供详细解答。"
    }
    
    ai_solution = solutions.get(category, "我们正在分析您的问题，请稍候...")
    
    # 判断是否需要人工审核
    requires_human = severity == "高" or category in ["财务问题", "技术故障"]
    
    print(f"🤖 AI方案: {ai_solution[:50]}...")
    print(f"🔍 需要人工审核: {'是' if requires_human else '否'}")
    
    return {
        "ai_solution": ai_solution,
        "requires_human_review": requires_human,
        "resolution_status": "solution_generated",
        "processing_log": state.get("processing_log", []) + ["AI方案生成完成"]
    }

def human_agent_review(state: TicketState) -> TicketState:
    """人工客服审核"""
    ticket_id = state.get("ticket_id", "")
    ai_solution = state.get("ai_solution", "")
    category = state.get("issue_category", "")
    
    # 模拟分配人工客服
    agents = ["Alice", "Bob", "Carol", "David"]
    assigned_agent = random.choice(agents)
    
    print(f"👨‍💼 人工客服 {assigned_agent} 审核工单 {ticket_id}")
    print(f"   审核AI方案: {ai_solution[:80]}...")
    
    # 模拟人工审核决定
    review_decisions = ["approve", "modify", "escalate"]
    decision = random.choice(review_decisions)
    
    if decision == "approve":
        print(f"✅ {assigned_agent}: AI方案已批准")
        status = "approved_by_human"
        log_msg = f"人工客服{assigned_agent}批准AI方案"
        
    elif decision == "modify":
        modified_solution = f"{ai_solution} [人工修改：建议优先联系技术支持团队]"
        print(f"✏️ {assigned_agent}: AI方案已修改")
        status = "modified_by_human"
        log_msg = f"人工客服{assigned_agent}修改了AI方案"
        
    else:  # escalate
        print(f"⬆️ {assigned_agent}: 问题升级至专业团队")
        status = "escalated"
        log_msg = f"人工客服{assigned_agent}将问题升级"
    
    return {
        "human_agent_assigned": assigned_agent,
        "resolution_status": status,
        "ai_solution": modified_solution if decision == "modify" else state.get("ai_solution", ""),
        "processing_log": state.get("processing_log", []) + [log_msg]
    }

def send_solution_to_customer(state: TicketState) -> TicketState:
    """向客户发送解决方案"""
    ticket_id = state.get("ticket_id", "")
    solution = state.get("ai_solution", "")
    
    print(f"📧 向客户发送工单 {ticket_id} 的解决方案")
    print(f"   方案内容: {solution[:100]}...")
    
    # 模拟客户反馈
    feedback_options = ["满意", "部分解决", "未解决"]
    customer_feedback = random.choice(feedback_options)
    
    print(f"📝 客户反馈: {customer_feedback}")
    
    if customer_feedback == "满意":
        status = "resolved"
    elif customer_feedback == "部分解决":
        status = "partially_resolved"
    else:
        status = "unresolved"
    
    return {
        "customer_feedback": customer_feedback,
        "resolution_status": status,
        "processing_log": state.get("processing_log", []) + [f"解决方案已发送，客户反馈：{customer_feedback}"]
    }

def escalate_to_specialist(state: TicketState) -> TicketState:
    """升级给专家处理"""
    ticket_id = state.get("ticket_id", "")
    category = state.get("issue_category", "")
    
    specialists = {
        "技术故障": "技术支持团队",
        "财务问题": "财务专员",
        "账户问题": "账户安全专家",
    }
    
    specialist = specialists.get(category, "高级客服专员")
    
    print(f"🎯 工单 {ticket_id} 已升级至 {specialist}")
    print(f"   预计24小时内获得专业解答")
    
    return {
        "resolution_status": "escalated_to_specialist",
        "human_agent_assigned": specialist,
        "processing_log": state.get("processing_log", []) + [f"升级至{specialist}"]
    }

# 决策函数
def decide_ticket_flow(state: TicketState) -> str:
    """决定工单处理流程"""
    status = state.get("resolution_status", "")
    requires_human = state.get("requires_human_review", False)
    
    if status == "analyzing":
        return "generate_solution"
    elif status == "solution_generated" and requires_human:
        return "human_review"
    elif status == "solution_generated" and not requires_human:
        return "send_solution"
    elif status in ["approved_by_human", "modified_by_human"]:
        return "send_solution"
    elif status == "escalated":
        return "escalate_specialist"
    else:
        return "end"

# 创建工单处理系统
def create_ticket_system():
    checkpointer = MemorySaver()
    graph = StateGraph(TicketState)
    
    graph.add_node("analyze", analyze_customer_issue)
    graph.add_node("generate_solution", generate_ai_solution)
    graph.add_node("human_review", human_agent_review)
    graph.add_node("send_solution", send_solution_to_customer)
    graph.add_node("escalate_specialist", escalate_to_specialist)
    
    graph.set_entry_point("analyze")
    
    # 添加条件边
    for node in ["analyze", "generate_solution", "human_review"]:
        graph.add_conditional_edges(
            node,
            decide_ticket_flow,
            {
                "generate_solution": "generate_solution",
                "human_review": "human_review",
                "send_solution": "send_solution",
                "escalate_specialist": "escalate_specialist",
                "end": END
            }
        )
    
    graph.add_edge("send_solution", END)
    graph.add_edge("escalate_specialist", END)
    
    return graph.compile(checkpointer=checkpointer)

# 测试工单系统
ticket_app = create_ticket_system()

print("\n=== 智能客服工单系统演示 ===")

# 测试不同类型的客户问题
test_issues = [
    "我无法登录账户，一直提示密码错误",
    "付款时遇到错误，订单状态异常",
    "如何使用新功能进行数据导出？",
    "系统出现严重故障，无法正常工作",
    "想了解你们的服务包含哪些内容"
]

for i, issue in enumerate(test_issues):
    print(f"\n{'='*20} 工单 {i+1} {'='*20}")
    print(f"客户问题: {issue}")
    
    ticket_thread_id = str(uuid.uuid4())
    ticket_config = {"configurable": {"thread_id": ticket_thread_id}}
    
    result = ticket_app.invoke({
        "customer_issue": issue
    }, config=ticket_config)
    
    print(f"\n📊 处理结果:")
    print(f"   工单ID: {result.get('ticket_id', 'N/A')}")
    print(f"   问题分类: {result.get('issue_category', 'N/A')}")
    print(f"   严重程度: {result.get('severity_level', 'N/A')}")
    print(f"   最终状态: {result.get('resolution_status', 'N/A')}")
    print(f"   分配客服: {result.get('human_agent_assigned', '无')}")
    print(f"   客户反馈: {result.get('customer_feedback', '未收到')}")
    print(f"   处理步骤: {' -> '.join(result.get('processing_log', []))}")