项目概述

构建一个小型多轮对话智能客服，支持工具调用以及模型与插件的热更新。

实现思路指

阶段一：基础对话系统搭建

使用 LangChain 构建基础 Chain：Prompt → LLM → OutputParser
用户说“我昨天下的单”，系统能结合当前时间推断“昨天”的具体日期
具体代码：

import os
from datetime import datetime, timedelta
from typing import Dict, Any, List
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.schema import BaseOutputParser
from langchain_openai import ChatOpenAI  # 使用新的导入方式

class SimpleConversationMemory:
    """简化的对话记忆实现"""

    def __init__(self, max_turns: int = 10):
        self.messages = []
        self.max_turns = max_turns

    def add_message(self, role: str, content: str):
        """添加消息"""
        self.messages.append({"role": role, "content": content})

        # 保持对话历史不超过最大轮数
        if len(self.messages) > self.max_turns * 2:
            self.messages = self.messages[-(self.max_turns * 2):]

    def get_recent_history(self) -> str:
        """获取最近的对话历史"""
        history_text = ""
        for msg in self.messages:
            role = "用户" if msg["role"] == "user" else "客服"
            history_text += f"{role}: {msg['content']}\n"
        return history_text

    def clear(self):
        """清空历史"""
        self.messages.clear()

class TimeAwareOutputParser(BaseOutputParser):
    """自定义输出解析器"""

    def parse(self, text: str) -> Dict[str, Any]:
        return {
            "response": text.strip(),
            "timestamp": datetime.now().isoformat(),
            "contains_time_reference": any(keyword in text.lower() for keyword in
                                           ['昨天', '今天', '明天', '日期', '时间'])
        }

class TimeAwareCustomerService:
    """时间感知的客服系统"""

    def __init__(self, model_name: str = "gpt-3.5-turbo"):
        self.memory = SimpleConversationMemory()
        self.model_name = model_name
        self.chain = self._setup_chain()
        self.output_parser = TimeAwareOutputParser()

    def _get_current_time_context(self) -> str:
        """获取当前时间上下文"""
        now = datetime.now()
        current_time = now.strftime("%Y年%m月%d日 %H:%M:%S")

        # 计算相关日期
        yesterday = (now - timedelta(days=1)).strftime("%Y年%m月%d日")
        tomorrow = (now + timedelta(days=1)).strftime("%Y年%m月%d日")

        current_time_response = f"""
            当前时间: {current_time}
            相关日期:
            - 昨天: {yesterday}
            - 今天: {now.strftime('%Y年%m月%d日')}
            - 明天: {tomorrow}
            """
        print(current_time_response)
        return current_time_response

    def _setup_chain(self) -> LLMChain:
        """设置对话链"""

        prompt_template = PromptTemplate(
            input_variables=["time_context", "user_input", "chat_history"],
            template="""你是一个智能客服助手，需要准确理解用户的时间相关查询，并结合当前时间进行回答。
                
                {time_context}
                
                对话历史:
                {chat_history}
                
                用户输入: {user_input}
                
                请根据当前时间信息，准确理解用户提到的相对时间（如"昨天"、"今天"、"明天"等），并提供有帮助的回复。
                
                客服回复:"""
                        )

        # 使用新的导入方式
        llm = ChatOpenAI(
            model_name=self.model_name,
            temperature=0.3,
            openai_api_key=os.getenv("OPENAI_API_KEY")
        )

        return LLMChain(
            llm=llm,
            prompt=prompt_template
        )

    def process_message(self, user_input: str) -> Dict[str, Any]:
        """处理用户消息"""

        # 获取对话历史
        chat_history = self.memory.get_recent_history()

        # 准备输入
        inputs = {
            "time_context": self._get_current_time_context(),
            "user_input": user_input,
            "chat_history": chat_history
        }

        # 执行对话链
        result = self.chain.invoke(inputs)

        # 使用输出解析器处理结果
        parsed_result = self.output_parser.parse(result["text"])

        # 保存到记忆
        self.memory.add_message("user", user_input)
        self.memory.add_message("assistant", parsed_result["response"])

        return {
            "response": parsed_result["response"],
            "timestamp": parsed_result["timestamp"],
            "contains_time_reference": parsed_result["contains_time_reference"]
        }

    def clear_conversation(self):
        """清空对话历史"""
        self.memory.clear()

# 使用示例
def main():
    # 初始化客服系统
    customer_service = TimeAwareCustomerService()

    # 测试对话
    test_messages = [
        "我昨天下的订单，现在到哪里了？",
        "那明天能送到吗？",
        "好的，谢谢"
    ]

    print("=== 智能客服对话演示 ===\n")

    for message in test_messages:
        print(f"用户: {message}")
        response = customer_service.process_message(message)
        print(f"客服: {response['response']}")
        if response['contains_time_reference']:
            print(f"[时间推理已应用]")
        print()

if __name__ == "__main__":
    main()

运行结果：

客服: 根据您提供的信息，您昨天下的订单是2025年10月14日下的。请您提供订单号或者其他相关信息，我可以帮您查询订单的当前状态和位置。
[时间推理已应用]

用户: 那明天能送到吗？

            当前时间: 2025年10月15日 16:47:08
            相关日期:
            - 昨天: 2025年10月14日
            - 今天: 2025年10月15日
            - 明天: 2025年10月16日
            
客服: 根据当前时间是2025年10月15日，明天是2025年10月16日，如果您现在下单，通常情况下明天是可以送达的。但具体送达时间可能会受到物流运输和其他因素的影响，建议您在下单时查看具体的送达时间。如果有特殊要求或紧急情况，也可以选择加急配送服务。祝您购物愉快！如果您有其他问题，欢迎随时咨询。
[时间推理已应用]

用户: 好的，谢谢

            当前时间: 2025年10月15日 16:47:09
            相关日期:
            - 昨天: 2025年10月14日
            - 今天: 2025年10月15日
            - 明天: 2025年10月16日
            
客服: 您好，如果您有任何其他问题或需要帮助，请随时告诉我。祝您一切顺利！如果您有任何其他问题或需要帮助，请随时告诉我。祝您一切顺利！


Process finished with exit code 0

阶段二：多轮对话与工具调用

实现“订单查询”“退款申请”等多轮交互流程，支持工具自动调用。
使用 LangGraph 构建以下流程：

• 用户说“查订单” → 追问“请提供订单号”
• 收到订单号后 → 调用 query_order(order_id) 工具
• 返回订单状态与物流信息

实现的langgraph图

具体代码：

import os
from datetime import datetime, timedelta
from typing import Dict, Any, List, TypedDict
from enum import Enum
from dotenv import load_dotenv
import re

# 加载环境变量
load_dotenv()

from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END

# 定义状态类型
class CustomerServiceState(TypedDict):
    user_input: str
    chat_history: List[Dict[str, str]]
    current_intent: str
    missing_info: Dict[str, Any]
    tool_result: Any
    response: str
    conversation_finished: bool
    step_count: int
    max_steps: int
    information_collected: bool
    information_extracted: bool

# 定义意图枚举
class Intent(Enum):
    QUERY_ORDER = "query_order"
    APPLY_REFUND = "apply_refund"
    GENERAL_QUERY = "general_query"
    UNKNOWN = "unknown"

# 定义每个意图所需的信息
REQUIRED_INFO = {
    Intent.QUERY_ORDER.value: ["order_id"],
    Intent.APPLY_REFUND.value: ["order_id", "reason"]
}

# 工具函数
class OrderTools:
    """订单相关工具函数"""

    @staticmethod
    def query_order(order_id: str) -> Dict[str, Any]:
        """查询订单信息"""
        orders = {
            "ORD123456": {
                "status": "已发货",
                "product": "智能手机",
                "order_date": "2024-06-10",
                "estimated_delivery": "2024-06-13",
                "shipping_company": "顺丰速运",
                "tracking_number": "SF1234567890"
            },
            "ORD789012": {
                "status": "处理中",
                "product": "笔记本电脑",
                "order_date": "2024-06-11",
                "estimated_delivery": "2024-06-15",
                "shipping_company": None,
                "tracking_number": None
            }
        }

        if order_id in orders:
            return {"success": True, "data": orders[order_id]}
        else:
            return {"success": False, "error": f"未找到订单 {order_id}"}

    @staticmethod
    def apply_refund(order_id: str, reason: str) -> Dict[str, Any]:
        """申请退款"""
        refund_id = f"REF{order_id[3:]}"
        return {
            "success": True,
            "refund_id": refund_id,
            "message": f"退款申请已提交，退款单号: {refund_id}",
            "estimated_processing": "3-5个工作日"
        }

# 意图识别节点
def intent_recognition_node(state: CustomerServiceState) -> CustomerServiceState:
    """识别用户意图"""
    state["step_count"] += 1

    # 检查步骤限制
    if state["step_count"] > state["max_steps"]:
        state["response"] = "抱歉，对话轮次过多，请重新开始咨询。"
        state["conversation_finished"] = True
        return state

    intent_prompt = PromptTemplate(
        template="""根据用户输入识别意图。可选意图：
- query_order: 用户想要查询订单状态
- apply_refund: 用户想要申请退款
- general_query: 一般咨询问题

用户输入: {user_input}
对话历史: {chat_history}

请只返回意图名称，不要返回其他内容。""",
        input_variables=["user_input", "chat_history"]
    )

    llm = ChatOpenAI(
        model_name="gpt-3.5-turbo",
        temperature=0,
        openai_api_key=os.getenv("OPENAI_API_KEY")
    )

    chat_history_text = "\n".join([
        f"{msg['role']}: {msg['content']}" for msg in state["chat_history"][-3:]
    ]) if state["chat_history"] else "无"

    try:
        intent_result = llm.invoke(intent_prompt.format(
            user_input=state["user_input"],
            chat_history=chat_history_text
        ))

        intent_str = intent_result.content.strip().lower()
        state["current_intent"] = intent_str

    except Exception:
        # 如果LLM调用失败，使用基于关键词的后备方案
        user_input_lower = state["user_input"].lower()
        if any(keyword in user_input_lower for keyword in ["查询", "订单", "查订单"]):
            state["current_intent"] = Intent.QUERY_ORDER.value
        elif any(keyword in user_input_lower for keyword in ["退款", "退货"]):
            state["current_intent"] = Intent.APPLY_REFUND.value
        else:
            state["current_intent"] = Intent.GENERAL_QUERY.value

    return state

# 信息收集节点
def information_collection_node(state: CustomerServiceState) -> CustomerServiceState:
    """收集执行工具所需的信息"""
    state["step_count"] += 1

    # 确保missing_info存在
    if "missing_info" not in state:
        state["missing_info"] = {}

    # 获取当前意图所需的信息字段
    required_fields = REQUIRED_INFO.get(state["current_intent"], [])

    # 确保missing_info包含所有必需字段
    for field in required_fields:
        if field not in state["missing_info"]:
            state["missing_info"][field] = None

        # 提取订单号
        order_match = re.search(r'[A-Za-z]{3}\d{6,}', state["user_input"])
        if order_match and state["missing_info"].get("order_id") is None:
            state["missing_info"]["order_id"] = order_match.group().upper()

        # 如果是退款申请且还没有原因，尝试提取原因
        if (state["current_intent"] == Intent.APPLY_REFUND.value and
                state["missing_info"].get("order_id") and
                state["missing_info"].get("reason") is None):

            # 简单的关键词提取
            reason_keywords = {
                "质量": "商品质量问题",
                "损坏": "商品损坏",
                "不满意": "对商品不满意",
                "错误": "订单信息错误",
                "不想要": "不再需要此商品",
                "不喜欢": "对商品不喜欢"
            }

            for keyword, reason in reason_keywords.items():
                if keyword in state["user_input"]:
                    state["missing_info"]["reason"] = reason
                    break

            if not state["missing_info"].get("reason"):
                # 如果没有匹配到关键词，使用用户原始输入作为原因
                state["missing_info"]["reason"] = state["user_input"]
    # 根据意图和缺失信息生成相应的询问
    if state["current_intent"] == Intent.QUERY_ORDER.value:
        if state["missing_info"]["order_id"] is None:
            state["response"] = "请问您的订单号是多少？"
        else:
            state["information_collect"] = True
            return state

    elif state["current_intent"] == Intent.APPLY_REFUND.value:
        if state["missing_info"]["order_id"] is None:
            state["response"] = "请问您要申请退款的订单号是多少？"
        elif state["missing_info"]["reason"] is None:
            state["response"] = "请问您申请退款的原因是什么？"
        else:
            state["information_collect"] = True
            return state

    return state

# 信息提取节点
def information_extraction_node(state: CustomerServiceState) -> CustomerServiceState:
    """从用户输入中提取所需信息"""
    state["step_count"] += 1

    # 确保missing_info存在
    if "missing_info" not in state:
        state["missing_info"] = {}

    print(f"[DEBUG] information_extraction_node - 开始 - missing_info: {state['missing_info']}")

    if state["current_intent"] in [Intent.QUERY_ORDER.value, Intent.APPLY_REFUND.value]:
        # 获取当前意图所需的信息字段
        required_fields = REQUIRED_INFO.get(state["current_intent"], [])

        # 确保所有必需字段都在missing_info中
        for field in required_fields:
            if field not in state["missing_info"]:
                state["missing_info"][field] = None

        # 提取订单号
        order_match = re.search(r'[A-Za-z]{3}\d{6,}', state["user_input"])
        if order_match and state["missing_info"].get("order_id") is None:
            state["missing_info"]["order_id"] = order_match.group().upper()

        # 如果是退款申请且还没有原因，尝试提取原因
        if (state["current_intent"] == Intent.APPLY_REFUND.value and
                state["missing_info"].get("order_id") and
                state["missing_info"].get("reason") is None):

            # 简单的关键词提取
            reason_keywords = {
                "质量": "商品质量问题",
                "损坏": "商品损坏",
                "不满意": "对商品不满意",
                "错误": "订单信息错误",
                "不想要": "不再需要此商品",
                "不喜欢": "对商品不喜欢"
            }

            for keyword, reason in reason_keywords.items():
                if keyword in state["user_input"]:
                    state["missing_info"]["reason"] = reason
                    break

            if not state["missing_info"].get("reason"):
                # 如果没有匹配到关键词，使用用户原始输入作为原因
                state["missing_info"]["reason"] = state["user_input"]

    print(f"[DEBUG] information_extraction_node - 结束 - missing_info: {state['missing_info']}")

    return state

# 工具调用节点
def tool_call_node(state: CustomerServiceState) -> CustomerServiceState:
    """调用相应的工具函数"""
    state["step_count"] += 1

    # 确保missing_info存在
    if "missing_info" not in state:
        state["missing_info"] = {}

    print(f"[DEBUG] tool_call_node - missing_info: {state['missing_info']}")

    if state["current_intent"] == Intent.QUERY_ORDER.value and state["missing_info"].get("order_id"):
        result = OrderTools.query_order(state["missing_info"]["order_id"])
        state["tool_result"] = result

        if result["success"]:
            order = result["data"]
            state["response"] = f"订单状态: {order['status']}\n"
            state["response"] += f"商品: {order['product']}\n"
            state["response"] += f"下单日期: {order['order_date']}\n"
            state["response"] += f"预计送达: {order['estimated_delivery']}"

            if order.get("tracking_number"):
                state["response"] += f"\n快递公司: {order['shipping_company']}"
                state["response"] += f"\n运单号: {order['tracking_number']}"
        else:
            state["response"] = f"抱歉，{result['error']}"

        state["conversation_finished"] = True

    elif state["current_intent"] == Intent.APPLY_REFUND.value and state["missing_info"].get("order_id") and state["missing_info"].get("reason"):
        result = OrderTools.apply_refund(
            state["missing_info"]["order_id"],
            state["missing_info"]["reason"]
        )
        state["tool_result"] = result
        state["response"] = result["message"]
        state["conversation_finished"] = True

    return state

# 通用回复节点
def general_response_node(state: CustomerServiceState) -> CustomerServiceState:
    """处理一般性查询"""
    state["step_count"] += 1

    if state["current_intent"] == Intent.GENERAL_QUERY.value:
        prompt_template = PromptTemplate(
            input_variables=["time_context", "user_input", "chat_history"],
            template="""你是一个智能客服助手，需要准确理解用户的时间相关查询，并结合当前时间进行回答。

{time_context}

对话历史:
{chat_history}

用户输入: {user_input}

请提供有帮助的回复。

客服回复:"""
        )

        llm = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.3,
            openai_api_key=os.getenv("OPENAI_API_KEY")
        )

        # 获取时间上下文
        now = datetime.now()
        current_time = now.strftime("%Y年%m月%d日 %H:%M:%S")
        yesterday = (now - timedelta(days=1)).strftime("%Y年%m月%d日")
        tomorrow = (now + timedelta(days=1)).strftime("%Y年%m月%d日")

        time_context = f"""
当前时间: {current_time}
相关日期:
- 昨天: {yesterday}
- 今天: {now.strftime('%Y年%m月%d日')}
- 明天: {tomorrow}
"""

        # 获取对话历史
        chat_history_text = "\n".join([
            f"{msg['role']}: {msg['content']}" for msg in state["chat_history"][-3:]
        ]) if state["chat_history"] else "无"

        try:
            result = llm.invoke(prompt_template.format(
                time_context=time_context,
                user_input=state["user_input"],
                chat_history=chat_history_text
            ))

            state["response"] = result.content
        except Exception:
            state["response"] = "抱歉，我现在无法处理您的请求，请稍后再试。"

        state["conversation_finished"] = True

    return state

# 更新历史节点
def update_history_node(state: CustomerServiceState) -> CustomerServiceState:
    """更新对话历史"""
    # 添加用户消息到历史
    state["chat_history"].append({
        "role": "user",
        "content": state["user_input"]
    })

    # 添加AI响应到历史
    if state["response"]:
        state["chat_history"].append({
            "role": "assistant",
            "content": state["response"]
        })

    return state

# 修复的路由函数
def route_conversation(state: CustomerServiceState) -> str:
    """决定下一步执行哪个节点"""

    # 如果已经生成响应，更新历史后结束
    if state.get("response"):
        return "update_history"

    # 如果对话已结束，更新历史后结束
    if state.get("conversation_finished", False):
        return "update_history"

    # 检查步骤限制
    if state.get("step_count", 0) > state.get("max_steps", 20):
        state["response"] = "为了更好的服务体验，本次对话将结束。如有需要请重新咨询。"
        return "update_history"

    current_intent = state.get("current_intent", Intent.UNKNOWN.value)

    # 一般查询直接回复
    if current_intent == Intent.GENERAL_QUERY.value:
        return "general_response"

    # 工具类意图的处理流程
    # 获取当前意图所需的信息字段
    required_fields = REQUIRED_INFO.get(current_intent, [])

    # 确保missing_info存在
    if "missing_info" not in state:
        state["missing_info"] = {}

    # 确保missing_info包含所有必需字段
    for field in required_fields:
        if field not in state["missing_info"]:
            state["missing_info"][field] = None

    print(f"[DEBUG] route_conversation - missing_info: {state['missing_info']}")

    # 检查是否所有必需信息都已收集
    all_info_collected = all(state["missing_info"].get(field) is not None for field in required_fields)

    if all_info_collected:
        return "tool_call"
    else:
        # 如果有用户输入但还没有响应，尝试提取信息
        if not state["information_collected"]:
            return "information_collection"
        elif not state["information_extracted"]:
            return "information_extraction"

# 构建图
def create_customer_service_graph():
    """创建客服对话图"""
    workflow = StateGraph(CustomerServiceState)

    # 添加节点
    workflow.add_node("intent_recognition", intent_recognition_node)
    workflow.add_node("information_collection", information_collection_node)
    workflow.add_node("information_extraction", information_extraction_node)
    workflow.add_node("tool_call", tool_call_node)
    workflow.add_node("general_response", general_response_node)
    workflow.add_node("update_history", update_history_node)

    # 设置入口点
    workflow.set_entry_point("intent_recognition")

    # 条件边
    workflow.add_conditional_edges(
        "intent_recognition",
        route_conversation,
        {
            "information_collection": "information_collection",
            "information_extraction": "information_extraction",
            "tool_call": "tool_call",
            "general_response": "general_response",
            "update_history": "update_history"
        }
    )

    workflow.add_conditional_edges(
        "information_collection",
        route_conversation,
        {
            "information_collection": "information_collection",
            "information_extraction": "information_extraction",
            "tool_call": "tool_call",
            "general_response": "general_response",
            "update_history": "update_history"
        }
    )

    workflow.add_conditional_edges(
        "information_extraction",
        route_conversation,
        {
            "information_collection": "information_collection",
            "tool_call": "tool_call",
            "general_response": "general_response",
            "update_history": "update_history"
        }
    )

    workflow.add_conditional_edges(
        "tool_call",
        route_conversation,
        {
            "update_history": "update_history"
        }
    )

    workflow.add_conditional_edges(
        "general_response",
        route_conversation,
        {
            "update_history": "update_history"
        }
    )

    # 添加最终边
    workflow.add_edge("update_history", END)

    return workflow.compile()

# 增强版客服系统
class EnhancedCustomerService:
    """支持多轮对话和工具调用的客服系统"""

    def __init__(self, model_name: str = "gpt-3.5-turbo"):
        self.model_name = model_name
        self.graph = create_customer_service_graph()
        self.conversation_id = id(self)

    def process_message(self, user_input: str, chat_history: List[Dict[str, str]] = None) -> Dict[str, Any]:
        """处理用户消息"""

        if chat_history is None:
            chat_history = []

        # 准备初始状态
        initial_state = {
            "user_input": user_input,
            "chat_history": chat_history,
            "current_intent": Intent.UNKNOWN.value,
            "missing_info": {},
            "tool_result": None,
            "response": "",
            "conversation_finished": False,
            "step_count": 0,
            "max_steps": 10,
            "information_collected": False,
            "information_extracted": False
        }

        try:
            # 执行图，显式设置递归限制
            config = {"recursion_limit": 50}
            final_state = self.graph.invoke(initial_state, config=config)

            return {
                "response": final_state["response"],
                "chat_history": final_state["chat_history"],
                "current_intent": final_state["current_intent"],
                "tool_used": final_state.get("tool_result") is not None,
                "conversation_id": self.conversation_id
            }
        except Exception as e:
            # 错误处理
            error_response = f"抱歉，系统暂时无法处理您的请求: {str(e)}"
            chat_history.extend([
                {"role": "user", "content": user_input},
                {"role": "assistant", "content": error_response}
            ])

            return {
                "response": error_response,
                "chat_history": chat_history,
                "current_intent": Intent.UNKNOWN.value,
                "tool_used": False,
                "conversation_id": self.conversation_id
            }

# 测试函数
def main():
    """测试增强版系统"""
    service = EnhancedCustomerService()

    # 分步测试用例
    test_cases = [
        # 订单查询流程
        "我想查询我的订单",
        "ORD123456",
    ]

    print("=== 增强版客服系统测试 ===\n")

    chat_history = []
    for i, user_input in enumerate(test_cases):
        print(f"用户 [{i+1}]: {user_input}")
        result = service.process_message(user_input, chat_history)
        chat_history = result["chat_history"]

        print(f"客服: {result['response']}")
        print(f"意图: {result['current_intent']}")
        if result['tool_used']:
            print("[工具已调用]")
        print("-" * 50)

if __name__ == "__main__":
    main()

运行结果：

=== 增强版客服系统测试 ===

用户 [1]: 我想查询我的订单
[DEBUG] route_conversation - missing_info: {'order_id': None}
客服: 请问您的订单号是多少？
意图: query_order
--------------------------------------------------
用户 [2]: ORD123456
[DEBUG] route_conversation - missing_info: {'order_id': None}
[DEBUG] route_conversation - missing_info: {'order_id': 'ORD123456'}
[DEBUG] tool_call_node - missing_info: {'order_id': 'ORD123456'}
客服: 订单状态: 已发货
商品: 智能手机
下单日期: 2024-06-10
预计送达: 2024-06-13
快递公司: 顺丰速运
运单号: SF1234567890
意图: query_order
[工具已调用]
--------------------------------------------------

阶段三：热更新与生产部署

实现模型与插件的热更新，完成系统部署与监控。

1. 模型热更新
2. 插件热重载
3. 暴露健康检查接口 /health
4. 编写自动化测试脚本

• 测试“发票开具”插件的功能正确性
• 验证热更新后旧会话不受影响

具体代码：

import os
import time
import json
import threading
from datetime import datetime, timedelta
from typing import Dict, Any, List, TypedDict, Optional
from enum import Enum
from dotenv import load_dotenv
import re
import importlib
import asyncio
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse
import uvicorn
import logging
from logging.handlers import RotatingFileHandler
import requests
import pytest

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        RotatingFileHandler('customer_service.log', maxBytes=10485760, backupCount=5),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 加载环境变量
load_dotenv()

from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END

# 定义状态类型
class CustomerServiceState(TypedDict):
    user_input: str
    chat_history: List[Dict[str, str]]
    current_intent: str
    missing_info: Dict[str, Any]
    tool_result: Any
    response: str
    conversation_finished: bool
    step_count: int
    max_steps: int
    information_collected: bool
    information_extracted: bool

# 定义意图枚举
class Intent(Enum):
    QUERY_ORDER = "query_order"
    APPLY_REFUND = "apply_refund"
    CREATE_INVOICE = "create_invoice"  # 新增发票开具意图
    GENERAL_QUERY = "general_query"
    UNKNOWN = "unknown"

# 定义每个意图所需的信息
REQUIRED_INFO = {
    Intent.QUERY_ORDER.value: ["order_id"],
    Intent.APPLY_REFUND.value: ["order_id", "reason"],
    Intent.CREATE_INVOICE.value: ["order_id", "invoice_title", "tax_number"]  # 新增发票开具所需信息
}

# 插件管理器
class PluginManager:
    """插件管理器，支持热重载"""

    def __init__(self):
        self.plugins = {}
        self.plugin_versions = {}
        self.load_plugins()

    def load_plugins(self):
        """加载所有插件"""
        self.plugins = {
            "order_tools": OrderTools(),
            "invoice_tools": InvoiceTools()  # 新增发票工具
        }
        self.plugin_versions = {
            "order_tools": "1.0.0",
            "invoice_tools": "1.0.0"
        }
        logger.info("所有插件已加载")

    def reload_plugin(self, plugin_name: str):
        """重新加载指定插件"""
        try:
            if plugin_name == "order_tools":
                # 重新加载OrderTools类
                global OrderTools
                importlib.reload(sys.modules[__name__])
                self.plugins[plugin_name] = OrderTools()
            elif plugin_name == "invoice_tools":
                # 重新加载InvoiceTools类
                global InvoiceTools
                importlib.reload(sys.modules[__name__])
                self.plugins[plugin_name] = InvoiceTools()

            # 更新版本号
            current_version = self.plugin_versions[plugin_name]
            major, minor, patch = current_version.split('.')
            new_version = f"{major}.{minor}.{int(patch) + 1}"
            self.plugin_versions[plugin_name] = new_version

            logger.info(f"插件 {plugin_name} 已重新加载，版本: {new_version}")
            return True
        except Exception as e:
            logger.error(f"重新加载插件 {plugin_name} 失败: {str(e)}")
            return False

    def get_plugin(self, plugin_name: str):
        """获取插件实例"""
        return self.plugins.get(plugin_name)

    def get_plugin_info(self):
        """获取所有插件信息"""
        return {
            name: {
                "version": version,
                "status": "loaded"
            } for name, version in self.plugin_versions.items()
        }

# 模型管理器
class ModelManager:
    """模型管理器，支持热更新"""

    def __init__(self):
        self.current_model = "gpt-3.5-turbo"
        self.model_config = {
            "gpt-3.5-turbo": {
                "temperature": 0.3,
                "max_tokens": 1000
            },
            "gpt-4": {
                "temperature": 0.3,
                "max_tokens": 1000
            }
        }
        self.model_history = []
        self.update_lock = threading.Lock()

    def get_model(self, model_name: str = None):
        """获取当前模型实例"""
        if model_name is None:
            model_name = self.current_model

        config = self.model_config.get(model_name, self.model_config["gpt-3.5-turbo"])

        return ChatOpenAI(
            model_name=model_name,
            temperature=config["temperature"],
            max_tokens=config["max_tokens"],
            openai_api_key=os.getenv("OPENAI_API_KEY")
        )

    def update_model(self, new_model: str, config: Dict[str, Any] = None):
        """更新模型配置"""
        with self.update_lock:
            old_model = self.current_model
            self.current_model = new_model

            if config:
                self.model_config[new_model] = config

            # 记录模型更新历史
            self.model_history.append({
                "timestamp": datetime.now().isoformat(),
                "from": old_model,
                "to": new_model,
                "config": self.model_config[new_model]
            })

            logger.info(f"模型已更新: {old_model} -> {new_model}")
            return True

    def get_model_info(self):
        """获取模型信息"""
        return {
            "current_model": self.current_model,
            "config": self.model_config[self.current_model],
            "available_models": list(self.model_config.keys()),
            "update_history": self.model_history[-5:]  # 最近5次更新
        }

# 工具函数
class OrderTools:
    """订单相关工具函数"""

    @staticmethod
    def query_order(order_id: str) -> Dict[str, Any]:
        """查询订单信息"""
        orders = {
            "ORD123456": {
                "status": "已发货",
                "product": "智能手机",
                "order_date": "2024-06-10",
                "estimated_delivery": "2024-06-13",
                "shipping_company": "顺丰速运",
                "tracking_number": "SF1234567890"
            },
            "ORD789012": {
                "status": "处理中",
                "product": "笔记本电脑",
                "order_date": "2024-06-11",
                "estimated_delivery": "2024-06-15",
                "shipping_company": None,
                "tracking_number": None
            }
        }

        if order_id in orders:
            return {"success": True, "data": orders[order_id]}
        else:
            return {"success": False, "error": f"未找到订单 {order_id}"}

    @staticmethod
    def apply_refund(order_id: str, reason: str) -> Dict[str, Any]:
        """申请退款"""
        refund_id = f"REF{order_id[3:]}"
        return {
            "success": True,
            "refund_id": refund_id,
            "message": f"退款申请已提交，退款单号: {refund_id}",
            "estimated_processing": "3-5个工作日"
        }

# 新增发票工具
class InvoiceTools:
    """发票相关工具函数"""

    @staticmethod
    def create_invoice(order_id: str, invoice_title: str, tax_number: str = None) -> Dict[str, Any]:
        """开具发票"""
        # 模拟发票开具
        invoice_id = f"INV{order_id[3:]}"

        invoice_data = {
            "invoice_id": invoice_id,
            "order_id": order_id,
            "invoice_title": invoice_title,
            "tax_number": tax_number,
            "issue_date": datetime.now().strftime("%Y-%m-%d"),
            "status": "已开具",
            "download_url": f"https://example.com/invoices/{invoice_id}.pdf"
        }

        return {
            "success": True,
            "message": f"发票开具成功，发票号: {invoice_id}",
            "data": invoice_data
        }

    @staticmethod
    def query_invoice(invoice_id: str) -> Dict[str, Any]:
        """查询发票状态"""
        # 模拟发票查询
        invoices = {
            "INV123456": {
                "status": "已开具",
                "order_id": "ORD123456",
                "issue_date": "2024-06-12",
                "amount": "5999.00"
            }
        }

        if invoice_id in invoices:
            return {"success": True, "data": invoices[invoice_id]}
        else:
            return {"success": False, "error": f"未找到发票 {invoice_id}"}

# 意图识别节点
def intent_recognition_node(state: CustomerServiceState) -> CustomerServiceState:
    """识别用户意图"""
    state["step_count"] += 1

    # 检查步骤限制
    if state["step_count"] > state["max_steps"]:
        state["response"] = "抱歉，对话轮次过多，请重新开始咨询。"
        state["conversation_finished"] = True
        return state

    intent_prompt = PromptTemplate(
        template="""根据用户输入识别意图。可选意图：
- query_order: 用户想要查询订单状态
- apply_refund: 用户想要申请退款
- create_invoice: 用户想要开具发票
- general_query: 一般咨询问题

用户输入: {user_input}
对话历史: {chat_history}

请只返回意图名称，不要返回其他内容。""",
        input_variables=["user_input", "chat_history"]
    )

    llm = app_state.model_manager.get_model()

    chat_history_text = "\n".join([
        f"{msg['role']}: {msg['content']}" for msg in state["chat_history"][-3:]
    ]) if state["chat_history"] else "无"

    try:
        intent_result = llm.invoke(intent_prompt.format(
            user_input=state["user_input"],
            chat_history=chat_history_text
        ))

        intent_str = intent_result.content.strip().lower()
        state["current_intent"] = intent_str

    except Exception:
        # 如果LLM调用失败，使用基于关键词的后备方案
        user_input_lower = state["user_input"].lower()
        if any(keyword in user_input_lower for keyword in ["查询", "订单", "查订单"]):
            state["current_intent"] = Intent.QUERY_ORDER.value
        elif any(keyword in user_input_lower for keyword in ["退款", "退货"]):
            state["current_intent"] = Intent.APPLY_REFUND.value
        elif any(keyword in user_input_lower for keyword in ["发票", "开票", "发票开具"]):
            state["current_intent"] = Intent.CREATE_INVOICE.value
        else:
            state["current_intent"] = Intent.GENERAL_QUERY.value

    return state

# 信息收集节点
def information_collection_node(state: CustomerServiceState) -> CustomerServiceState:
    """收集执行工具所需的信息"""
    state["step_count"] += 1

    # 确保missing_info存在
    if "missing_info" not in state:
        state["missing_info"] = {}

    # 获取当前意图所需的信息字段
    required_fields = REQUIRED_INFO.get(state["current_intent"], [])

    # 确保missing_info包含所有必需字段
    for field in required_fields:
        if field not in state["missing_info"]:
            state["missing_info"][field] = None

    # 根据意图和缺失信息生成相应的询问
    if state["current_intent"] == Intent.QUERY_ORDER.value:
        if state["missing_info"]["order_id"] is None:
            state["response"] = "请问您的订单号是多少？"
        else:
            state["information_collected"] = True
            return state

    elif state["current_intent"] == Intent.APPLY_REFUND.value:
        if state["missing_info"]["order_id"] is None:
            state["response"] = "请问您要申请退款的订单号是多少？"
        elif state["missing_info"]["reason"] is None:
            state["response"] = "请问您申请退款的原因是什么？"
        else:
            state["information_collected"] = True
            return state

    elif state["current_intent"] == Intent.CREATE_INVOICE.value:
        if state["missing_info"]["order_id"] is None:
            state["response"] = "请问您要为哪个订单开具发票？请提供订单号。"
        elif state["missing_info"]["invoice_title"] is None:
            state["response"] = "请问发票抬头是什么？"
        elif state["missing_info"]["tax_number"] is None:
            state["response"] = "请问纳税人识别号是什么？（如不需要可回复'无'）"
        else:
            state["information_collected"] = True
            return state

    return state

# 信息提取节点
def information_extraction_node(state: CustomerServiceState) -> CustomerServiceState:
    """从用户输入中提取所需信息"""
    state["step_count"] += 1

    # 确保missing_info存在
    if "missing_info" not in state:
        state["missing_info"] = {}

    if state["current_intent"] in [Intent.QUERY_ORDER.value, Intent.APPLY_REFUND.value, Intent.CREATE_INVOICE.value]:
        # 获取当前意图所需的信息字段
        required_fields = REQUIRED_INFO.get(state["current_intent"], [])

        # 确保所有必需字段都在missing_info中
        for field in required_fields:
            if field not in state["missing_info"]:
                state["missing_info"][field] = None

        # 提取订单号
        order_match = re.search(r'[A-Za-z]{3}\d{6,}', state["user_input"])
        if order_match and state["missing_info"].get("order_id") is None:
            state["missing_info"]["order_id"] = order_match.group().upper()

        # 提取发票抬头和税号
        if state["current_intent"] == Intent.CREATE_INVOICE.value:
            # 简单的发票抬头提取（假设用户直接提供了抬头）
            if state["missing_info"].get("invoice_title") is None and len(state["user_input"]) > 2:
                # 如果不是订单号和税号，且长度合适，认为是发票抬头
                if not order_match and not re.search(r'\d{15,20}', state["user_input"]):
                    state["missing_info"]["invoice_title"] = state["user_input"]

            # 提取税号（15-20位数字）
            tax_match = re.search(r'\d{15,20}', state["user_input"])
            if tax_match and state["missing_info"].get("tax_number") is None:
                state["missing_info"]["tax_number"] = tax_match.group()

            # 如果用户说"无"或"不需要"，设置税号为空
            if "无" in state["user_input"] or "不需要" in state["user_input"]:
                state["missing_info"]["tax_number"] = ""

        # 如果是退款申请且还没有原因，尝试提取原因
        if (state["current_intent"] == Intent.APPLY_REFUND.value and
                state["missing_info"].get("order_id") and
                state["missing_info"].get("reason") is None):

            # 简单的关键词提取
            reason_keywords = {
                "质量": "商品质量问题",
                "损坏": "商品损坏",
                "不满意": "对商品不满意",
                "错误": "订单信息错误",
                "不想要": "不再需要此商品",
                "不喜欢": "对商品不喜欢"
            }

            for keyword, reason in reason_keywords.items():
                if keyword in state["user_input"]:
                    state["missing_info"]["reason"] = reason
                    break

            if not state["missing_info"].get("reason"):
                # 如果没有匹配到关键词，使用用户原始输入作为原因
                state["missing_info"]["reason"] = state["user_input"]

    state["information_extracted"] = True
    return state

# 工具调用节点
def tool_call_node(state: CustomerServiceState) -> CustomerServiceState:
    """调用相应的工具函数"""
    state["step_count"] += 1

    # 确保missing_info存在
    if "missing_info" not in state:
        state["missing_info"] = {}

    if state["current_intent"] == Intent.QUERY_ORDER.value and state["missing_info"].get("order_id"):
        order_tools = app_state.plugin_manager.get_plugin("order_tools")
        result = order_tools.query_order(state["missing_info"]["order_id"])
        state["tool_result"] = result

        if result["success"]:
            order = result["data"]
            state["response"] = f"订单状态: {order['status']}\n"
            state["response"] += f"商品: {order['product']}\n"
            state["response"] += f"下单日期: {order['order_date']}\n"
            state["response"] += f"预计送达: {order['estimated_delivery']}"

            if order.get("tracking_number"):
                state["response"] += f"\n快递公司: {order['shipping_company']}"
                state["response"] += f"\n运单号: {order['tracking_number']}"
        else:
            state["response"] = f"抱歉，{result['error']}"

        state["conversation_finished"] = True

    elif state["current_intent"] == Intent.APPLY_REFUND.value and state["missing_info"].get("order_id") and state["missing_info"].get("reason"):
        order_tools = app_state.plugin_manager.get_plugin("order_tools")
        result = order_tools.apply_refund(
            state["missing_info"]["order_id"],
            state["missing_info"]["reason"]
        )
        state["tool_result"] = result
        state["response"] = result["message"]
        state["conversation_finished"] = True

    elif state["current_intent"] == Intent.CREATE_INVOICE.value and state["missing_info"].get("order_id") and state["missing_info"].get("invoice_title"):
        invoice_tools = app_state.plugin_manager.get_plugin("invoice_tools")
        result = invoice_tools.create_invoice(
            state["missing_info"]["order_id"],
            state["missing_info"]["invoice_title"],
            state["missing_info"].get("tax_number")
        )
        state["tool_result"] = result
        state["response"] = result["message"]
        state["conversation_finished"] = True

    return state

# 通用回复节点
def general_response_node(state: CustomerServiceState) -> CustomerServiceState:
    """处理一般性查询"""
    state["step_count"] += 1

    if state["current_intent"] == Intent.GENERAL_QUERY.value:
        prompt_template = PromptTemplate(
            input_variables=["time_context", "user_input", "chat_history"],
            template="""你是一个智能客服助手，需要准确理解用户的时间相关查询，并结合当前时间进行回答。

{time_context}

对话历史:
{chat_history}

用户输入: {user_input}

请提供有帮助的回复。

客服回复:"""
        )

        llm = app_state.model_manager.get_model()

        # 获取时间上下文
        now = datetime.now()
        current_time = now.strftime("%Y年%m月%d日 %H:%M:%S")
        yesterday = (now - timedelta(days=1)).strftime("%Y年%m月%d日")
        tomorrow = (now + timedelta(days=1)).strftime("%Y年%m月%d日")

        time_context = f"""
当前时间: {current_time}
相关日期:
- 昨天: {yesterday}
- 今天: {now.strftime('%Y年%m月%d日')}
- 明天: {tomorrow}
"""

        # 获取对话历史
        chat_history_text = "\n".join([
            f"{msg['role']}: {msg['content']}" for msg in state["chat_history"][-3:]
        ]) if state["chat_history"] else "无"

        try:
            result = llm.invoke(prompt_template.format(
                time_context=time_context,
                user_input=state["user_input"],
                chat_history=chat_history_text
            ))

            state["response"] = result.content
        except Exception:
            state["response"] = "抱歉，我现在无法处理您的请求，请稍后再试。"

        state["conversation_finished"] = True

    return state

# 更新历史节点
def update_history_node(state: CustomerServiceState) -> CustomerServiceState:
    """更新对话历史"""
    # 添加用户消息到历史
    state["chat_history"].append({
        "role": "user",
        "content": state["user_input"]
    })

    # 添加AI响应到历史
    if state["response"]:
        state["chat_history"].append({
            "role": "assistant",
            "content": state["response"]
        })

    return state

# 修复的路由函数
def route_conversation(state: CustomerServiceState) -> str:
    """决定下一步执行哪个节点"""

    # 如果已经生成响应，更新历史后结束
    if state.get("response"):
        return "update_history"

    # 如果对话已结束，更新历史后结束
    if state.get("conversation_finished", False):
        return "update_history"

    # 检查步骤限制
    if state.get("step_count", 0) > state.get("max_steps", 20):
        state["response"] = "为了更好的服务体验，本次对话将结束。如有需要请重新咨询。"
        return "update_history"

    current_intent = state.get("current_intent", Intent.UNKNOWN.value)

    # 一般查询直接回复
    if current_intent == Intent.GENERAL_QUERY.value:
        return "general_response"

    # 工具类意图的处理流程
    # 获取当前意图所需的信息字段
    required_fields = REQUIRED_INFO.get(current_intent, [])

    # 确保missing_info存在
    if "missing_info" not in state:
        state["missing_info"] = {}

    # 确保missing_info包含所有必需字段
    for field in required_fields:
        if field not in state["missing_info"]:
            state["missing_info"][field] = None

    # 检查是否所有必需信息都已收集
    all_info_collected = all(state["missing_info"].get(field) is not None for field in required_fields)

    if all_info_collected:
        return "tool_call"
    else:
        # 如果有用户输入但还没有响应，尝试提取信息
        if not state.get("information_collected", False):
            return "information_collection"
        elif not state.get("information_extracted", False):
            return "information_extraction"
        else:
            return "information_collection"

# 构建图
def create_customer_service_graph():
    """创建客服对话图"""
    workflow = StateGraph(CustomerServiceState)

    # 添加节点
    workflow.add_node("intent_recognition", intent_recognition_node)
    workflow.add_node("information_collection", information_collection_node)
    workflow.add_node("information_extraction", information_extraction_node)
    workflow.add_node("tool_call", tool_call_node)
    workflow.add_node("general_response", general_response_node)
    workflow.add_node("update_history", update_history_node)

    # 设置入口点
    workflow.set_entry_point("intent_recognition")

    # 条件边
    workflow.add_conditional_edges(
        "intent_recognition",
        route_conversation,
        {
            "information_collection": "information_collection",
            "information_extraction": "information_extraction",
            "tool_call": "tool_call",
            "general_response": "general_response",
            "update_history": "update_history"
        }
    )

    workflow.add_conditional_edges(
        "information_collection",
        route_conversation,
        {
            "information_collection": "information_collection",
            "information_extraction": "information_extraction",
            "tool_call": "tool_call",
            "general_response": "general_response",
            "update_history": "update_history"
        }
    )

    workflow.add_conditional_edges(
        "information_extraction",
        route_conversation,
        {
            "information_collection": "information_collection",
            "tool_call": "tool_call",
            "general_response": "general_response",
            "update_history": "update_history"
        }
    )

    workflow.add_conditional_edges(
        "tool_call",
        route_conversation,
        {
            "update_history": "update_history"
        }
    )

    workflow.add_conditional_edges(
        "general_response",
        route_conversation,
        {
            "update_history": "update_history"
        }
    )

    # 添加最终边
    workflow.add_edge("update_history", END)

    return workflow.compile()

# 增强版客服系统
class EnhancedCustomerService:
    """支持多轮对话和工具调用的客服系统"""

    def __init__(self, model_name: str = "gpt-3.5-turbo"):
        self.model_name = model_name
        self.graph = create_customer_service_graph()
        self.conversation_sessions = {}  # 存储会话状态
        self.session_timeout = 3600  # 会话超时时间（秒）

    def _cleanup_sessions(self):
        """清理过期的会话"""
        current_time = time.time()
        expired_sessions = []

        for session_id, session_data in self.conversation_sessions.items():
            if current_time - session_data.get("last_activity", 0) > self.session_timeout:
                expired_sessions.append(session_id)

        for session_id in expired_sessions:
            del self.conversation_sessions[session_id]
            logger.info(f"清理过期会话: {session_id}")

    def process_message(self, user_input: str, session_id: str = None, chat_history: List[Dict[str, str]] = None) -> Dict[str, Any]:
        """处理用户消息"""

        # 清理过期会话
        self._cleanup_sessions()

        if session_id is None:
            session_id = f"session_{int(time.time())}_{id(self)}"

        if chat_history is None:
            if session_id in self.conversation_sessions:
                chat_history = self.conversation_sessions[session_id]["chat_history"]
            else:
                chat_history = []

        # 准备初始状态
        initial_state = {
            "user_input": user_input,
            "chat_history": chat_history,
            "current_intent": Intent.UNKNOWN.value,
            "missing_info": {},
            "tool_result": None,
            "response": "",
            "conversation_finished": False,
            "step_count": 0,
            "max_steps": 10,
            "information_collected": False,
            "information_extracted": False
        }

        try:
            # 执行图，显式设置递归限制
            config = {"recursion_limit": 50}
            final_state = self.graph.invoke(initial_state, config=config)

            # 更新会话状态
            self.conversation_sessions[session_id] = {
                "chat_history": final_state["chat_history"],
                "last_activity": time.time()
            }

            return {
                "response": final_state["response"],
                "chat_history": final_state["chat_history"],
                "current_intent": final_state["current_intent"],
                "tool_used": final_state.get("tool_result") is not None,
                "session_id": session_id
            }
        except Exception as e:
            # 错误处理
            logger.error(f"处理消息时出错: {str(e)}")
            error_response = "抱歉，系统暂时无法处理您的请求，请稍后再试。"
            chat_history.extend([
                {"role": "user", "content": user_input},
                {"role": "assistant", "content": error_response}
            ])

            # 更新会话状态
            self.conversation_sessions[session_id] = {
                "chat_history": chat_history,
                "last_activity": time.time()
            }

            return {
                "response": error_response,
                "chat_history": chat_history,
                "current_intent": Intent.UNKNOWN.value,
                "tool_used": False,
                "session_id": session_id
            }

# 全局应用状态
class AppState:
    def __init__(self):
        self.customer_service = EnhancedCustomerService()
        self.plugin_manager = PluginManager()
        self.model_manager = ModelManager()
        self.start_time = datetime.now()
        self.request_count = 0

app_state = AppState()

# 创建FastAPI应用
app = FastAPI(
    title="智能客服系统",
    description="支持多轮对话和工具调用的智能客服系统",
    version="1.0.0"
)

# 健康检查端点
@app.get("/health")
async def health_check():
    """健康检查接口"""
    current_time = datetime.now()
    uptime = current_time - app_state.start_time

    # 检查关键组件状态
    components_healthy = True
    try:
        # 测试模型连接
        test_model = app_state.model_manager.get_model()
        test_response = test_model.invoke("测试")
        model_healthy = True
    except Exception as e:
        model_healthy = False
        components_healthy = False
        logger.error(f"模型健康检查失败: {str(e)}")

    health_status = {
        "status": "healthy" if components_healthy else "unhealthy",
        "timestamp": current_time.isoformat(),
        "uptime_seconds": uptime.total_seconds(),
        "version": "1.0.0",
        "components": {
            "model": "healthy" if model_healthy else "unhealthy",
            "plugins": "healthy",
            "graph": "healthy"
        },
        "metrics": {
            "total_requests": app_state.request_count,
            "active_sessions": len(app_state.customer_service.conversation_sessions)
        }
    }

    status_code = 200 if components_healthy else 503
    return JSONResponse(content=health_status, status_code=status_code)

# 对话端点
@app.post("/chat")
async def chat_endpoint(request: Dict[str, Any]):
    """处理用户对话"""
    app_state.request_count += 1

    user_input = request.get("message", "")
    session_id = request.get("session_id")

    if not user_input:
        raise HTTPException(status_code=400, detail="消息内容不能为空")

    try:
        result = app_state.customer_service.process_message(user_input, session_id)
        return {
            "success": True,
            "response": result["response"],
            "session_id": result["session_id"],
            "current_intent": result["current_intent"],
            "tool_used": result["tool_used"]
        }
    except Exception as e:
        logger.error(f"对话处理失败: {str(e)}")
        raise HTTPException(status_code=500, detail="内部服务器错误")

# 模型管理端点
@app.post("/model/update")
async def update_model(request: Dict[str, Any]):
    """更新模型配置"""
    new_model = request.get("model_name")
    config = request.get("config")

    if not new_model:
        raise HTTPException(status_code=400, detail="模型名称不能为空")

    success = app_state.model_manager.update_model(new_model, config)

    if success:
        return {
            "success": True,
            "message": f"模型已更新为 {new_model}",
            "current_model": app_state.model_manager.current_model
        }
    else:
        raise HTTPException(status_code=500, detail="模型更新失败")

@app.get("/model/info")
async def get_model_info():
    """获取模型信息"""
    return app_state.model_manager.get_model_info()

# 插件管理端点
@app.post("/plugin/reload")
async def reload_plugin(request: Dict[str, Any]):
    """重新加载插件"""
    plugin_name = request.get("plugin_name")

    if not plugin_name:
        raise HTTPException(status_code=400, detail="插件名称不能为空")

    success = app_state.plugin_manager.reload_plugin(plugin_name)

    if success:
        return {
            "success": True,
            "message": f"插件 {plugin_name} 重新加载成功",
            "new_version": app_state.plugin_manager.plugin_versions[plugin_name]
        }
    else:
        raise HTTPException(status_code=500, detail="插件重新加载失败")

@app.get("/plugin/info")
async def get_plugin_info():
    """获取插件信息"""
    return app_state.plugin_manager.get_plugin_info()

# 会话管理端点
@app.get("/sessions")
async def get_sessions():
    """获取活跃会话列表"""
    sessions = {}
    for session_id, session_data in app_state.customer_service.conversation_sessions.items():
        sessions[session_id] = {
            "last_activity": session_data["last_activity"],
            "message_count": len(session_data["chat_history"]) // 2
        }

    return {
        "active_sessions": len(sessions),
        "sessions": sessions
    }

@app.delete("/sessions/{session_id}")
async def delete_session(session_id: str):
    """删除指定会话"""
    if session_id in app_state.customer_service.conversation_sessions:
        del app_state.customer_service.conversation_sessions[session_id]
        return {"success": True, "message": f"会话 {session_id} 已删除"}
    else:
        raise HTTPException(status_code=404, detail="会话不存在")

# 自动化测试
def test_invoice_plugin():
    """测试发票开具插件的功能正确性"""
    invoice_tools = InvoiceTools()

    # 测试正常开具发票
    result = invoice_tools.create_invoice("ORD123456", "测试公司", "123456789012345")
    assert result["success"] == True
    assert "invoice_id" in result["data"]
    assert result["data"]["invoice_title"] == "测试公司"

    # 测试无税号开具发票
    result_no_tax = invoice_tools.create_invoice("ORD789012", "个人用户")
    assert result_no_tax["success"] == True

    # 测试发票查询
    result_query = invoice_tools.query_invoice("INV123456")
    assert result_query["success"] == True
    assert result_query["data"]["status"] == "已开具"

    # 测试查询不存在的发票
    result_invalid = invoice_tools.query_invoice("INV000000")
    assert result_invalid["success"] == False

    print("✅ 发票插件测试通过")

def test_hot_reload():
    """测试热更新后旧会话不受影响"""
    # 创建初始会话
    service = EnhancedCustomerService()
    session_id = "test_session"

    # 发送初始消息
    result1 = service.process_message("我要开发票", session_id)
    assert "订单号" in result1["response"]  # 应该询问订单号

    # 模拟热更新（这里只是测试会话保持，实际热更新需要更复杂的逻辑）
    old_chat_history = result1["chat_history"].copy()

    # 继续对话
    result2 = service.process_message("ORD123456", session_id)
    assert "订单号" in result2["response"]  # 应该询问发票抬头

    # 验证会话历史保持连续
    assert len(result2["chat_history"]) > len(old_chat_history)
    assert result2["chat_history"][0] == old_chat_history[0]  # 历史消息应该保持一致

    print("✅ 热更新会话测试通过")

def run_tests():
    """运行所有测试"""
    print("开始运行自动化测试...")

    try:
        test_invoice_plugin()
        test_hot_reload()
        print("🎉 所有测试通过！")
        return True
    except Exception as e:
        print(f"❌ 测试失败: {str(e)}")
        return False

# 启动函数
def start_server(host: str = "127.0.0.1", port: int = 8000):
    """启动FastAPI服务器"""
    logger.info(f"启动智能客服系统服务器: http://{host}:{port}")

    # 运行自动化测试
    if run_tests():
        logger.info("自动化测试通过，启动服务器")
    else:
        logger.warning("自动化测试失败，但继续启动服务器")

    uvicorn.run(app, host=host, port=port)

# 测试函数
def main():
    """测试增强版系统"""
    service = EnhancedCustomerService()

    # 分步测试用例
    test_cases = [
        # 订单查询流程
        "我想查询我的订单",
        "ORD123456",

        # 发票开具流程
        "我要开发票",
        "ORD789012",
        "科技有限公司",
        "123456789012345"
    ]

    print("=== 增强版客服系统测试 ===\n")

    chat_history = []
    session_id = None

    for i, user_input in enumerate(test_cases):
        print(f"用户 [{i+1}]: {user_input}")
        result = service.process_message(user_input, session_id, chat_history)
        chat_history = result["chat_history"]
        session_id = result["session_id"]

        print(f"客服: {result['response']}")
        print(f"意图: {result['current_intent']}")
        if result['tool_used']:
            print("[工具已调用]")
        print("-" * 50)

if __name__ == "__main__":
    # 如果是直接运行，启动服务器
    start_server()

整合项目

根据阶段三出来的代码，我们可以拆分出几小块的python文件，这样结构会更清晰，便于维护。
我们按照以下结构组织：

项目结构规划

smart_customer_service/
├── main.py                 # 主入口文件
├── config.py              # 配置和常量
├── models.py              # 数据模型和状态定义
├── tools_service.py               # 工具函数和插件
├── managers.py            # 管理类（插件管理器、模型管理器）
├── graph_nodes.py         # LangGraph节点函数
├── customer_service.py    # 客服系统核心类
├── api.py                 # FastAPI路由和端点
└── tests_service.py               # 自动化测试

1. config.py - 配置和常量

import os
from dotenv import load_dotenv
import logging
from logging.handlers import RotatingFileHandler

# 加载环境变量
load_dotenv()

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        RotatingFileHandler('customer_service.log', maxBytes=10485760, backupCount=5),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# API配置
API_CONFIG = {
    "host": "127.0.0.1",
    "port": 8000,
    "debug": True
}

# 会话配置
SESSION_CONFIG = {
    "timeout": 3600,  # 会话超时时间（秒）
    "max_steps": 10,   # 最大对话步数
    "recursion_limit": 50  # 图递归限制
}

# 模型配置
MODEL_CONFIG = {
    "default_model": "gpt-3.5-turbo",
    "models": {
        "gpt-3.5-turbo": {
            "temperature": 0.3,
            "max_tokens": 1000
        },
        "gpt-4": {
            "temperature": 0.3,
            "max_tokens": 1000
        }
    }
}

2. models.py - 数据模型和状态定义

from typing import Dict, Any, List, TypedDict
from enum import Enum
from datetime import datetime

# 定义意图枚举
class Intent(Enum):
    QUERY_ORDER = "query_order"
    APPLY_REFUND = "apply_refund"
    CREATE_INVOICE = "create_invoice"
    GENERAL_QUERY = "general_query"
    UNKNOWN = "unknown"

# 定义每个意图所需的信息
REQUIRED_INFO = {
    Intent.QUERY_ORDER.value: ["order_id"],
    Intent.APPLY_REFUND.value: ["order_id", "reason"],
    Intent.CREATE_INVOICE.value: ["order_id", "invoice_title", "tax_number"]
}

# 定义状态类型
class CustomerServiceState(TypedDict):
    user_input: str
    chat_history: List[Dict[str, str]]
    current_intent: str
    missing_info: Dict[str, Any]
    tool_result: Any
    response: str
    conversation_finished: bool
    step_count: int
    max_steps: int
    information_collected: bool
    information_extracted: bool

# 健康检查响应模型
class HealthStatus:
    def __init__(self, status: str, components: Dict, metrics: Dict):
        self.status = status
        self.timestamp = datetime.now().isoformat()
        self.components = components
        self.metrics = metrics
        self.version = "1.0.0"

# API响应模型
class APIResponse:
    def __init__(self, success: bool, message: str = "", data: Any = None):
        self.success = success
        self.message = message
        self.data = data
        self.timestamp = datetime.now().isoformat()

3. tools_service.py - 工具函数和插件

import sys
import os
from tools_service import InvoiceTools
from customer_service import EnhancedCustomerService

def test_invoice_plugin():
    """测试发票开具插件的功能正确性"""
    invoice_tools = InvoiceTools()

    # 测试正常开具发票
    result = invoice_tools.create_invoice("ORD123456", "测试公司", "123456789012345")
    assert result["success"] == True
    assert "invoice_id" in result["data"]
    assert result["data"]["invoice_title"] == "测试公司"

    # 测试无税号开具发票
    result_no_tax = invoice_tools.create_invoice("ORD789012", "个人用户")
    assert result_no_tax["success"] == True

    # 测试发票查询
    result_query = invoice_tools.query_invoice("INV123456")
    assert result_query["success"] == True
    assert result_query["data"]["status"] == "已开具"

    # 测试查询不存在的发票
    result_invalid = invoice_tools.query_invoice("INV000000")
    assert result_invalid["success"] == False

    print("✅ 发票插件测试通过")

def test_hot_reload():
    """测试热更新后旧会话不受影响"""
    from managers import PluginManager, ModelManager

    # 创建初始会话
    service = EnhancedCustomerService(ModelManager(), PluginManager())
    session_id = "test_session"

    # 发送初始消息
    result1 = service.process_message("我要开发票", session_id)
    assert "订单号" in result1["response"]  # 应该询问订单号

    # 模拟热更新（这里只是测试会话保持，实际热更新需要更复杂的逻辑）
    old_chat_history = result1["chat_history"].copy()

    # 继续对话
    result2 = service.process_message("ORD123456", session_id)
    assert "发票抬头" in result2["response"]  # 应该询问发票抬头

    # 验证会话历史保持连续
    assert len(result2["chat_history"]) > len(old_chat_history)
    assert result2["chat_history"][0] == old_chat_history[0]  # 历史消息应该保持一致

    print("✅ 热更新会话测试通过")

def run_tests():
    """运行所有测试"""
    print("开始运行自动化测试...")

    try:
        test_invoice_plugin()
        test_hot_reload()
        print("🎉 所有测试通过！")
        return True
    except Exception as e:
        print(f"❌ 测试失败: {str(e)}")
        return False

if __name__ == "__main__":
    success = run_tests()
    sys.exit(0 if success else 1)

4. managers.py - 管理类

import os
import threading
import importlib
import sys
from datetime import datetime
from typing import Dict, Any
from langchain_openai import ChatOpenAI
from tools_service import OrderTools, InvoiceTools
from config import logger, MODEL_CONFIG

class PluginManager:
    """插件管理器，支持热重载"""

    def __init__(self):
        self.plugins = {}
        self.plugin_versions = {}
        self.load_plugins()

    def load_plugins(self):
        """加载所有插件"""
        self.plugins = {
            "order_tools": OrderTools(),
            "invoice_tools": InvoiceTools()
        }
        self.plugin_versions = {
            "order_tools": "1.0.0",
            "invoice_tools": "1.0.0"
        }
        logger.info("所有插件已加载")

    def reload_plugin(self, plugin_name: str):
        """重新加载指定插件"""
        try:
            if plugin_name == "order_tools":
                # 重新加载OrderTools类
                importlib.reload(sys.modules[__name__])
                self.plugins[plugin_name] = OrderTools()
            elif plugin_name == "invoice_tools":
                # 重新加载InvoiceTools类
                importlib.reload(sys.modules[__name__])
                self.plugins[plugin_name] = InvoiceTools()

            # 更新版本号
            current_version = self.plugin_versions[plugin_name]
            major, minor, patch = current_version.split('.')
            new_version = f"{major}.{minor}.{int(patch) + 1}"
            self.plugin_versions[plugin_name] = new_version

            logger.info(f"插件 {plugin_name} 已重新加载，版本: {new_version}")
            return True
        except Exception as e:
            logger.error(f"重新加载插件 {plugin_name} 失败: {str(e)}")
            return False

    def get_plugin(self, plugin_name: str):
        """获取插件实例"""
        return self.plugins.get(plugin_name)

    def get_plugin_info(self):
        """获取所有插件信息"""
        return {
            name: {
                "version": version,
                "status": "loaded"
            } for name, version in self.plugin_versions.items()
        }

class ModelManager:
    """模型管理器，支持热更新"""

    def __init__(self):
        self.current_model = MODEL_CONFIG["default_model"]
        self.model_config = MODEL_CONFIG["models"]
        self.model_history = []
        self.update_lock = threading.Lock()

    def get_model(self, model_name: str = None):
        """获取当前模型实例"""
        if model_name is None:
            model_name = self.current_model

        config = self.model_config.get(model_name, self.model_config[MODEL_CONFIG["default_model"]])

        return ChatOpenAI(
            model_name=model_name,
            temperature=config["temperature"],
            max_tokens=config["max_tokens"],
            openai_api_key=os.getenv("OPENAI_API_KEY")
        )

    def update_model(self, new_model: str, config: Dict[str, Any] = None):
        """更新模型配置"""
        with self.update_lock:
            old_model = self.current_model
            self.current_model = new_model

            if config:
                self.model_config[new_model] = config

            # 记录模型更新历史
            self.model_history.append({
                "timestamp": datetime.now().isoformat(),
                "from": old_model,
                "to": new_model,
                "config": self.model_config[new_model]
            })

            logger.info(f"模型已更新: {old_model} -> {new_model}")
            return True

    def get_model_info(self):
        """获取模型信息"""
        return {
            "current_model": self.current_model,
            "config": self.model_config[self.current_model],
            "available_models": list(self.model_config.keys()),
            "update_history": self.model_history[-5:]
        }

5. graph_nodes.py - LangGraph节点函数

import re
from datetime import datetime, timedelta
from langchain.prompts import PromptTemplate
from models import CustomerServiceState, Intent, REQUIRED_INFO
from config import logger

class GraphNodes:
    """LangGraph节点函数集合"""

    def __init__(self, model_manager, plugin_manager):
        self.model_manager = model_manager
        self.plugin_manager = plugin_manager

    def intent_recognition_node(self, state: CustomerServiceState) -> CustomerServiceState:
        """识别用户意图"""
        state["step_count"] += 1

        # 检查步骤限制
        if state["step_count"] > state["max_steps"]:
            state["response"] = "抱歉，对话轮次过多，请重新开始咨询。"
            state["conversation_finished"] = True
            return state

        intent_prompt = PromptTemplate(
            template="""根据用户输入识别意图。可选意图：
- query_order: 用户想要查询订单状态
- apply_refund: 用户想要申请退款
- create_invoice: 用户想要开具发票
- general_query: 一般咨询问题

用户输入: {user_input}
对话历史: {chat_history}

请只返回意图名称，不要返回其他内容。""",
            input_variables=["user_input", "chat_history"]
        )

        llm = self.model_manager.get_model()

        chat_history_text = "\n".join([
            f"{msg['role']}: {msg['content']}" for msg in state["chat_history"][-3:]
        ]) if state["chat_history"] else "无"

        try:
            intent_result = llm.invoke(intent_prompt.format(
                user_input=state["user_input"],
                chat_history=chat_history_text
            ))

            intent_str = intent_result.content.strip().lower()
            state["current_intent"] = intent_str

        except Exception:
            # 如果LLM调用失败，使用基于关键词的后备方案
            user_input_lower = state["user_input"].lower()
            if any(keyword in user_input_lower for keyword in ["查询", "订单", "查订单"]):
                state["current_intent"] = Intent.QUERY_ORDER.value
            elif any(keyword in user_input_lower for keyword in ["退款", "退货"]):
                state["current_intent"] = Intent.APPLY_REFUND.value
            elif any(keyword in user_input_lower for keyword in ["发票", "开票", "发票开具"]):
                state["current_intent"] = Intent.CREATE_INVOICE.value
            else:
                state["current_intent"] = Intent.GENERAL_QUERY.value

        return state

    def information_collection_node(self, state: CustomerServiceState) -> CustomerServiceState:
        """收集执行工具所需的信息"""
        state["step_count"] += 1

        # 确保missing_info存在
        if "missing_info" not in state:
            state["missing_info"] = {}

        # 获取当前意图所需的信息字段
        required_fields = REQUIRED_INFO.get(state["current_intent"], [])

        # 确保missing_info包含所有必需字段
        for field in required_fields:
            if field not in state["missing_info"]:
                state["missing_info"][field] = None

        # 根据意图和缺失信息生成相应的询问
        if state["current_intent"] == Intent.QUERY_ORDER.value:
            if state["missing_info"]["order_id"] is None:
                state["response"] = "请问您的订单号是多少？"
                state["information_extracted"] = False
            else:
                state["information_collected"] = True
                return state

        elif state["current_intent"] == Intent.APPLY_REFUND.value:
            if state["missing_info"]["order_id"] is None:
                state["response"] = "请问您要申请退款的订单号是多少？"
                state["information_extracted"] = False
            elif state["missing_info"]["reason"] is None:
                state["response"] = "请问您申请退款的原因是什么？"
                state["information_extracted"] = False
            else:
                state["information_collected"] = True
                return state

        elif state["current_intent"] == Intent.CREATE_INVOICE.value:
            if state["missing_info"]["order_id"] is None:
                state["response"] = "请问您要为哪个订单开具发票？请提供订单号。"
                state["information_extracted"] = False
            elif state["missing_info"]["invoice_title"] is None:
                state["response"] = "请问发票抬头是什么？"
                state["information_extracted"] = False
            elif state["missing_info"]["tax_number"] is None:
                state["response"] = "请问纳税人识别号是什么？（如不需要可回复'无'）"
                state["information_extracted"] = False
            else:
                state["information_collected"] = True
                return state

        return state

    def information_extraction_node(self, state: CustomerServiceState) -> CustomerServiceState:
        """从用户输入中提取所需信息"""
        state["step_count"] += 1

        # 确保missing_info存在
        if "missing_info" not in state:
            state["missing_info"] = {}

        if state["current_intent"] in [Intent.QUERY_ORDER.value, Intent.APPLY_REFUND.value, Intent.CREATE_INVOICE.value]:
            # 获取当前意图所需的信息字段
            required_fields = REQUIRED_INFO.get(state["current_intent"], [])

            # 确保所有必需字段都在missing_info中
            for field in required_fields:
                if field not in state["missing_info"]:
                    state["missing_info"][field] = None

            # 提取订单号
            order_match = re.search(r'[A-Za-z]{3}\d{6,}', state["user_input"])
            if order_match and state["missing_info"].get("order_id") is None:
                state["missing_info"]["order_id"] = order_match.group().upper()

            # 提取发票抬头和税号
            if state["current_intent"] == Intent.CREATE_INVOICE.value:
                # 简单的发票抬头提取（假设用户直接提供了抬头）
                if state["missing_info"].get("invoice_title") is None and len(state["user_input"]) > 2:
                    # 如果不是订单号和税号，且长度合适，认为是发票抬头
                    if not order_match and not re.search(r'\d{15,20}', state["user_input"]):
                        state["missing_info"]["invoice_title"] = state["user_input"]

                # 提取税号（15-20位数字）
                tax_match = re.search(r'\d{15,20}', state["user_input"])
                if tax_match and state["missing_info"].get("tax_number") is None:
                    state["missing_info"]["tax_number"] = tax_match.group()

                # 如果用户说"无"或"不需要"，设置税号为空
                if "无" in state["user_input"] or "不需要" in state["user_input"]:
                    state["missing_info"]["tax_number"] = ""

            # 如果是退款申请且还没有原因，尝试提取原因
            if (state["current_intent"] == Intent.APPLY_REFUND.value and
                    state["missing_info"].get("order_id") and
                    state["missing_info"].get("reason") is None):

                # 简单的关键词提取
                reason_keywords = {
                    "质量": "商品质量问题",
                    "损坏": "商品损坏",
                    "不满意": "对商品不满意",
                    "错误": "订单信息错误",
                    "不想要": "不再需要此商品",
                    "不喜欢": "对商品不喜欢"
                }

                for keyword, reason in reason_keywords.items():
                    if keyword in state["user_input"]:
                        state["missing_info"]["reason"] = reason
                        break

                if not state["missing_info"].get("reason"):
                    # 如果没有匹配到关键词，使用用户原始输入作为原因
                    state["missing_info"]["reason"] = state["user_input"]

        state["information_extracted"] = True
        return state

    def tool_call_node(self, state: CustomerServiceState) -> CustomerServiceState:
        """调用相应的工具函数"""
        state["step_count"] += 1

        # 确保missing_info存在
        if "missing_info" not in state:
            state["missing_info"] = {}

        if state["current_intent"] == Intent.QUERY_ORDER.value and state["missing_info"].get("order_id"):
            order_tools = self.plugin_manager.get_plugin("order_tools")
            result = order_tools.query_order(state["missing_info"]["order_id"])
            state["tool_result"] = result

            if result["success"]:
                order = result["data"]
                state["response"] = f"订单状态: {order['status']}\n"
                state["response"] += f"商品: {order['product']}\n"
                state["response"] += f"下单日期: {order['order_date']}\n"
                state["response"] += f"预计送达: {order['estimated_delivery']}"

                if order.get("tracking_number"):
                    state["response"] += f"\n快递公司: {order['shipping_company']}"
                    state["response"] += f"\n运单号: {order['tracking_number']}"
            else:
                state["response"] = f"抱歉，{result['error']}"

            state["conversation_finished"] = True

        elif state["current_intent"] == Intent.APPLY_REFUND.value and state["missing_info"].get("order_id") and state["missing_info"].get("reason"):
            order_tools = self.plugin_manager.get_plugin("order_tools")
            result = order_tools.apply_refund(
                state["missing_info"]["order_id"],
                state["missing_info"]["reason"]
            )
            state["tool_result"] = result
            state["response"] = result["message"]
            state["conversation_finished"] = True

        elif state["current_intent"] == Intent.CREATE_INVOICE.value and state["missing_info"].get("order_id") and state["missing_info"].get("invoice_title"):
            invoice_tools = self.plugin_manager.get_plugin("invoice_tools")
            result = invoice_tools.create_invoice(
                state["missing_info"]["order_id"],
                state["missing_info"]["invoice_title"],
                state["missing_info"].get("tax_number")
            )
            state["tool_result"] = result
            state["response"] = result["message"]
            state["conversation_finished"] = True

        return state

    def general_response_node(self, state: CustomerServiceState) -> CustomerServiceState:
        """处理一般性查询"""
        state["step_count"] += 1

        if state["current_intent"] == Intent.GENERAL_QUERY.value:
            prompt_template = PromptTemplate(
                input_variables=["time_context", "user_input", "chat_history"],
                template="""你是一个智能客服助手，需要准确理解用户的时间相关查询，并结合当前时间进行回答。

{time_context}

对话历史:
{chat_history}

用户输入: {user_input}

请提供有帮助的回复。

客服回复:"""
            )

            llm = self.model_manager.get_model()

            # 获取时间上下文
            now = datetime.now()
            current_time = now.strftime("%Y年%m月%d日 %H:%M:%S")
            yesterday = (now - timedelta(days=1)).strftime("%Y年%m月%d日")
            tomorrow = (now + timedelta(days=1)).strftime("%Y年%m月%d日")

            time_context = f"""
当前时间: {current_time}
相关日期:
- 昨天: {yesterday}
- 今天: {now.strftime('%Y年%m月%d日')}
- 明天: {tomorrow}
"""

            # 获取对话历史
            chat_history_text = "\n".join([
                f"{msg['role']}: {msg['content']}" for msg in state["chat_history"][-3:]
            ]) if state["chat_history"] else "无"

            try:
                result = llm.invoke(prompt_template.format(
                    time_context=time_context,
                    user_input=state["user_input"],
                    chat_history=chat_history_text
                ))

                state["response"] = result.content
            except Exception:
                state["response"] = "抱歉，我现在无法处理您的请求，请稍后再试。"

            state["conversation_finished"] = True

        return state

    def update_history_node(self, state: CustomerServiceState) -> CustomerServiceState:
        """更新对话历史"""
        # 添加用户消息到历史
        state["chat_history"].append({
            "role": "user",
            "content": state["user_input"]
        })

        # 添加AI响应到历史
        if state["response"]:
            state["chat_history"].append({
                "role": "assistant",
                "content": state["response"]
            })

        return state

    def route_conversation(self, state: CustomerServiceState) -> str:
        """决定下一步执行哪个节点"""

        # 如果已经生成响应，更新历史后结束
        if state.get("response"):
            return "update_history"

        # 如果对话已结束，更新历史后结束
        if state.get("conversation_finished", False):
            return "update_history"

        # 检查步骤限制
        if state.get("step_count", 0) > state.get("max_steps", 20):
            state["response"] = "为了更好的服务体验，本次对话将结束。如有需要请重新咨询。"
            return "update_history"

        current_intent = state.get("current_intent", Intent.UNKNOWN.value)

        # 一般查询直接回复
        if current_intent == Intent.GENERAL_QUERY.value:
            return "general_response"

        # 工具类意图的处理流程
        # 获取当前意图所需的信息字段
        required_fields = REQUIRED_INFO.get(current_intent, [])

        # 确保missing_info存在
        if "missing_info" not in state:
            state["missing_info"] = {}

        # 确保missing_info包含所有必需字段
        for field in required_fields:
            if field not in state["missing_info"]:
                state["missing_info"][field] = None

        # 检查是否所有必需信息都已收集
        all_info_collected = all(state["missing_info"].get(field) is not None for field in required_fields)

        if all_info_collected:
            return "tool_call"
        else:
            # 如果有用户输入但还没有响应，尝试提取信息
            if not state.get("information_extracted", False):
                return "information_extraction"
            elif not state.get("information_collected", False):
                return "information_collection"
            else:
                return "information_collection"

6. customer_service.py - 客服系统核心类

import time
from typing import Dict, Any, List
from langgraph.graph import StateGraph, END
from models import CustomerServiceState, Intent
from config import logger, SESSION_CONFIG
from graph_nodes import GraphNodes

class EnhancedCustomerService:
    """支持多轮对话和工具调用的客服系统"""

    def __init__(self, model_manager, plugin_manager, model_name: str = "gpt-3.5-turbo"):
        self.model_name = model_name
        self.model_manager = model_manager
        self.plugin_manager = plugin_manager
        self.graph_nodes = GraphNodes(model_manager, plugin_manager)
        self.graph = self._create_customer_service_graph()
        self.conversation_sessions = {}  # 存储会话状态
        self.session_timeout = SESSION_CONFIG["timeout"]

    def _create_customer_service_graph(self):
        """创建客服对话图"""
        workflow = StateGraph(CustomerServiceState)

        # 添加节点
        workflow.add_node("intent_recognition", self.graph_nodes.intent_recognition_node)
        workflow.add_node("information_collection", self.graph_nodes.information_collection_node)
        workflow.add_node("information_extraction", self.graph_nodes.information_extraction_node)
        workflow.add_node("tool_call", self.graph_nodes.tool_call_node)
        workflow.add_node("general_response", self.graph_nodes.general_response_node)
        workflow.add_node("update_history", self.graph_nodes.update_history_node)

        # 设置入口点
        workflow.set_entry_point("intent_recognition")

        # 条件边
        workflow.add_conditional_edges(
            "intent_recognition",
            self.graph_nodes.route_conversation,
            {
                "information_collection": "information_collection",
                "information_extraction": "information_extraction",
                "tool_call": "tool_call",
                "general_response": "general_response",
                "update_history": "update_history"
            }
        )

        workflow.add_conditional_edges(
            "information_collection",
            self.graph_nodes.route_conversation,
            {
                "information_collection": "information_collection",
                "information_extraction": "information_extraction",
                "tool_call": "tool_call",
                "general_response": "general_response",
                "update_history": "update_history"
            }
        )

        workflow.add_conditional_edges(
            "information_extraction",
            self.graph_nodes.route_conversation,
            {
                "information_collection": "information_collection",
                "tool_call": "tool_call",
                "general_response": "general_response",
                "update_history": "update_history"
            }
        )

        workflow.add_conditional_edges(
            "tool_call",
            self.graph_nodes.route_conversation,
            {
                "update_history": "update_history"
            }
        )

        workflow.add_conditional_edges(
            "general_response",
            self.graph_nodes.route_conversation,
            {
                "update_history": "update_history"
            }
        )

        # 添加最终边
        workflow.add_edge("update_history", END)

        return workflow.compile()

    def _cleanup_sessions(self):
        """清理过期的会话"""
        current_time = time.time()
        expired_sessions = []

        for session_id, session_data in self.conversation_sessions.items():
            if current_time - session_data.get("last_activity", 0) > self.session_timeout:
                expired_sessions.append(session_id)

        for session_id in expired_sessions:
            del self.conversation_sessions[session_id]
            logger.info(f"清理过期会话: {session_id}")

    def process_message(self, user_input: str, session_id: str = None, chat_history: List[Dict[str, str]] = None) -> Dict[str, Any]:
        """处理用户消息"""

        # 清理过期会话
        self._cleanup_sessions()

        if session_id is None:
            session_id = f"session_{int(time.time())}_{id(self)}"

        if chat_history is None:
            if session_id in self.conversation_sessions:
                chat_history = self.conversation_sessions[session_id]["chat_history"]
            else:
                chat_history = []

        # 准备初始状态
        initial_state = {
            "user_input": user_input,
            "chat_history": chat_history,
            "current_intent": Intent.UNKNOWN.value,
            "missing_info": {},
            "tool_result": None,
            "response": "",
            "conversation_finished": False,
            "step_count": 0,
            "max_steps": SESSION_CONFIG["max_steps"],
            "information_collected": False,
            "information_extracted": False
        }

        try:
            # 执行图，显式设置递归限制
            config = {"recursion_limit": SESSION_CONFIG["recursion_limit"]}
            final_state = self.graph.invoke(initial_state, config=config)

            # 更新会话状态
            self.conversation_sessions[session_id] = {
                "chat_history": final_state["chat_history"],
                "last_activity": time.time()
            }

            return {
                "response": final_state["response"],
                "chat_history": final_state["chat_history"],
                "current_intent": final_state["current_intent"],
                "tool_used": final_state.get("tool_result") is not None,
                "session_id": session_id
            }
        except Exception as e:
            # 错误处理
            logger.error(f"处理消息时出错: {str(e)}")
            error_response = "抱歉，系统暂时无法处理您的请求，请稍后再试。"
            chat_history.extend([
                {"role": "user", "content": user_input},
                {"role": "assistant", "content": error_response}
            ])

            # 更新会话状态
            self.conversation_sessions[session_id] = {
                "chat_history": chat_history,
                "last_activity": time.time()
            }

            return {
                "response": error_response,
                "chat_history": chat_history,
                "current_intent": Intent.UNKNOWN.value,
                "tool_used": False,
                "session_id": session_id
            }

# 全局应用状态
class AppState:
    def __init__(self):
        from managers import PluginManager, ModelManager
        self.plugin_manager = PluginManager()
        self.model_manager = ModelManager()
        self.customer_service = EnhancedCustomerService(self.model_manager, self.plugin_manager)
        self.start_time = time.time()
        self.request_count = 0

7. api.py - FastAPI路由和端点

from datetime import datetime
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from config import logger, API_CONFIG
from customer_service import AppState
from models import HealthStatus

# 创建全局应用状态
app_state = AppState()

# 创建FastAPI应用
app = FastAPI(
    title="智能客服系统",
    description="支持多轮对话和工具调用的智能客服系统",
    version="1.0.0"
)

# 健康检查端点
@app.get("/health")
async def health_check():
    """健康检查接口"""
    current_time = datetime.now()
    uptime = current_time - datetime.fromtimestamp(app_state.start_time)

    # 检查关键组件状态
    components_healthy = True
    try:
        # 测试模型连接
        test_model = app_state.model_manager.get_model()
        test_response = test_model.invoke("测试")
        model_healthy = True
    except Exception as e:
        model_healthy = False
        components_healthy = False
        logger.error(f"模型健康检查失败: {str(e)}")

    health_status = HealthStatus(
        status="healthy" if components_healthy else "unhealthy",
        components={
            "model": "healthy" if model_healthy else "unhealthy",
            "plugins": "healthy",
            "graph": "healthy"
        },
        metrics={
            "total_requests": app_state.request_count,
            "active_sessions": len(app_state.customer_service.conversation_sessions),
            "uptime_seconds": uptime.total_seconds()
        }
    )

    status_code = 200 if components_healthy else 503
    return JSONResponse(content=health_status.__dict__, status_code=status_code)

# 对话端点
@app.post("/chat")
async def chat_endpoint(request: dict):
    """处理用户对话"""
    app_state.request_count += 1

    user_input = request.get("message", "")
    session_id = request.get("session_id")

    if not user_input:
        raise HTTPException(status_code=400, detail="消息内容不能为空")

    try:
        result = app_state.customer_service.process_message(user_input, session_id)
        return {
            "success": True,
            "response": result["response"],
            "session_id": result["session_id"],
            "current_intent": result["current_intent"],
            "tool_used": result["tool_used"]
        }
    except Exception as e:
        logger.error(f"对话处理失败: {str(e)}")
        raise HTTPException(status_code=500, detail="内部服务器错误")

# 模型管理端点
@app.post("/model/update")
async def update_model(request: dict):
    """更新模型配置"""
    new_model = request.get("model_name")
    config = request.get("config")

    if not new_model:
        raise HTTPException(status_code=400, detail="模型名称不能为空")

    success = app_state.model_manager.update_model(new_model, config)

    if success:
        return {
            "success": True,
            "message": f"模型已更新为 {new_model}",
            "current_model": app_state.model_manager.current_model
        }
    else:
        raise HTTPException(status_code=500, detail="模型更新失败")

@app.get("/model/info")
async def get_model_info():
    """获取模型信息"""
    return app_state.model_manager.get_model_info()

# 插件管理端点
@app.post("/plugin/reload")
async def reload_plugin(request: dict):
    """重新加载插件"""
    plugin_name = request.get("plugin_name")

    if not plugin_name:
        raise HTTPException(status_code=400, detail="插件名称不能为空")

    success = app_state.plugin_manager.reload_plugin(plugin_name)

    if success:
        return {
            "success": True,
            "message": f"插件 {plugin_name} 重新加载成功",
            "new_version": app_state.plugin_manager.plugin_versions[plugin_name]
        }
    else:
        raise HTTPException(status_code=500, detail="插件重新加载失败")

@app.get("/plugin/info")
async def get_plugin_info():
    """获取插件信息"""
    return app_state.plugin_manager.get_plugin_info()

# 会话管理端点
@app.get("/sessions")
async def get_sessions():
    """获取活跃会话列表"""
    sessions = {}
    for session_id, session_data in app_state.customer_service.conversation_sessions.items():
        sessions[session_id] = {
            "last_activity": session_data["last_activity"],
            "message_count": len(session_data["chat_history"]) // 2
        }

    return {
        "active_sessions": len(sessions),
        "sessions": sessions
    }

@app.delete("/sessions/{session_id}")
async def delete_session(session_id: str):
    """删除指定会话"""
    if session_id in app_state.customer_service.conversation_sessions:
        del app_state.customer_service.conversation_sessions[session_id]
        return {"success": True, "message": f"会话 {session_id} 已删除"}
    else:
        raise HTTPException(status_code=404, detail="会话不存在")

8. tests_service.py - 自动化测试

import sys
import os
from tools_service import InvoiceTools
from customer_service import EnhancedCustomerService

def test_invoice_plugin():
    """测试发票开具插件的功能正确性"""
    invoice_tools = InvoiceTools()

    # 测试正常开具发票
    result = invoice_tools.create_invoice("ORD123456", "测试公司", "123456789012345")
    assert result["success"] == True
    assert "invoice_id" in result["data"]
    assert result["data"]["invoice_title"] == "测试公司"

    # 测试无税号开具发票
    result_no_tax = invoice_tools.create_invoice("ORD789012", "个人用户")
    assert result_no_tax["success"] == True

    # 测试发票查询
    result_query = invoice_tools.query_invoice("INV123456")
    assert result_query["success"] == True
    assert result_query["data"]["status"] == "已开具"

    # 测试查询不存在的发票
    result_invalid = invoice_tools.query_invoice("INV000000")
    assert result_invalid["success"] == False

    print("✅ 发票插件测试通过")

def test_hot_reload():
    """测试热更新后旧会话不受影响"""
    from managers import PluginManager, ModelManager

    # 创建初始会话
    service = EnhancedCustomerService(ModelManager(), PluginManager())
    session_id = "test_session"

    # 发送初始消息
    result1 = service.process_message("我要开发票", session_id)
    assert "订单号" in result1["response"]  # 应该询问订单号

    # 模拟热更新（这里只是测试会话保持，实际热更新需要更复杂的逻辑）
    old_chat_history = result1["chat_history"].copy()

    # 继续对话
    result2 = service.process_message("ORD123456", session_id)
    assert "发票抬头" in result2["response"]  # 应该询问发票抬头

    # 验证会话历史保持连续
    assert len(result2["chat_history"]) > len(old_chat_history)
    assert result2["chat_history"][0] == old_chat_history[0]  # 历史消息应该保持一致

    print("✅ 热更新会话测试通过")

def run_tests():
    """运行所有测试"""
    print("开始运行自动化测试...")

    try:
        test_invoice_plugin()
        test_hot_reload()
        print("🎉 所有测试通过！")
        return True
    except Exception as e:
        print(f"❌ 测试失败: {str(e)}")
        return False

if __name__ == "__main__":
    success = run_tests()
    sys.exit(0 if success else 1)

9. main.py - 主入口文件

import uvicorn
from config import logger, API_CONFIG
from api import app
from test_service import run_tests

def start_server():
    """启动FastAPI服务器"""
    host = API_CONFIG["host"]
    port = API_CONFIG["port"]

    logger.info(f"启动智能客服系统服务器: http://{host}:{port}")

    # 运行自动化测试
    if run_tests():
        logger.info("自动化测试通过，启动服务器")
    else:
        logger.warning("自动化测试失败，但继续启动服务器")

    uvicorn.run(app, host=host, port=port, log_level="info")

if __name__ == "__main__":
    start_server()

运行测试

python main.py

运行结果：

2025-10-16 12:42:43,743 - config - INFO - 所有插件已加载
2025-10-16 12:42:43,769 - config - INFO - 启动智能客服系统服务器: http://127.0.0.1:8000
2025-10-16 12:42:43,769 - config - INFO - 所有插件已加载
开始运行自动化测试...
✅ 发票插件测试通过
2025-10-16 12:42:47,835 - httpx - INFO - HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
2025-10-16 12:42:49,591 - httpx - INFO - HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
✅ 热更新会话测试通过
🎉 所有测试通过！
2025-10-16 12:42:49,608 - config - INFO - 自动化测试通过，启动服务器
INFO:     Started server process [9056]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

API测试：

curl http://localhost:8000/health

运行结果：

{"status":"healthy","timestamp":"2025-10-16T12:43:56.253406","components":{"model":"healthy","plugins":"healthy","graph":"healthy"},"metrics":{"total_requests":0,"active_sessions":0,"uptime_seconds":71.260164},"version":"1.0.0"}