这是一个集成了真实大语言模型的LangGraph高级案例，展示如何构建一个企业级的智能客服助手系统。本案例从基础的聊天机器人进阶到完整的客服工作流系统，具备以下核心功能：

核心功能模块：

  1. 问题理解与分类 - 使用LLM智能识别问题类型

    • 多维度分类：技术支持、账户问题、产品咨询、投诉建议等
    • 置信度评估：确保分类的准确性和可靠性
    • 动态分类规则：支持新问题类型的快速适配

  2. 情感分析 - 识别用户情绪和紧急程度

    • 情感维度：积极、中性、消极、沮丧、愤怒等多层次情感识别
    • 紧急度评估：low/medium/high 三级紧急程度判断
    • 情感变化追踪：监控对话过程中的情感变化趋势

  3. 知识库搜索 - 基于分类和关键词的智能搜索

    • 多策略搜索：关键词匹配、语义搜索、同义词扩展
    • 相关性评分：基于多种因素的智能排序算法
    • 动态索引：支持知识库的实时更新和优化

  4. 智能回答生成 - 结合上下文和知识库的回答

    • 上下文感知：基于对话历史和用户状态生成个性化回答
    • 模板化回答：确保回答的专业性和一致性
    • 情感适应：根据用户情绪调整回答的语调和风格

  5. 升级判断 - 多维度评估是否需要人工客服

    • 智能路由：基于多个维度的综合评估决策
    • 优先级管理：确保重要问题得到及时处理
    • 升级策略：灵活的升级规则和降级机制

核心架构特点：

企业级设计理念：

  • 真实LLM集成（OpenAI GPT/通义千问/文心一言等）
  • 复杂的多节点工作流设计
  • 基于多因素的智能路由决策
  • 完整的状态管理和追踪系统
  • 结构化的LLM输出解析
  • 错误处理和降级策略

可扩展性考虑：

  • 模块化的节点设计，便于功能扩展
  • 标准化的状态接口，支持新节点的无缝集成
  • 配置化的决策规则，支持业务规则的动态调整
  • 插件式的工具集成，便于添加新的外部服务

高可用性保障：

  • 多层次的错误处理和恢复机制
  • 优雅的降级策略，确保系统的持续可用
  • 完善的日志和监控体系
  • 性能优化和资源管理

技术亮点：

提示工程创新：

  • JSON结构化提示工程：确保LLM输出的一致性和可解析性
  • 上下文窗口优化：有效管理长对话的上下文信息
  • 多轮对话支持：保持对话的连贯性和上下文记忆
  • 错误恢复提示：处理LLM输出异常的自动修复机制

LLM集成最佳实践：

  • 多提供商适配：支持OpenAI、通义千问、文心一言等多种LLM
  • 统一接口抽象：屏蔽不同LLM的API差异
  • 成本控制策略：智能的请求优化和缓存机制
  • 性能监控：实时监控LLM调用的性能和成本

工作流优化：

  • 状态驱动的决策逻辑：基于完整状态信息的智能决策
  • 知识库智能匹配算法：多维度的知识检索和排序
  • 情感驱动的服务升级：基于情感分析的动态服务调整
  • 全流程可观测性：完整的处理过程记录和分析

业务价值：

客户体验提升：

  • 24/7 全天候服务：无间断的客户支持
  • 快速响应时间：秒级的问题分析和回答
  • 个性化服务：基于用户历史和偏好的定制化回答
  • 情感感知服务：根据用户情绪提供适当的关怀

运营效率优化：

  • 智能分流：自动识别并分配合适的处理方式
  • 工作负载均衡：合理分配人工和AI的工作量
  • 知识管理：统一的知识库和经验积累
  • 质量监控：全面的服务质量评估和改进

成本控制：

  • 人力成本降低：AI处理大部分常规问题
  • 培训成本减少：标准化的服务流程和知识库
  • 响应时间缩短：减少客户等待时间
  • 满意度提升：提高客户满意度和忠诚度

学习目标：

初级开发者：

  • 理解企业级LangGraph应用的设计模式
  • 掌握真实LLM的集成和调用方法
  • 学会结构化提示工程的最佳实践
  • 了解客服业务的核心流程和挑战

中级开发者：

  • 掌握复杂状态管理和工作流设计
  • 学会多维度决策逻辑的实现
  • 理解企业级应用的错误处理策略
  • 掌握性能优化和成本控制技巧

高级开发者：

  • 学会可扩展架构的设计原则
  • 掌握多模态和多轮对话的处理技术
  • 理解企业级部署和运维考虑
  • 学会系统监控和持续优化方法

实际应用场景：

电商客服：

  • 订单查询和处理
  • 产品咨询和推荐
  • 售后服务和退换货
  • 投诉处理和满意度调查

金融客服：

  • 账户查询和操作指导
  • 产品介绍和风险提示
  • 交易问题和争议处理
  • 合规要求和安全提醒

技术支持：

  • 产品使用指导
  • 故障诊断和解决
  • 技术文档和资源推荐
  • 版本更新和新功能介绍

企业内部服务：

  • IT支持和设备管理
  • HR政策咨询和流程指导
  • 财务报销和审批流程
  • 行政服务和资源申请

扩展方向：

技术扩展：

  • 集成语音识别和合成：支持语音客服
  • 多模态理解：图片、文档、视频的智能处理
  • 实时翻译：多语言客服支持
  • 知识图谱：更智能的知识推理和关联

业务扩展：

  • CRM集成：客户信息管理和历史记录
  • 工单系统：完整的问题跟踪和处理流程
  • 分析dashboard：服务质量和效率分析
  • A/B测试：持续优化和改进验证

智能化升级：

  • 机器学习模型：个性化推荐和预测
  • 自动化工作流：复杂业务流程的自动化
  • 智能质检：自动化的服务质量评估
  • 预测性服务：主动识别和解决潜在问题

这个案例不仅是技术演示，更是企业级AI客服系统的完整实现参考，
涵盖了从技术架构到业务应用的各个层面，为实际项目提供宝贵的经验和指导。

import os
import json
from typing import Dict, List, TypedDict, Annotated, Literal
from datetime import datetime

# LangChain 核心组件 - 提供LLM抽象和消息管理
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser, StrOutputParser

# LangGraph 组件 - 提供工作流编排能力
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

LLM 提供商集成 - 支持多种LLM服务，适配不同的LLM提供商，体现了企业级应用的兼容性考虑

try:
    # OpenAI API兼容的服务（包括各种国产大模型）
    # 这种导入方式支持OpenAI官方API以及兼容OpenAI接口的其他服务
    from langchain_openai import ChatOpenAI
    LLM_AVAILABLE = "openai"
except ImportError:
    try:
        # 阿里云通义千问 - 国产大模型的集成示例
        from langchain_community.llms import Tongyi
        LLM_AVAILABLE = "tongyi"
    except ImportError:
        try:
            # 百度文心一言 - 另一个主流国产大模型
            from langchain_community.chat_models import QianfanChatEndpoint
            LLM_AVAILABLE = "qianfan"
        except ImportError:
            # 如果所有LLM都不可用，系统会使用模拟模式
            LLM_AVAILABLE = None

配置和初始化

配置类 - 集中管理所有配置参数，企业级配置管理设计原则：

  1. 分离关注点（Separation of Concerns）：

    • 将配置与业务逻辑分离
    • 不同类型的配置分组管理
    • 便于不同环境的配置切换

  2. 安全性考虑：

    • 敏感信息（如API密钥）应从环境变量读取
    • 支持配置加密和访问控制
    • 避免在代码中硬编码敏感信息

  3. 可维护性：

    • 配置项命名规范，易于理解
    • 提供合理的默认值
    • 支持配置验证和类型检查

  4. 扩展性：

    • 支持动态配置更新
    • 便于添加新的配置项
    • 支持配置的继承和覆盖

实际应用中的配置管理最佳实践：

  1. 多环境支持：

    • 开发环境（dev）：使用测试数据和宽松的配置
    • 测试环境（test）：模拟生产环境的配置
    • 生产环境（prod）：严格的安全和性能配置

  2. 配置来源优先级：

    • 环境变量 > 配置文件 > 默认值
    • 支持配置的动态覆盖和热更新
    • 提供配置来源的可追溯性

  3. 配置验证：

    • 启动时验证配置的完整性和正确性
    • 提供配置错误的详细诊断信息
    • 支持配置的向后兼容性检查

  4. 监控和审计：

    • 记录配置的变更历史
    • 监控配置的使用情况和性能影响
    • 支持配置回滚和紧急修复
class Config:
    # === API配置 ===
    # 注意：在生产环境中，API密钥应该从环境变量获取，而不是硬编码
    # 示例：OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    OPENAI_API_KEY = "123-********-4576"
    OPENAI_BASE_URL = "https://ark.cn-beijing.volces.com/api/v3"
    
    # === 模型配置 ===
    # 这些参数控制LLM的行为特征，需要根据具体业务需求调优
    MODEL_NAME = "deepseek-v3-250324"    # 模型名称 - 选择合适的模型版本
    TEMPERATURE = 0.7                     # 控制输出随机性，0.7提供创造性和一致性的平衡
                                         # 0.0 = 完全确定性，1.0 = 高度随机性
    MAX_TOKENS = 1000                     # 限制输出长度，控制成本和响应时间

初始化大语言模型

LLM初始化的企业级设计考虑：

  1. 多提供商支持策略：

    • 主备模式：主要LLM失败时自动切换到备用LLM
    • 负载均衡：根据性能和成本在多个LLM间分配请求
    • 能力适配：根据任务类型选择最适合的LLM

  2. 初始化最佳实践：

    • 延迟初始化：只在实际使用时创建LLM实例
    • 连接池管理：复用连接，减少建立连接的开销
    • 超时配置：合理设置请求超时时间
    • 重试机制：处理网络异常和临时故障

  3. 成本控制策略：

    • 请求缓存：缓存相似请求的结果
    • 批量处理：合并多个小请求减少API调用次数
    • 模型选择：根据任务复杂度选择合适的模型
    • 使用监控：实时监控API使用量和成本

  4. 性能优化：

    • 异步调用：使用异步I/O提高并发性能
    • 流式输出：对于长文本生成使用流式API
    • 预热机制：系统启动时预热模型连接
    • 性能监控：监控响应时间和成功率

降级策略设计：

  1. 服务降级层次：

    • Level 1：主要LLM服务正常
    • Level 2：切换到备用LLM服务
    • Level 3：使用本地模型或缓存结果
    • Level 4：返回预定义的安全回答

  2. 降级触发条件：

    • API错误率超过阈值
    • 响应时间超过预期
    • 成本超出预算限制
    • 服务维护或升级

  3. 降级恢复机制：

    • 健康检查：定期检测服务恢复状态
    • 逐步恢复：从低风险请求开始恢复
    • 性能验证：确保恢复后的服务质量
    • 用户通知：适当的用户体验管理
def initialize_llm():
    if LLM_AVAILABLE == "openai":
        # 使用OpenAI兼容的API（可以是OpenAI、Azure、或其他兼容服务）
        # 这种配置方式支持多种兼容OpenAI API的服务，包括：
        # - OpenAI 官方API
        # - Azure OpenAI Service
        # - 各种国产大模型的OpenAI兼容接口
        # - 私有部署的兼容服务
        return ChatOpenAI(
            model=Config.MODEL_NAME,
            temperature=Config.TEMPERATURE,
            max_tokens=Config.MAX_TOKENS,
            openai_api_key=Config.OPENAI_API_KEY,
            openai_api_base=Config.OPENAI_BASE_URL
        )
    else:
        # 如果没有可用的LLM，返回模拟器
        # 这确保了系统的可用性，即使在没有API访问的情况下也能演示
        # 在实际应用中，这里应该实现更完善的降级策略
        return MockLLM()

模拟LLM类 - 用于演示和测试

模拟服务的设计价值：

  1. 开发阶段支持：

    • 离线开发：无需网络连接即可开发和测试
    • 成本控制：避免开发阶段的API调用费用
    • 稳定测试：提供一致和可预测的测试结果
    • 功能验证：专注于业务逻辑而非API集成

  2. 系统可靠性：

    • 降级保障：在真实LLM不可用时提供基础服务
    • 性能基准：作为性能对比的基准线
    • 错误处理：测试系统在异常情况下的表现
    • 文档演示：提供一致的演示效果

  3. 模拟策略设计：

    • 真实性：模拟结果应该尽可能接近真实LLM的输出
    • 多样性：提供多种类型的模拟响应
    • 可配置性：支持不同场景的模拟配置
    • 调试友好：提供清晰的模拟逻辑和状态

企业级模拟服务实现：

  1. 数据驱动：

    • 使用真实的历史数据训练模拟模型
    • 基于统计分析生成符合分布的随机响应
    • 支持A/B测试和效果验证

  2. 动态响应：

    • 根据输入内容生成相应的模拟输出
    • 支持上下文感知的响应生成
    • 模拟真实LLM的响应时间特征

  3. 配置化管理：

    • 支持不同场景的模拟配置
    • 允许运行时调整模拟行为
    • 提供模拟质量的评估指标
class MockLLM:
   def invoke(self, messages):
        """
        模拟LLM调用 - 根据输入类型返回相应的模拟响应
        
        智能响应生成策略：
        =================
        
        1. 内容分析：
           - 关键词匹配：识别输入中的关键概念
           - 意图识别：理解用户的查询意图
           - 上下文理解：考虑对话的历史记录
        
        2. 响应生成：
           - 模板匹配：基于预定义模板生成响应
           - 动态组合：组合多个信息片段形成完整响应
           - 格式保证：确保输出格式的一致性和可解析性
        
        3. 质量控制：
           - 长度适中：控制响应的长度符合预期
           - 逻辑一致：确保响应在逻辑上合理
           - 格式正确：特别是JSON格式的正确性
        
        模拟响应的业务价值：
        ===================
        
        1. 功能验证：
           - 验证工作流的完整性和正确性
           - 测试异常情况的处理能力
           - 评估系统的性能特征
        
        2. 用户体验：
           - 在真实LLM不可用时提供基础服务
           - 为用户提供一致的交互体验
           - 避免系统完全不可用的情况
        
        3. 开发效率：
           - 加速开发和测试周期
           - 降低开发和调试成本
           - 提供稳定的开发环境
        """
        
        # 处理不同类型的输入格式
        # 这种兼容性设计确保了模拟器能够处理各种调用方式
        if isinstance(messages, list):
            user_content = messages[-1].content if messages else ""
        else:
            user_content = str(messages)
        
        # 根据内容类型返回不同的模拟响应
        # 这些响应模拟了真实LLM可能产生的结构化输出
        if "分类" in user_content or "category" in user_content.lower():
            # 模拟问题分类的JSON响应
            # 这种结构化的响应确保了后续处理的一致性
            return AIMessage(content='{"category": "技术支持", "confidence": 0.85, "reason": "用户询问技术相关问题"}')
        elif "情感" in user_content or "sentiment" in user_content.lower():
            # 模拟情感分析的JSON响应
            # 包含多个维度的情感信息，便于后续的决策处理
            return AIMessage(content='{"sentiment": "neutral", "score": 0.6, "emotion": "平静"}')
        else:
            # 默认的文本回答
            # 提供友好的用户体验，即使在模拟模式下也要保持专业性
            return AIMessage(content="这是一个模拟的LLM响应。请配置真实的API密钥以获得更好的效果。")

状态定义

客服助手状态定义 - 复杂业务场景的状态设计
企业级状态设计哲学：

  1. 业务完整性原则：

    • 全流程覆盖：状态应该覆盖整个客服流程的所有环节
    • 数据完整性：确保重要的业务数据不会丢失
    • 状态一致性：不同节点对状态的理解应该一致
    • 版本兼容性：支持状态结构的演进和升级

  2. 可扩展性设计：

    • 模块化结构：将相关的状态字段分组管理
    • 预留扩展：为未来的功能扩展预留空间
    • 嵌套支持：支持复杂的嵌套数据结构
    • 配置灵活性：允许运行时的状态配置调整

  3. 性能考虑：

    • 内存效率：避免存储过大的状态对象
    • 序列化优化：确保状态的高效序列化和反序列化
    • 访问模式：优化常用字段的访问性能
    • 缓存友好：设计便于缓存的状态结构

客服业务状态设计考虑：

  1. 客户体验相关：

    • 完整的对话历史记录
    • 情感变化的跟踪记录
    • 个性化偏好和历史
    • 服务质量评估数据

  2. 业务流程管理：

    • 问题分类和路由信息
    • 处理优先级和时限
    • 升级条件和触发机制
    • 质量监控和评估指标

  3. 运营数据分析：

    • 完整的处理步骤记录
    • 性能指标和时间统计
    • 决策过程和依据记录
    • 异常情况和错误日志

实际应用中的状态扩展：

  1. 用户维度扩展：

    • 用户画像和偏好
    • 历史问题和解决方案
    • 满意度和反馈记录
    • VIP等级和特殊需求

  2. 业务维度扩展：

    • 产品信息和版本
    • 订单状态和历史
    • 账户信息和权限
    • 合规要求和限制

  3. 技术维度扩展：

    • 渠道信息（Web、App、电话等）
    • 设备信息和环境
    • 性能指标和监控数据
    • 安全验证和审计日志
class CustomerServiceState(TypedDict):
    # === 基本信息 ===
    # 这部分存储会话的基础标识和时间信息，用于追踪和管理
    user_id: str                    # 用户唯一标识 - 支持个性化服务和历史追踪
    session_id: str                 # 会话唯一标识 - 用于追踪单次对话的完整过程
    timestamp: str                  # 时间戳 - 记录处理开始时间，支持性能分析和审计
    
    # === 用户输入 ===
    # 存储用户的原始输入和对话历史，是整个处理流程的数据源
    user_question: str              # 用户的原始问题 - 保持问题的原始性和完整性
    messages: Annotated[List[BaseMessage], add_messages]  # 完整的对话历史
                                   # 使用add_messages确保消息的自动管理和去重
    
    # === 分析结果 ===
    # 存储各种AI分析的结果，是后续决策的重要依据
    question_category: str          # 问题分类结果（技术支持/账户问题/产品咨询/投诉建议等）
                                   # 支持多级分类和动态分类规则
    category_confidence: float      # 分类的置信度（0-1）- 用于评估分类结果的可靠性
    sentiment_analysis: Dict        # 情感分析的详细结果
                                   # 包含情感倾向、强度、关键词等多维度信息
    
    # === 知识库相关 ===
    # 管理知识检索和应用的相关信息
    knowledge_results: List[Dict]   # 知识库搜索的结果列表
                                   # 每个结果包含内容、来源、相关性评分等信息
    relevant_info: str             # 提取的相关信息摘要
                                   # 格式化后的知识内容，便于生成回答
    
    # === 决策相关 ===
    # 存储各种业务决策的结果和依据
    need_human_handover: bool      # 是否需要转接人工客服
                                   # 基于多维度评估的最终决策结果
    urgency_level: str             # 紧急程度级别（low/medium/high）
                                   # 影响处理优先级和资源分配
    
    # === 输出结果 ===
    # 存储最终的输出和建议，直接影响用户体验
    ai_response: str               # AI生成的最终回答
                                   # 经过多轮优化和格式化的用户友好回答
    suggested_actions: List[str]   # 系统建议的后续操作
                                   # 为用户或客服人员提供的行动建议
    
    # === 元数据和追踪 ===
    # 记录处理过程的详细信息，支持监控、调试和优化
    processing_steps: List[str]    # 处理步骤的详细记录
                                   # 完整的流程追踪，便于问题排查和流程优化
    confidence_scores: Dict        # 各环节的置信度分数
                                   # 量化的质量评估，支持持续改进

知识库模拟

模拟企业知识库 - 结构化的知识管理，在实际应用中，这通常是：

  • 关系数据库（MySQL、PostgreSQL）
  • 文档数据库（MongoDB、Elasticsearch）
  • 向量数据库（Pinecone、Weaviate、Chroma）
  • 图数据库（Neo4j）
KNOWLEDGE_BASE = {
    "技术支持": [
        {
            "id": "tech_001",
            "title": "登录问题解决方案",
            "content": "如果无法登录，请先检查用户名密码是否正确，然后尝试清除浏览器缓存，如果问题持续存在，请联系技术支持。",
            "tags": ["登录", "密码", "账户"],
            "priority": "high",          # 优先级
            "last_updated": "2024-01-15" # 最后更新时间
        },
        {
            "id": "tech_002", 
            "title": "网络连接故障排查",
            "content": "网络连接问题可能由多种原因造成：1）检查网络连接状态；2）重启路由器；3）检查防火墙设置；4）联系网络服务提供商。",
            "tags": ["网络", "连接", "故障"],
            "priority": "medium",
            "last_updated": "2024-01-10"
        },
        {
            "id": "tech_003",
            "title": "软件安装指南", 
            "content": "软件安装步骤：1）下载官方安装包；2）以管理员身份运行；3）按照向导完成安装；4）重启系统。如有问题请查看安装日志。",
            "tags": ["软件", "安装", "设置"],
            "priority": "low",
            "last_updated": "2024-01-05"
        }
    ],
    "账户问题": [
        {
            "id": "account_001",
            "title": "忘记密码处理流程",
            "content": "忘记密码可以通过以下方式重置：1）点击登录页面的'忘记密码'；2）输入注册邮箱；3）查收重置邮件；4）按邮件指引重置密码。",
            "tags": ["密码", "重置", "邮箱"],
            "priority": "high",
            "last_updated": "2024-01-12"
        },
        {
            "id": "account_002",
            "title": "账户被冻结的原因",
            "content": "账户可能因以下原因被冻结：1）连续输入错误密码；2）异常登录行为；3）违反使用条款。请联系客服进行账户解冻。",
            "tags": ["冻结", "安全", "解冻"],
            "priority": "high",
            "last_updated": "2024-01-08"
        }
    ],
    "产品咨询": [
        {
            "id": "product_001",
            "title": "产品功能介绍",
            "content": "我们的产品提供以下核心功能：1）智能数据分析；2）实时监控报警；3）自定义报表生成；4）多平台同步。支持试用版本。",
            "tags": ["功能", "特性", "试用"],
            "priority": "medium",
            "last_updated": "2024-01-14"
        },
        {
            "id": "product_002",
            "title": "价格套餐说明",
            "content": "我们提供三种套餐：基础版（免费）、专业版（99元/月）、企业版（299元/月）。企业版包含更多高级功能和优先技术支持。",
            "tags": ["价格", "套餐", "付费"],
            "priority": "medium",
            "last_updated": "2024-01-11"
        }
    ],
    "投诉建议": [
        {
            "id": "complaint_001",
            "title": "投诉处理流程",
            "content": "我们重视每一个客户的反馈。投诉处理流程：1）记录问题详情；2）48小时内专人跟进；3）提供解决方案；4）回访确认满意度。",
            "tags": ["投诉", "流程", "反馈"],
            "priority": "high",
            "last_updated": "2024-01-13"
        }
    ]
}

搜索知识库

搜索知识库 - 智能知识检索算法，企业级知识检索系统设计：

  1. 多策略搜索引擎：

    • 关键词搜索：基于TF-IDF和布尔匹配的传统搜索
    • 语义搜索：使用词向量和句向量的语义理解
    • 模糊搜索：支持拼写错误和同义词的智能匹配
    • 全文搜索：支持复杂查询语法和高级过滤

  2. 相关性排序算法：

    • 文本匹配度：基于关键词重叠和权重的评分
    • 语义相似度：使用预训练模型计算语义距离
    • 业务优先级：结合文档重要性和时效性
    • 用户反馈：基于历史点击和满意度数据

  3. 查询优化策略：

    • 查询重写：自动扩展和优化用户查询
    • 分词处理：中文分词和词性标注
    • 同义词扩展：使用同义词词典增强召回
    • 个性化：基于用户历史和偏好调整结果

实际应用中的高级特性：

  1. 智能推荐：

    • 相关问题推荐：基于当前问题推荐类似问题
    • 热门问题：统计和推荐高频问题
    • 个性化推荐：基于用户画像的定制推荐
    • 实时趋势：基于当前热点的动态推荐

  2. 知识质量管理：

    • 内容审核：自动检测和标记低质量内容
    • 更新提醒：基于时效性的内容更新提醒
    • 用户反馈：收集用户对知识质量的评价
    • A/B测试：测试不同版本内容的效果

  3. 性能优化：

    • 索引优化：构建高效的搜索索引
    • 缓存策略：缓存热门查询的结果
    • 并行搜索：支持多数据源的并行查询
    • 增量更新：支持知识库的增量更新

参数说明：

  • query: 用户的搜索查询，可能包含自然语言问题
  • category: 问题分类，用于缩小搜索范围提高精度

返回值：
返回按相关性排序的知识条目列表，每个条目包含：

  • 原始知识内容和元数据
  • 相关性评分和匹配信息
  • 推荐理由和使用建议
def search_knowledge_base(query: str, category: str = None) -> List[Dict]:
    results = []
    
    # 第一步：基于分类过滤知识范围
    # 这种分层搜索策略大大提高了搜索的精确度和效率
    if category and category != "其他":
        if category in KNOWLEDGE_BASE:
            # 精确匹配分类 - 最理想的情况
            knowledge_items = KNOWLEDGE_BASE[category]
        else:
            # 模糊匹配分类名称 - 处理分类识别不准确的情况
            # 这种容错机制确保了系统的鲁棒性
            knowledge_items = []
            for kb_category, items in KNOWLEDGE_BASE.items():
                if category in kb_category or kb_category in category:
                    knowledge_items.extend(items)
            # 如果还是没有找到，搜索所有分类
            if not knowledge_items:
                for items in KNOWLEDGE_BASE.values():
                    knowledge_items.extend(items)
    else:
        # 搜索所有分类 - 兜底策略，确保总是有搜索结果
        knowledge_items = []
        for items in KNOWLEDGE_BASE.values():
            knowledge_items.extend(items)
    
    print(f"   搜索范围：{len(knowledge_items)} 个知识条目")
    
    # 第二步：改进的关键词匹配算法
    query_words = query.lower().split()
    
    # 同义词映射 - 提高搜索召回率
    # 在实际应用中，这通常来自：
    # - 专业的同义词词典（如HowNet、WordNet）
    # - 词向量模型的相似词计算
    # - 用户行为数据的挖掘结果
    # - 领域专家的知识整理
    synonym_map = {
        "冻结": ["冻结", "锁定", "被锁", "封"],
        "账户": ["账户", "账号", "帐户", "帐号"],
        "登录": ["登录", "登陆", "登入", "进入"],
        "密码": ["密码", "口令", "秘码"],
        "功能": ["功能", "特性", "能力"],
        "价格": ["价格", "费用", "收费", "钱"],
        "安装": ["安装", "下载", "部署"],
        "网络": ["网络", "网", "连接"],
        "投诉": ["投诉", "抱怨", "反馈"]
    }
    
    # 扩展查询词 - 使用同义词提高匹配率
    expanded_words = set(query_words)
    for word in query_words:
        for key, synonyms in synonym_map.items():
            if word in synonyms:
                expanded_words.update(synonyms)
    
    # 第三步：相关性评分算法
    # 这是知识检索的核心算法，决定了搜索结果的质量
    for item in knowledge_items:
        score = 0
        content_lower = (item["title"] + " " + item["content"]).lower()
        tags_lower = [tag.lower() for tag in item["tags"]]
        
        # 计算匹配分数 - 多因素评分模型
        for word in expanded_words:
            # 内容匹配（权重：1）
            if word in content_lower:
                score += 1
            
            # 标签匹配（权重：3）- 标签通常更精确和重要
            for tag in tags_lower:
                if word in tag or tag in word:
                    score += 3
        
        # 添加优先级加权 - 重要知识条目优先显示
        # 这体现了业务优先级在搜索排序中的重要性
        if item.get("priority") == "high":
            score += 1
        elif item.get("priority") == "medium":
            score += 0.5
        
        # 只保留有匹配的结果
        if score > 0:
            item_copy = item.copy()
            item_copy["relevance_score"] = score
            results.append(item_copy)
    
    print(f"   匹配到 {len(results)} 个相关结果")
    
    # 第四步：结果排序和截取
    # 按分数排序，返回最相关的结果
    results.sort(key=lambda x: x["relevance_score"], reverse=True)
    return results[:3]  # 返回最相关的3个结果，避免信息过载

节点函数定义

问题分类节点 - 使用LLM进行智能分类，这个节点展示了如何：

  1. 设计结构化的提示工程
  2. 处理LLM的JSON输出
  3. 实现错误处理和降级策略
  4. 记录处理过程和调试信息

提示工程最佳实践：

  • 明确的任务描述
  • 具体的分类标准
  • 结构化的输出格式
  • 示例和约束条件

错误处理策略：

  • JSON解析失败的降级
  • LLM调用异常的处理
  • 默认值的合理设置
  • 详细的错误记录
def question_classification_node(state: CustomerServiceState) -> CustomerServiceState:
    print("🔍 正在分析和分类用户问题...")
    
    # 初始化LLM - 支持多种提供商
    llm = initialize_llm()
    user_question = state["user_question"]
    
    # 构建分类提示词 - 结构化提示工程的典型例子
    classification_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个专业的客服问题分类专家。请分析用户问题并进行分类。
			
			分类选项：
			1. 技术支持 - 软件使用、故障排查、安装问题等
			2. 账户问题 - 登录、密码、账户状态等  
			3. 产品咨询 - 功能介绍、价格、购买等
			4. 投诉建议 - 服务投诉、意见建议等
			5. 其他 - 不属于以上分类的问题
			
			请严格按照以下JSON格式返回结果，不要添加任何额外的文字或格式：
			{{"category": "分类名称", "confidence": 0.85, "reason": "分类理由"}}"""),
			        ("user", "用户问题：{question}")
			    ])
    
    try:
        print(f"   正在调用LLM进行分类...")
        formatted_messages = classification_prompt.format_messages(question=user_question)
        print(f"   发送给LLM的消息: {formatted_messages}")
        response = llm.invoke(formatted_messages)
        print(f"   LLM调用成功，响应类型: {type(response)}")
        
        # 处理LLM响应 - 清理和解析JSON输出
        content = response.content.strip()
        print(f"   LLM原始响应：{content}")
        
        # 清理响应格式 - 处理可能的Markdown格式包装
        if content.startswith('```json'):
            content = content.replace('```json', '').replace('```', '').strip()
        elif content.startswith('```'):
            content = content.replace('```', '').strip()
        
        print(f"   清理后的内容：{repr(content)}")
        
        # 多行内容处理 - 提取JSON部分
        lines = content.split('\n')
        json_line = None
        for line in lines:
            line = line.strip()
            if line.startswith('{') and line.endswith('}'):
                json_line = line
                break
        
        if json_line:
            content = json_line
        
        # JSON解析和数据提取
        try:
            print(f"   准备解析的JSON内容：{repr(content)}")
            result = json.loads(content)
            category = result.get("category", "其他")
            confidence = result.get("confidence", 0.5)
            reason = result.get("reason", "")
            print(f"   解析成功：{result}")
        except json.JSONDecodeError as je:
            # JSON解析失败的降级处理
            print(f"   JSON解析失败：{je}")
            print(f"   尝试解析的内容：{repr(content)}")
            print(f"   内容长度：{len(content)}")
            print(f"   内容类型：{type(content)}")
            # 使用默认值确保系统继续运行
            category = "其他"
            confidence = 0.3
            reason = f"JSON解析失败：{str(je)}"
        
    except Exception as e:
        # 全面的异常处理
        print(f"分类失败：{e}")
        print(f"异常类型：{type(e)}")
        import traceback
        print(f"详细错误信息：")
        traceback.print_exc()
        # 降级到安全的默认值
        category = "其他"
        confidence = 0.1
        reason = f"分类过程出错：{str(e)}"
    
    # 更新状态 - 记录分类结果和过程
    state["question_category"] = category
    state["category_confidence"] = confidence
    state["processing_steps"].append(f"问题分类：{category} (置信度: {confidence:.2f})")
    state["messages"].append(SystemMessage(content=f"问题已分类为：{category}，理由：{reason}"))
    
    # 输出处理结果
    print(f"   分类结果：{category}")
    print(f"   置信度：{confidence:.2f}")
    print(f"   理由：{reason}")
    
    return state

情感分析节点

情感分析节点 - 识别用户情绪和紧急程度，这个节点的重要性：

  1. 客服质量：理解用户情绪，提供合适的服务语调
  2. 优先级排序：紧急情况优先处理
  3. 升级决策：负面情绪可能需要人工介入
  4. 服务改进：情感数据可用于服务质量分析

情感分析的应用：

  • 自动调整回复语调
  • 识别服务升级需求
  • 客户满意度监控
  • 风险预警和干预

技术实现要点：

  • 多维度情感识别
  • 紧急程度量化
  • 关键词提取
  • 结构化输出
def sentiment_analysis_node(state: CustomerServiceState) -> CustomerServiceState:
    print("😊 正在进行情感分析...")
    
    llm = initialize_llm()
    user_question = state["user_question"]
    
    # 情感分析提示词 - 多维度情感识别
    sentiment_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个情感分析专家。请分析用户消息的情感倾向和紧急程度。

		情感类别：
		- positive: 积极正面
		- neutral: 中性平和  
		- negative: 消极负面
		- frustrated: 沮丧不满
		- angry: 愤怒生气
		
		紧急程度：
		- low: 一般咨询，不紧急
		- medium: 需要及时回复
		- high: 紧急问题，需要立即处理
		
		请严格按照以下JSON格式返回结果，不要添加任何额外的文字或格式：
		{{"sentiment": "neutral", "score": 0.5, "urgency": "medium", "emotion": "具体情绪描述", "keywords": ["关键词1", "关键词2"]}}"""),
		        ("user", "用户消息：{message}")
		    ])
    
    try:
        print(f"   正在调用LLM进行情感分析...")
        formatted_messages = sentiment_prompt.format_messages(message=user_question)
        print(f"   发送给LLM的消息: {formatted_messages}")
        response = llm.invoke(formatted_messages)
        print(f"   LLM调用成功，响应类型: {type(response)}")
        
        # 响应处理流程（与分类节点类似）
        content = response.content.strip()
        print(f"   LLM原始响应：{content}")
        
        # 清理响应格式
        if content.startswith('```json'):
            content = content.replace('```json', '').replace('```', '').strip()
        elif content.startswith('```'):
            content = content.replace('```', '').strip()
        
        print(f"   清理后的内容：{repr(content)}")
        
        # 提取JSON部分
        lines = content.split('\n')
        json_line = None
        for line in lines:
            line = line.strip()
            if line.startswith('{') and line.endswith('}'):
                json_line = line
                break
        
        if json_line:
            content = json_line
        
        try:
            print(f"   准备解析的JSON内容：{repr(content)}")
            sentiment_result = json.loads(content)
            print(f"   解析成功：{sentiment_result}")
        except json.JSONDecodeError as je:
            # 情感分析失败的降级处理
            print(f"   JSON解析失败：{je}")
            print(f"   尝试解析的内容：{repr(content)}")
            print(f"   内容长度：{len(content)}")
            print(f"   内容类型：{type(content)}")
            # 使用中性的默认值
            sentiment_result = {
                "sentiment": "neutral",
                "score": 0.5,
                "urgency": "medium",
                "emotion": "平和",
                "keywords": []
            }
    except Exception as e:
        # 全面的异常处理
        print(f"情感分析失败：{e}")
        print(f"异常类型：{type(e)}")
        import traceback
        print(f"详细错误信息：")
        traceback.print_exc()
        # 降级到安全的默认值
        sentiment_result = {
            "sentiment": "neutral", 
            "score": 0.5,
            "urgency": "medium",
            "emotion": "未知",
            "keywords": []
        }
    
    # 更新状态 - 保存情感分析结果
    state["sentiment_analysis"] = sentiment_result
    state["urgency_level"] = sentiment_result.get("urgency", "medium")
    state["processing_steps"].append(f"情感分析：{sentiment_result.get('sentiment')} (强度: {sentiment_result.get('score', 0):.2f})")
    
    # 输出分析结果
    print(f"   情感：{sentiment_result.get('sentiment')}")
    print(f"   强度：{sentiment_result.get('score', 0):.2f}")
    print(f"   紧急程度：{sentiment_result.get('urgency')}")
    
    return state

知识库搜索节点

知识库搜索节点 - 智能知识检索，这个节点的作用：

  1. 基于问题分类和内容搜索相关知识
  2. 为后续的回答生成提供信息支持
  3. 计算知识匹配的置信度
  4. 整理和格式化搜索结果

知识检索策略：

  • 分类优先：先在对应分类中搜索
  • 关键词匹配：使用分词和同义词
  • 相关性排序：按匹配度排序结果
  • 结果限制：避免信息过载

实际应用扩展：

  • 向量语义搜索
  • 机器学习排序
  • 个性化推荐
  • 实时更新索引

def knowledge_search_node(state: CustomerServiceState) -> CustomerServiceState:
    print("📚 正在搜索知识库...")
    
    user_question = state["user_question"]
    category = state["question_category"]
    
    # 调用知识库搜索函数
    search_results = search_knowledge_base(user_question, category)
    
    # 整理相关信息 - 格式化搜索结果用于后续处理
    relevant_info = ""
    if search_results:
        relevant_info = "找到以下相关信息：\n"
        for i, result in enumerate(search_results, 1):
            # 格式化每个搜索结果
            relevant_info += f"{i}. {result['title']}\n{result['content']}\n\n"
    else:
        relevant_info = "未找到直接相关的知识库信息。"
    
    # 更新状态 - 保存搜索结果和格式化信息
    state["knowledge_results"] = search_results
    state["relevant_info"] = relevant_info
    state["processing_steps"].append(f"知识库搜索：找到 {len(search_results)} 条相关结果")
    
    # 输出搜索统计
    print(f"   搜索结果：{len(search_results)} 条")
    for result in search_results:
        print(f"   - {result['title']} (相关度: {result['relevance_score']})")
    
    return state

智能回答生成节点

智能回答生成节点 - 整合上下文生成个性化回答，这个节点是整个客服系统的核心输出环节：

  1. 整合所有前置分析结果
  2. 生成个性化的客服回答
  3. 根据情感调整语调
  4. 提供后续操作建议

回答生成策略：

  • 上下文感知：使用所有可用信息
  • 情感适应：根据用户情绪调整语调
  • 知识融合：优先使用知识库信息
  • 降级处理：确保总是有合理回答

质量保证：

  • 专业性：使用客服标准用语
  • 完整性：提供充分的信息
  • 可操作性：给出具体的解决步骤
  • 友好性：保持良好的服务态度
def response_generation_node(state: CustomerServiceState) -> CustomerServiceState:
    print("💬 正在生成智能回答...")
    
    llm = initialize_llm()
    
    # 收集所有上下文信息 - 整合前置节点的分析结果
    user_question = state["user_question"]
    category = state["question_category"]
    sentiment = state["sentiment_analysis"]
    knowledge_info = state["relevant_info"]
    urgency = state["urgency_level"]
    
    # 构建回答生成提示词 - 复杂的上下文整合
    response_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个专业的客服助手。请根据以下信息为用户生成有帮助的回答：

		回答要求：
		1. 语气要友好、专业、耐心
		2. 根据用户情感调整语调（如果用户生气要更加耐心）
		3. 优先使用知识库信息
		4. 如果知识库没有答案，提供通用建议
		5. 回答要具体、可执行
		6. 适当表达歉意和理解
		
		用户信息：
		- 问题分类：{category}
		- 情感状态：{sentiment}
		- 紧急程度：{urgency}
		
		知识库信息：
		{knowledge_info}
		
		请生成一个专业、有帮助的客服回答。"""),
		        ("user", "用户问题：{question}")
		    ])
    
    try:
        # 调用LLM生成回答
        response = llm.invoke(response_prompt.format_messages(
            category=category,
            sentiment=sentiment.get("sentiment", "neutral"),
            urgency=urgency,
            knowledge_info=knowledge_info,
            question=user_question
        ))
        
        print(f"   LLM原始响应：{response.content}")
        ai_response = response.content
        
    except Exception as e:
        print(f"回答生成失败：{e}")
        # 降级到通用回答模板
        ai_response = f"感谢您的咨询。关于您提到的{category}问题，我已经记录了您的需求。我们会尽快为您提供解决方案。如果问题紧急，建议您联系我们的人工客服。"
    
    # 生成建议操作 - 基于分析结果的智能建议
    suggested_actions = []
    
    # 基于紧急程度的建议
    if urgency == "high":
        suggested_actions.append("转接人工客服")
    
    # 基于情感状态的建议
    if sentiment.get("sentiment") in ["negative", "frustrated", "angry"]:
        suggested_actions.append("情绪安抚")
        suggested_actions.append("主管介入")
    
    # 基于知识匹配的建议
    if state["knowledge_results"]:
        suggested_actions.append("提供相关文档链接")
    
    # 更新状态 - 保存生成的回答和建议
    state["ai_response"] = ai_response
    state["suggested_actions"] = suggested_actions
    state["processing_steps"].append("生成智能回答")
    state["messages"].append(AIMessage(content=ai_response))
    
    # 输出生成统计
    print(f"   回答长度：{len(ai_response)} 字符")
    print(f"   建议操作：{suggested_actions}")
    
    return state

升级决策节点

升级决策节点 - 多维度评估是否需要人工介入，这个节点实现了智能客服的核心决策逻辑：

  1. 综合评估多个维度的指标
  2. 基于业务规则做出升级决策
  3. 记录决策过程和原因
  4. 为后续路由提供判断依据

决策维度：

  • 情感因素：负面情绪需要人工关怀
  • 紧急程度：高优先级问题需要快速响应
  • 分类置信度：不确定的问题需要专家判断
  • 知识匹配度：无法自动解决的问题
  • 特殊关键词：涉及法律、投诉等敏感问题

业务价值：

  • 提高客户满意度
  • 优化人力资源配置
  • 减少处理时间
  • 降低服务成本
def escalation_decision_node(state: CustomerServiceState) -> CustomerServiceState:
    print("⚖️ 正在判断是否需要人工介入...")
    
    # 初始化决策变量
    need_human = False
    reasons = []
    
    # 维度1：情感因素评估
    sentiment = state["sentiment_analysis"]
    if sentiment.get("sentiment") in ["angry", "frustrated"]:
        need_human = True
        reasons.append("用户情绪负面")
    
    # 维度2：紧急程度评估
    if state["urgency_level"] == "high":
        need_human = True
        reasons.append("问题紧急")
    
    # 维度3：分类置信度评估
    if state["category_confidence"] < 0.6:
        need_human = True  
        reasons.append("问题分类不明确")
    
    # 维度4：知识库匹配度评估
    if not state["knowledge_results"]:
        need_human = True
        reasons.append("知识库无相关信息")
    
    # 维度5：特殊关键词检测
    # 某些敏感问题必须由人工处理
    complaint_keywords = ["投诉", "退款", "赔偿", "法律", "起诉"]
    if any(keyword in state["user_question"] for keyword in complaint_keywords):
        need_human = True
        reasons.append("涉及投诉或法律问题")
    
    # 更新状态 - 记录决策结果和原因
    state["need_human_handover"] = need_human
    state["processing_steps"].append(f"升级判断：{'需要' if need_human else '不需要'}人工介入")
    
    # 如果需要升级，添加详细说明
    if need_human:
        escalation_msg = f"建议转接人工客服。原因：{'; '.join(reasons)}"
        state["messages"].append(SystemMessage(content=escalation_msg))
    
    # 输出决策结果
    print(f"   需要人工介入：{need_human}")
    if reasons:
        print(f"   原因：{'; '.join(reasons)}")
    
    return state

条件判断函数

路由决策函数 - 决定工作流的最终路径，这是LangGraph条件边的实现，用于：

  1. 检查升级决策的结果
  2. 选择相应的处理路径
  3. 确保工作流的正确执行

返回值说明：

  • “human_handover”: 转接人工客服
  • “ai_response”: 使用AI回答结束
def should_escalate(state: CustomerServiceState) -> Literal["human_handover", "ai_response"]:
    if state.get("need_human_handover", False):
        return "human_handover"
    else:
        return "ai_response"

人工接管节点

人工接管节点 - 生成详细的交接信息，这个节点的职责：

  1. 整理完整的上下文信息
  2. 生成结构化的交接文档
  3. 提供处理建议和历史记录
  4. 确保人工客服能够快速上手

交接信息包含：

  • 用户基本信息
  • 问题分析结果
  • 处理历史记录
  • AI建议和推荐操作
  • 注意事项和风险提示

业务价值：

  • 提高人工客服效率
  • 减少重复信息收集
  • 保证服务连续性
  • 提升客户体验
def human_handover_node(state: CustomerServiceState) -> CustomerServiceState:
    print("👥 转接人工客服...")
    
    # 生成详细的交接信息文档
    handover_info = f"""
		=== 客服交接信息 ===
		用户ID: {state['user_id']}
		问题分类: {state['question_category']} (置信度: {state['category_confidence']:.2f})
		情感状态: {state['sentiment_analysis'].get('sentiment', 'unknown')}
		紧急程度: {state['urgency_level']}
		用户问题: {state['user_question']}
		
		处理步骤: {' -> '.join(state['processing_steps'])}
		
		知识库搜索结果: {len(state['knowledge_results'])} 条
		AI建议回答: {state.get('ai_response', '未生成')}
		
		建议操作: {', '.join(state['suggested_actions'])}
		"""
    
    # 生成最终的客户回复
    final_response = "您的问题已转接给人工客服，我们的专业客服人员会尽快为您处理。感谢您的耐心等待。"
    
    # 更新状态
    state["ai_response"] = final_response
    state["messages"].append(AIMessage(content=final_response))
    state["messages"].append(SystemMessage(content=handover_info))
    state["processing_steps"].append("转接人工客服")
    
    print("   已生成交接信息")
    return state

工作流构建

创建客服助手工作流图 - 复杂业务流程的编排，这个函数展示了企业级应用的工作流设计：

  1. 多节点的顺序处理
  2. 复杂的条件路由
  3. 状态的完整传递
  4. 错误处理和降级

工作流特点：

  • 线性处理：问题分析 -> 情感分析 -> 知识搜索 -> 回答生成 -> 升级决策
  • 条件分支：最后基于升级决策选择AI回答或人工接管
  • 状态驱动：每个节点都基于完整的状态信息做决策
  • 可扩展性：易于添加新的处理节点

业务流程设计原则：

  • 用户体验优先
  • 处理效率最大化
  • 错误容忍和恢复
  • 数据完整性保证
def create_customer_service_graph():
    print("🏗️ 正在构建智能客服工作流...")
    
    # 创建状态图 - 定义工作流的状态管理
    workflow = StateGraph(CustomerServiceState)
    
    # 添加处理节点 - 每个节点负责特定的业务逻辑
    workflow.add_node("classify_question", question_classification_node)    # 问题分类
    workflow.add_node("analyze_sentiment", sentiment_analysis_node)        # 情感分析
    workflow.add_node("search_knowledge", knowledge_search_node)           # 知识搜索
    workflow.add_node("generate_response", response_generation_node)       # 回答生成
    workflow.add_node("decide_escalation", escalation_decision_node)       # 升级决策
    workflow.add_node("human_handover", human_handover_node)              # 人工接管
    
    # 设置工作流入口点
    workflow.set_entry_point("classify_question")
    
    # 添加线性处理边 - 定义主要的处理流程
    workflow.add_edge("classify_question", "analyze_sentiment")    # 分类 -> 情感分析
    workflow.add_edge("analyze_sentiment", "search_knowledge")     # 情感分析 -> 知识搜索
    workflow.add_edge("search_knowledge", "generate_response")     # 知识搜索 -> 回答生成
    workflow.add_edge("generate_response", "decide_escalation")    # 回答生成 -> 升级决策
    
    # 添加条件分支边 - 基于升级决策选择最终路径
    workflow.add_conditional_edges(
        "decide_escalation",    # 决策节点
        should_escalate,        # 条件函数
        {
            "human_handover": "human_handover",  # 需要人工 -> 人工接管节点
            "ai_response": END                   # 不需要人工 -> 直接结束
        }
    )
    
    # 人工接管后结束流程
    workflow.add_edge("human_handover", END)
    
    # 编译工作流图 - 生成可执行的应用
    app = workflow.compile()
    
    print("✅ 智能客服工作流构建完成！")
    return app

主函数和演示

运行智能客服助手 - 完整的客服处理流程， 这是整个客服系统的主要接口：

  1. 初始化工作流应用
  2. 创建初始状态
  3. 执行完整的处理流程
  4. 返回详细的处理结果

参数：

  • user_question: 用户问题
  • user_id: 用户标识

返回：

  • 完整的处理状态，包含所有分析结果和处理记录

使用场景：

  • 在线客服系统
  • 智能问答机器人
  • 客户支持平台
  • 服务质量评估
def run_customer_service(user_question: str, user_id: str = "user_001") -> CustomerServiceState:
    print(f"\n{'='*60}")
    print(f"🤖 智能客服助手开始服务")
    print(f"{'='*60}")
    
    # 创建工作流应用
    app = create_customer_service_graph()
    
    # 生成会话标识
    session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    
    # 初始化完整的状态对象
    initial_state = CustomerServiceState(
        # 基本信息
        user_id=user_id,
        session_id=session_id,
        timestamp=datetime.now().isoformat(),
        
        # 用户输入
        user_question=user_question,
        messages=[HumanMessage(content=user_question)],
        
        # 分析结果（初始为空）
        question_category="",
        category_confidence=0.0,
        sentiment_analysis={},
        
        # 知识库结果（初始为空）
        knowledge_results=[],
        relevant_info="",
        
        # 决策结果（初始为默认值）
        need_human_handover=False,
        urgency_level="medium",
        
        # 输出结果（初始为空）
        ai_response="",
        suggested_actions=[],
        
        # 元数据（初始为空）
        processing_steps=[],
        confidence_scores={}
    )
    
    print(f"\n📥 收到用户咨询：{user_question}")
    print(f"👤 用户ID：{user_id}")
    print(f"🆔 会话ID：{session_id}")
    
    # 执行完整的工作流
    result = app.invoke(initial_state)
    
    # 输出处理结果
    print(f"\n{'='*60}")
    print(f"📤 客服回复：")
    print(f"{'='*60}")
    print(result["ai_response"])
    
    # 输出详细的处理统计
    print(f"\n📊 处理统计：")
    print(f"   - 问题分类：{result['question_category']} (置信度: {result['category_confidence']:.2f})")
    print(f"   - 情感状态：{result['sentiment_analysis'].get('sentiment', 'unknown')}")
    print(f"   - 紧急程度：{result['urgency_level']}")
    print(f"   - 知识库结果：{len(result['knowledge_results'])} 条")
    print(f"   - 需要人工：{'是' if result['need_human_handover'] else '否'}")
    print(f"   - 建议操作：{', '.join(result['suggested_actions'])}")
    print(f"   - 处理步骤：{' -> '.join(result['processing_steps'])}")
    
    return result

演示智能客服系统

演示智能客服系统 - 全面测试不同场景，这个演示函数设计了多种测试场景：

  1. 不同类型的问题（技术、账户、产品、投诉）
  2. 不同的情感状态（平和、紧急、负面、愤怒）
  3. 不同的复杂程度（简单咨询、复杂问题）

测试目标：

  • 验证问题分类的准确性
  • 测试情感识别的效果
  • 检查知识库搜索的相关性
  • 评估升级决策的合理性
  • 确保回答质量的一致性

业务价值：

  • 系统功能验证
  • 性能基准测试
  • 用户体验评估
  • 改进方向识别
def demo_customer_service():
    print("🚀 智能客服助手演示开始")
    print("=" * 80)
    
    # 精心设计的测试用例 - 覆盖不同业务场景
    test_cases = [
        {
            "question": "我的账户被冻结了，急需解决！",
            "user_id": "user_001",
            "description": "账户问题 + 紧急情况 + 负面情绪"
        },
        {
            "question": "你们的产品有哪些功能？价格如何？",
            "user_id": "user_002", 
            "description": "产品咨询 + 平和语气 + 信息需求"
        },
        {
            "question": "软件老是闪退，安装了好几次都不行，太垃圾了！",
            "user_id": "user_003",
            "description": "技术支持 + 负面情绪 + 重复问题"
        },
        {
            "question": "我要投诉你们的服务态度，要求退款赔偿！",
            "user_id": "user_004",
            "description": "投诉问题 + 愤怒情绪 + 法律相关"
        },
        {
            "question": "请问如何重置密码？",
            "user_id": "user_005",
            "description": "简单咨询 + 礼貌语气 + 标准流程"
        }
    ]
    
    # 执行所有测试用例
    for i, test_case in enumerate(test_cases, 1):
        print(f"\n🔹 测试案例 {i}: {test_case['description']}")
        print("-" * 50)
        
        # 运行客服处理流程
        result = run_customer_service(
            test_case["question"],
            test_case["user_id"]
        )
        
        # 添加分隔线
        if i < len(test_cases):
            print("\n" + "=" * 80)

def show_architecture_info():
    """
    显示系统架构信息 - 完整的技术文档
    
    这个函数提供了系统的全面介绍：
    1. 架构设计的总体思路
    2. 各个组件的职责分工
    3. 数据流和控制流
    4. 技术栈和实现方案
    5. 扩展方向和改进建议
    
    文档价值：
    - 新人快速理解系统
    - 技术选型参考
    - 系统改进指导
    - 最佳实践分享
    """
    print("""
🏛️ 智能客服助手系统架构：

┌─────────────────────────────────────────────────────────────┐
│                      LangGraph 工作流                        │
├─────────────────────────────────────────────────────────────┤
│  用户输入                                                    │
│      ↓                                                      │
│  1. 问题分类 (LLM)     ← 技术支持/账户/产品/投诉/其他        │
│      ↓                                                      │
│  2. 情感分析 (LLM)     ← 积极/中性/消极/沮丧/愤怒            │
│      ↓                                                      │
│  3. 知识库搜索         ← 关键词匹配 + 分类过滤               │
│      ↓                                                      │
│  4. 智能回答生成 (LLM) ← 上下文 + 知识库信息                 │
│      ↓                                                      │
│  5. 升级决策           ← 情感/紧急度/置信度判断              │
│      ↓                                                      │
│  6. [条件分支]                                              │
│     ├─ AI回答 → 结束                                        │
│     └─ 转人工 → 生成交接信息                                │
└─────────────────────────────────────────────────────────────┘

🎯 核心特性：
• 真实LLM集成 (OpenAI/通义千问/文心一言等)
• 多维度问题分析 (分类+情感+知识匹配)
• 智能升级决策 (基于多因素判断)
• 状态全程追踪 (每一步都有记录)
• 灵活配置支持 (不同LLM提供商)

🔧 技术栈：
• LangGraph: 工作流编排和状态管理
• LangChain: LLM集成和提示工程
• TypedDict: 状态类型安全和结构化
• JSON解析: 结构化LLM输出处理
• 知识库: 基于关键词的智能搜索匹配

💡 扩展可能：
• 集成向量数据库 (更好的语义搜索)
• 添加语音识别/合成 (多模态交互)
• 集成CRM系统 (客户信息管理)
• 多轮对话上下文 (对话历史记忆)
• 实时情感监控 (情绪变化追踪)
• A/B测试框架 (效果评估优化)
• 机器学习排序 (个性化推荐)
• 知识图谱集成 (关系推理)
""")

主函数

主函数 - 系统启动和演示入口，这个主函数展示了：

  1. 系统初始化和检查
  2. 配置验证和提示
  3. 演示流程的执行
  4. 使用指导和建议

运行模式：

  • 开发模式：显示详细调试信息
  • 演示模式：运行预设测试用例
  • 生产模式：处理真实用户请求

最佳实践展示：

  • 环境检查和降级策略
  • 错误处理和用户提示
  • 详细的使用说明
  • 性能和成本优化建议
if __name__ == "__main__":
    # 显示系统架构信息
    show_architecture_info()
    
    # 检查LLM可用性并提供配置指导
    print(f"🔧 LLM状态：{LLM_AVAILABLE or '使用模拟模式'}")
    if not LLM_AVAILABLE:
        print("⚠️  未检测到可用的LLM库，将使用模拟模式演示")
        print("💡 要使用真实LLM，请安装并配置：")
        print("   pip install langchain-openai  # OpenAI")
        print("   pip install langchain-community  # 其他提供商")
        print("   export OPENAI_API_KEY=your_key_here")
    
    print("\n" + "="*80)
    
    # 运行完整的演示流程
    demo_customer_service()
    
    print("\n🎉 智能客服助手演示完成！")
    print("\n💡 使用建议：")
    print("1. 配置真实的LLM API密钥以获得更好效果")
    print("2. 可以修改 test_cases 来测试更多场景")
    print("3. 可以扩展知识库内容和搜索算法")
    print("4. 可以调整升级判断的条件和阈值")
    print("5. 可以集成向量数据库提升搜索效果")
    print("6. 可以添加用户反馈机制改进服务质量")
    print("7. 可以实现多轮对话和上下文记忆")
    print("8. 可以集成实时监控和性能分析")
# 人工智能
Logo
智能体开发者社区

中国智能体开发者社区，聚焦智能体与大模型开发，提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动，促进经验交流与协作，助力开发者快速构建创新智能应用。

更多推荐

cover

王仕宇是如何做 AI Agent 的：从模型调用，到真正能干活的智能体

avatar 智能体开发者社区
cover

[特殊字符]DeepSeek‑Harness（DSH）小白保姆教程

avatar 智能体开发者社区

Deepseek Agent Harness教程(一) | 什么是agent

这篇文章揭示了AI Agent与大语言模型（LLM）的本质区别，并详细解释了Agent的工作原理。核心观点包括： 大语言模型本质是无状态的文本处理函数，所有"记忆"都是通过客户端重传历史对话实现的，这导致对话成本呈平方级增长。 AI Agent是"大脑+手脚+循环"的系统，由三部分组成： 大模型作为决策核心 工具调用作为执行手段 循环机制实现持续交互 Agent的实际工作流程： ① 客户端维护对话

avatar 智能体开发者社区