👥 客服机器人实战项目

// projects/customer-service-bot/data/faq.json
[
  {
    "question": "配送时间是多久？",
    "answer": "一般情况下，订单在提交后 1-2 个工作日内发货。国内配送通常需要 3-5 个工作日，偏远地区可能需要 5-7 个工作日。您可以在订单详情页查看实时物流信息。",
    "category": "配送"
  },
  {
    "question": "如何申请退货？",
    "answer": "您可以在订单详情页点击'申请售后'按钮，选择退货原因并提交。我们会在 1-2 个工作日内审核您的申请。退货商品需保持原包装完好，未使用，收到退货后 3-5 个工作日退款。",
    "category": "售后"
  },
  {
    "question": "支持哪些支付方式？",
    "answer": "我们支持多种支付方式：1）支付宝、微信支付 2）银行卡（借记卡/信用卡）3）货到付款（部分地区支持）。所有支付渠道均采用加密技术，保障您的资金安全。",
    "category": "支付"
  },
  {
    "question": "会员等级有什么权益？",
    "answer": "我们的会员体系分为普通、银卡、金卡、钻石四个等级。等级越高，享受的折扣越多（最高可达 9.5 折），并可获得专属客服、优先配送、生日礼品等特权。会员等级根据年度消费金额自动升级。",
    "category": "会员"
  },
  {
    "question": "如何修改收货地址？",
    "answer": "订单未发货前，您可以在订单详情页点击'修改地址'进行更改。如果订单已发货，请联系客服协助修改（可能产生额外运费）。建议在个人中心预先设置常用地址，下单时可快速选择。",
    "category": "订单管理"
  }
]

"""
FAQ 检索工具实现
功能：基于 RAG 的语义搜索
"""
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
from langchain.tools import tool
import json

# 加载 FAQ 数据
def load_faq_data(file_path: str) -> list[Document]:
    """加载 FAQ 数据并转换为 Document 对象"""
    with open(file_path, 'r', encoding='utf-8') as f:
        faq_list = json.load(f)

    documents = []
    for item in faq_list:
        # 将问题和答案组合成文档
        content = f"问题：{item['question']}\n答案：{item['answer']}"
        doc = Document(
            page_content=content,
            metadata={
                "category": item["category"],
                "question": item["question"]
            }
        )
        documents.append(doc)

    return documents

# 初始化向量存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
documents = load_faq_data("data/faq.json")

vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    collection_name="faq_knowledge"
)

retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 定义 FAQ 检索工具
@tool
def search_faq(query: str) -> str:
    """
    搜索常见问题（FAQ）知识库。

    适用于：
    - 配送时间、退换货政策、支付方式等常见问题
    - 会员权益、优惠活动等通用信息

    参数：
        query: 用户的问题描述

    返回：
        相关的 FAQ 答案
    """
    # 检索相关文档
    docs = retriever.invoke(query)

    if not docs:
        return "抱歉，我没有找到相关的常见问题解答。您可以详细描述您的问题，或者转接人工客服。"

    # 组合检索结果
    result = "根据常见问题库，我找到以下相关信息：\n\n"
    for i, doc in enumerate(docs, 1):
        result += f"{i}. {doc.metadata['question']}\n"
        result += f"   {doc.page_content.split('答案：')[1]}\n\n"

    return result.strip()

"""
测试 FAQ 检索工具
"""
# 直接调用工具
result = search_faq.invoke({"query": "我想知道多久能收到货"})
print(result)

# 预期输出：
# 根据常见问题库，我找到以下相关信息：
#
# 1. 配送时间是多久？
#    一般情况下，订单在提交后 1-2 个工作日内发货...
#
# 2. 如何修改收货地址？
#    订单未发货前，您可以在订单详情页点击'修改地址'...

"""
订单数据模拟（生产环境应连接真实数据库）
"""
import sqlite3
from datetime import datetime, timedelta

# 初始化数据库
def init_order_database():
    """创建订单数据库并插入测试数据"""
    conn = sqlite3.connect("data/orders.db")
    cursor = conn.cursor()

    # 创建订单表
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS orders (
            order_id TEXT PRIMARY KEY,
            user_id TEXT NOT NULL,
            product_name TEXT NOT NULL,
            quantity INTEGER NOT NULL,
            total_price REAL NOT NULL,
            status TEXT NOT NULL,
            shipping_address TEXT,
            tracking_number TEXT,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
    """)

    # 插入测试数据
    test_orders = [
        ("ORDER-2024001", "user_123", "iPhone 15 Pro", 1, 7999.0,
         "已发货", "北京市朝阳区xxx小区", "SF1234567890",
         (datetime.now() - timedelta(days=3)).isoformat(),
         (datetime.now() - timedelta(days=1)).isoformat()),

        ("ORDER-2024002", "user_123", "AirPods Pro 2", 1, 1899.0,
         "配送中", "北京市朝阳区xxx小区", "SF0987654321",
         (datetime.now() - timedelta(days=2)).isoformat(),
         datetime.now().isoformat()),

        ("ORDER-2024003", "user_123", "MacBook Pro 14", 1, 14999.0,
         "处理中", "北京市朝阳区xxx小区", None,
         (datetime.now() - timedelta(hours=6)).isoformat(),
         (datetime.now() - timedelta(hours=6)).isoformat()),
    ]

    cursor.executemany("""
        INSERT OR REPLACE INTO orders VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, test_orders)

    conn.commit()
    conn.close()
    print("订单数据库初始化完成")

# 执行初始化
init_order_database()

"""
订单查询工具集
功能：查询订单状态、物流信息
"""
from langchain.tools import tool
from langchain.tools import ToolRuntime
from typing_extensions import TypedDict
import sqlite3

# 定义上下文 Schema
class CustomerContext(TypedDict):
    user_id: str
    session_id: str

@tool
def get_my_orders(runtime: ToolRuntime[CustomerContext]) -> str:
    """
    查询当前用户的所有订单。

    返回：
        订单列表，包括订单号、商品名称、状态等信息
    """
    user_id = runtime.context["user_id"]

    conn = sqlite3.connect("data/orders.db")
    cursor = conn.cursor()

    cursor.execute("""
        SELECT order_id, product_name, quantity, total_price, status, created_at
        FROM orders
        WHERE user_id = ?
        ORDER BY created_at DESC
    """, (user_id,))

    orders = cursor.fetchall()
    conn.close()

    if not orders:
        return "您目前没有任何订单记录。"

    result = f"您共有 {len(orders)} 个订单：\n\n"
    for order in orders:
        order_id, product, qty, price, status, created = order
        result += f"📦 订单号：{order_id}\n"
        result += f"   商品：{product} x{qty}\n"
        result += f"   金额：¥{price:.2f}\n"
        result += f"   状态：{status}\n"
        result += f"   下单时间：{created[:10]}\n\n"

    return result.strip()

@tool
def get_order_details(order_id: str, runtime: ToolRuntime[CustomerContext]) -> str:
    """
    查询特定订单的详细信息，包括物流信息。

    参数：
        order_id: 订单号（例如：ORDER-2024001）

    返回：
        订单详细信息和物流状态
    """
    user_id = runtime.context["user_id"]

    conn = sqlite3.connect("data/orders.db")
    cursor = conn.cursor()

    cursor.execute("""
        SELECT * FROM orders
        WHERE order_id = ? AND user_id = ?
    """, (order_id, user_id))

    order = cursor.fetchone()
    conn.close()

    if not order:
        return f"未找到订单 {order_id}，请检查订单号是否正确。"

    _, _, product, qty, price, status, address, tracking, created, updated = order

    result = f"📦 订单详情\n\n"
    result += f"订单号：{order_id}\n"
    result += f"商品：{product} x{qty}\n"
    result += f"金额：¥{price:.2f}\n"
    result += f"状态：{status}\n"
    result += f"收货地址：{address}\n"
    result += f"下单时间：{created[:16]}\n"
    result += f"更新时间：{updated[:16]}\n"

    if tracking:
        result += f"\n🚚 物流信息\n"
        result += f"快递单号：{tracking}\n"
        result += f"物流状态：{status}\n"

        # 模拟物流轨迹
        if status == "已发货":
            result += "最新轨迹：您的包裹已从仓库发出，正在运输途中\n"
        elif status == "配送中":
            result += "最新轨迹：您的包裹已到达北京分拨中心，即将派送\n"

    return result

@tool
def cancel_order(order_id: str, reason: str, runtime: ToolRuntime[CustomerContext]) -> str:
    """
    取消订单（仅限未发货的订单）。

    参数：
        order_id: 订单号
        reason: 取消原因

    返回：
        取消结果
    """
    user_id = runtime.context["user_id"]

    conn = sqlite3.connect("data/orders.db")
    cursor = conn.cursor()

    # 检查订单状态
    cursor.execute("""
        SELECT status FROM orders
        WHERE order_id = ? AND user_id = ?
    """, (order_id, user_id))

    result = cursor.fetchone()

    if not result:
        conn.close()
        return f"未找到订单 {order_id}"

    status = result[0]

    if status != "处理中":
        conn.close()
        return f"订单 {order_id} 当前状态为 '{status}'，无法取消。已发货的订单请申请退货。"

    # 更新订单状态
    cursor.execute("""
        UPDATE orders
        SET status = '已取消', updated_at = ?
        WHERE order_id = ? AND user_id = ?
    """, (datetime.now().isoformat(), order_id, user_id))

    conn.commit()
    conn.close()

    return f"订单 {order_id} 已成功取消。取消原因：{reason}\n退款将在 3-5 个工作日内原路退回。"

"""
用户偏好管理（长期记忆）
功能：记住用户的语言偏好、常用地址等信息
"""
from langchain.tools import tool, ToolRuntime
from langgraph.store.memory import InMemoryStore

# 初始化 Store
store = InMemoryStore()

@tool
def save_user_preference(
    preference_type: str,
    preference_value: str,
    runtime: ToolRuntime[CustomerContext]
) -> str:
    """
    保存用户偏好设置。

    参数：
        preference_type: 偏好类型（language, default_address, notification 等）
        preference_value: 偏好值

    返回：
        保存结果
    """
    user_id = runtime.context["user_id"]
    store_obj = runtime.store

    # 获取现有偏好
    current_prefs = store_obj.get(("users", user_id), "preferences")

    if current_prefs:
        prefs = current_prefs.value
    else:
        prefs = {}

    # 更新偏好
    prefs[preference_type] = preference_value

    # 保存到 Store
    store_obj.put(("users", user_id), "preferences", prefs)

    return f"已保存您的偏好：{preference_type} = {preference_value}"

@tool
def get_user_preference(
    preference_type: str,
    runtime: ToolRuntime[CustomerContext]
) -> str:
    """
    获取用户偏好设置。

    参数：
        preference_type: 偏好类型

    返回：
        偏好值
    """
    user_id = runtime.context["user_id"]
    store_obj = runtime.store

    prefs = store_obj.get(("users", user_id), "preferences")

    if not prefs:
        return f"未设置 {preference_type} 偏好"

    value = prefs.value.get(preference_type)

    if not value:
        return f"未设置 {preference_type} 偏好"

    return f"您的 {preference_type} 偏好为：{value}"

@tool
def record_interaction_history(
    interaction_summary: str,
    runtime: ToolRuntime[CustomerContext]
) -> str:
    """
    记录重要的交互历史（用于后续分析和改进服务）。

    参数：
        interaction_summary: 交互摘要

    返回：
        记录结果
    """
    user_id = runtime.context["user_id"]
    store_obj = runtime.store

    # 获取历史记录
    history = store_obj.get(("users", user_id, "history"), "interactions")

    if history:
        interactions = history.value
    else:
        interactions = []

    # 添加新记录
    interactions.append({
        "timestamp": datetime.now().isoformat(),
        "summary": interaction_summary
    })

    # 保留最近 50 条
    interactions = interactions[-50:]

    # 保存
    store_obj.put(("users", user_id, "history"), "interactions", interactions)

    return "交互历史已记录"

"""
完整的电商客服机器人系统
整合：RAG + 工具 + 对话管理 + 流式输出 + 人工介入 + 长期记忆
"""
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from typing_extensions import TypedDict
import uuid

# ==================== 上下文定义 ====================

class CustomerContext(TypedDict):
    user_id: str
    session_id: str

# ==================== 工具集合 ====================

# 前面定义的所有工具
all_tools = [
    # FAQ 工具
    search_faq,

    # 订单工具
    get_my_orders,
    get_order_details,
    cancel_order,

    # 用户偏好工具
    save_user_preference,
    get_user_preference,
    record_interaction_history,
]

# ==================== 敏感操作审核配置 ====================

# 退款和取消订单需要人工审核
hitl_middleware = HumanInTheLoopMiddleware(
    interrupt_on={
        "cancel_order": True,  # 取消订单需要审核
        "search_faq": False,
        "get_my_orders": False,
        "get_order_details": False,
        "save_user_preference": False,
        "get_user_preference": False,
        "record_interaction_history": False,
    },
    description=lambda req: f"用户请求：{req['name']}({req['args']})"
)

# ==================== 创建客服 Agent ====================

# 初始化 Checkpointer 和 Store
checkpointer = InMemorySaver()
store = InMemoryStore()

customer_service_agent = create_agent(
    model="gpt-4o",
    tools=all_tools,
    middleware=[hitl_middleware],
    checkpointer=checkpointer,
    store=store,
    context_schema=CustomerContext,
    system_prompt="""你是一个专业的电商客服助手。

职责：
1. 解答客户关于配送、退换货、支付等常见问题（使用 search_faq 工具）
2. 查询和管理订单信息（使用订单相关工具）
3. 记住用户的偏好，提供个性化服务
4. 对于复杂或敏感的问题，及时转接人工客服

工作原则：
- 始终保持礼貌、专业、耐心
- 优先使用工具查询准确信息，不要猜测
- 对于订单取消、退款等敏感操作，需要人工审核
- 记录重要的交互历史，帮助改进服务
- 如果无法解决问题，主动建议转接人工

回答风格：
- 简洁明了，重点突出
- 使用列表和段落增强可读性
- 提供具体的操作步骤
"""
)

# ==================== 流式对话处理 ====================

def chat_with_customer_service(
    user_message: str,
    user_id: str = "user_123",
    thread_id: str = None
):
    """
    与客服机器人进行流式对话

    参数：
        user_message: 用户消息
        user_id: 用户 ID
        thread_id: 会话 ID（用于多轮对话）
    """
    if thread_id is None:
        thread_id = str(uuid.uuid4())

    config = {
        "configurable": {
            "thread_id": thread_id,
        },
        "context": {
            "user_id": user_id,
            "session_id": thread_id
        }
    }

    print(f"\n{'='*60}")
    print(f"用户: {user_message}")
    print('='*60)

    # 流式调用
    print("\n🤖 客服: ", end="", flush=True)

    full_response = ""
    interrupted = False

    for stream_mode, chunk in customer_service_agent.stream(
        {"messages": [{"role": "user", "content": user_message}]},
        config=config,
        stream_mode=["messages", "updates"]
    ):
        if stream_mode == "messages":
            token, metadata = chunk
            if token:
                print(token, end="", flush=True)
                full_response += token

        elif stream_mode == "updates":
            for step, data in chunk.items():
                if "__interrupt__" in data:
                    interrupted = True
                    print("\n\n⏸️ 需要人工审核，请稍候...")

    print("\n" + "="*60)

    return {
        "thread_id": thread_id,
        "response": full_response,
        "interrupted": interrupted
    }

# ==================== 测试场景 ====================

def run_test_scenarios():
    """运行多个测试场景"""

    print("\n" + "🧪 开始测试电商客服机器人".center(60, "="))

    # 场景 1：FAQ 问答
    print("\n\n📌 场景 1：查询配送时间")
    result1 = chat_with_customer_service(
        "我想知道订单一般多久能送到？",
        user_id="user_123"
    )
    thread_id = result1["thread_id"]

    input("\n按回车继续下一个场景...")

    # 场景 2：查询订单（多轮对话，使用相同 thread_id）
    print("\n\n📌 场景 2：查询我的订单")
    result2 = chat_with_customer_service(
        "帮我查一下我的订单",
        user_id="user_123",
        thread_id=thread_id  # 继续之前的对话
    )

    input("\n按回车继续下一个场景...")

    # 场景 3：查询特定订单详情
    print("\n\n📌 场景 3：查询订单详情")
    result3 = chat_with_customer_service(
        "帮我查一下 ORDER-2024002 的物流信息",
        user_id="user_123",
        thread_id=thread_id
    )

    input("\n按回车继续下一个场景...")

    # 场景 4：取消订单（会触发人工审核）
    print("\n\n📌 场景 4：取消订单（触发审核）")
    result4 = chat_with_customer_service(
        "我想取消订单 ORDER-2024003，因为买错了",
        user_id="user_123",
        thread_id=thread_id
    )

    if result4["interrupted"]:
        print("\n💡 提示：此操作需要人工审核")
        print("   在生产环境中，客服人员会在后台审核此请求")

        # 模拟审核通过
        from langgraph.types import Command
        print("\n✅ 模拟审核通过，继续执行...")

        approval_result = customer_service_agent.invoke(
            Command(resume={"decisions": [{"type": "approve"}]}),
            config={
                "configurable": {"thread_id": thread_id},
                "context": {"user_id": "user_123", "session_id": thread_id}
            }
        )

        print(f"\n最终结果: {approval_result['messages'][-1].content}")

    input("\n按回车继续下一个场景...")

    # 场景 5：保存用户偏好
    print("\n\n📌 场景 5：保存常用地址")
    result5 = chat_with_customer_service(
        "帮我把常用地址设为：北京市朝阳区xxx小区",
        user_id="user_123",
        thread_id=thread_id
    )

    print("\n\n" + "✅ 测试完成！".center(60, "="))
    print("\n系统演示了以下功能：")
    print("  1. ✅ FAQ 智能问答（RAG）")
    print("  2. ✅ 订单查询和管理")
    print("  3. ✅ 多轮对话（Checkpointer）")
    print("  4. ✅ 人工审核（HITL）")
    print("  5. ✅ 用户偏好（Store）")
    print("  6. ✅ 流式输出")

# ==================== 主程序 ====================

if __name__ == "__main__":
    # 初始化数据
    print("正在初始化系统...")
    init_order_database()
    print("系统初始化完成！\n")

    # 运行测试场景
    run_test_scenarios()

"""
生产环境配置示例
"""
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore
import os

# 数据库连接
DB_URI = os.getenv("DATABASE_URL", "postgresql://user:pass@localhost:5432/customer_service")

# 使用 PostgreSQL 持久化
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()

    with PostgresStore.from_conn_string(DB_URI) as store:
        store.setup()

        # 创建生产级 Agent
        production_agent = create_agent(
            model="gpt-4o",
            tools=all_tools,
            middleware=[
                hitl_middleware,
                # 可添加更多中间件：日志、监控、限流等
            ],
            checkpointer=checkpointer,
            store=store,
            context_schema=CustomerContext,
            system_prompt="""..."""  # 生产级提示词
        )

# 用户 A 的会话
config_a = {
    "configurable": {"thread_id": "user_a_session_001"},
    "context": {"user_id": "user_a"}
}

# 用户 B 的会话
config_b = {
    "configurable": {"thread_id": "user_b_session_002"},
    "context": {"user_id": "user_b"}
}

# 两个用户可以并发对话，互不影响
agent.invoke({...}, config=config_a)
agent.invoke({...}, config=config_b)

# 添加新的 FAQ
new_faq = Document(
    page_content="问题：xxx\n答案：xxx",
    metadata={"category": "新增"}
)

vectorstore.add_documents([new_faq])

# 删除过时的 FAQ
vectorstore.delete(ids=["outdated_faq_id"])