AI应用的进化路线

1. 聊天机器人：

   • 基础问答（如客服FAQ）
   • 示例：回答“营业时间几点？”
   • 局限：仅能处理预设问题

2. AI助手：

   • 执行简单任务（如Siri设闹钟）
   • 示例：“明早8点叫我起床”
   • 局限：功能受限，需明确指令

3. AI Agent：

   • 主动规划+执行复杂任务
   • 示例：自动完成“查机票→比价→订票→通知”全流程
   • 核心能力：记忆、决策、工具调用

Function Calling：AI的“动手能力”

1. 本质是什么？

• 传统LLM：只会“说话”的知识库
• Function Calling：给AI配“工具箱”，使其能操作外部系统

2. 工作流程（运动商品店案例）

3. 关键代码解析

# 1. 定义工具：查商品库存
def query_by_product_name(product_name):
    return sql("SELECT * FROM products WHERE name LIKE '%篮球%'")

# 2. 告诉AI工具用法
tools = [{
    "name": "query_by_product_name",
    "description": "按商品名查库存",
    "parameters": {"product_name": "str"}
}]

# 3. AI自动决策调用
if 用户问库存：
    AI生成参数 → 调用工具 → 返回结果

Function Calling的局限性

1. 单任务局限

• 只能响应单次请求
• 示例：
  • ✅ 用户问“篮球优惠？” → 调用两个工具（查库存+查优惠）
  • ❌ 但无法自动关联“买篮球后推荐护膝”

2. 多步任务需Agent升级

• 需ReAct框架：让AI自主规划步骤（思考→行动→再思考）

AI Agent的不可替代性

能力	Function Calling	AI Agent
单次工具调用	✅	✅
多步骤规划	❌	✅
记忆上下文	❌	✅
主动决策	❌	✅

场景对比：

• Function Calling：导购员用扫码枪查商品信息
• AI Agent：资深导购帮你配全套装备+申请隐藏优惠

为什么2025属于AI Agent？

1. 技术成熟：
   • LLM理解力↑ + 工具生态完善 = Agent可行性↑
2. 商业需求：
   • 企业需要“数字员工”自动化复杂流程（如订票+报销全流程）
3. 场景爆发：
   • 医疗Agent：诊断→预约→保险理赔
   • 电商Agent：选品→比价→下单→售后

核心问题：大模型的“幻觉”与“无力”

比如ChatGPT、通义千问、Llama等大模型。你问它：“特斯拉现在的股价是多少？”

• 老式AI（没有函数调用）：
  • 它可能根据“特斯拉”、“股价”这些词，瞎猜一个：“可能在200美元左右吧？”（这就是“幻觉”）。
  • 或者一个诚实但无力的AI会说：“抱歉，我是语言模型，没有实时数据，不知道现在股价。”（这是事实，但帮不了你）。
• 你的需求： 你想知道此时此刻、真实准确的股价！

函数调用：给AI装上“手脚”和“工具箱”

函数调用让大语言模型（LLM）从一个“只会动嘴皮子的秀才”，变成了一个“能指挥各种专家干活儿的军师”。它知道自己能干什么、不能干什么，遇到干不了的事情，就清晰地告诉开发者“该摇哪个专家、怎么摇”，开发者负责执行**这个专家（函数）**并把结果给它，它再用这个结果给你一个靠谱的回答。函数调用（Function Calling）就是解决这个问题的神奇钥匙！它的本质是：

让AI知道自己能力的边界，并学会“摇人”！

1. AI知道自己“不能”： 当AI识别出你的问题（“特斯拉股价”）需要实时数据或者它本身做不到的事情（比如操作电脑、查数据库、控制设备）时…
2. AI决定“摇人”： AI会说：“等等，这事我干不了，但我认识个‘专家’能搞定！我让他（函数）来帮忙！”
3. 明确告诉你要摇谁、怎么摇： AI会清晰地说出：
   • 摇谁： 要调用哪个函数（比如叫 get_stock_price 的函数）。
   • 怎么摇： 给这个函数提供什么参数（比如 {"stock_symbol": "TSLA"}）。
4. 你（开发者）去“摇人”： 你写的程序收到AI的这个“摇人请求”，就去真正执行那个函数（比如调用一个金融数据API）。
5. 把结果告诉AI： 函数执行完，你拿到真实数据（比如 {"price": 279.24, "currency": "USD"}），把这个结果回传给AI。
6. AI给你最终答案： AI拿到真实数据后，就能组织语言回答你：“截至此刻，特斯拉的股价是 279.24 美元。”

整个过程就像这样：

你： “特斯拉现在的股价是多少？”
AI： (识别需要实时数据) -> (决定调用 `get_stock_price` 函数) -> 输出： `调用函数 get_stock_price(参数：{"stock_symbol": "TSLA"})`
你的程序：(执行函数，调用真实API拿到价格 279.24) -> 把结果 `279.24` 传给AI
AI： (拿到真实数据) -> 组织语言回答你： “截至此刻，特斯拉的股价是 279.24 美元。”

为什么说它重要？解锁大模型的“超能力”

• 突破知识限制： AI不再依赖“死记硬背”的训练数据，能获取最新、实时、特定领域的信息（股票、天气、新闻、数据库记录）。
• 突破能力限制： AI本身只是个“大脑”，不能操作电脑、不能发邮件、不能控制设备。函数调用让它能操作外部世界（发邮件、写文件、开灯关灯、订外卖）。
• 更可靠、更精确： 避免了AI“瞎猜”（幻觉），答案基于真实执行的函数结果。
• 构建复杂应用： 让AI成为复杂系统的“智能调度中心”，串联起各种API和服务（CRM系统、数据库、计算服务）。

开发者需要做什么？（以文章中的文件列表示例）

1. 定义“专家”（函数）： 告诉AI有什么“专家”可用。就像给AI一个“通讯录”。
   • 名字(Name)： list_files (函数名)
   • 技能(Description)： “列出目录中的文件” (告诉AI这函数干嘛的)
   • 需要什么(Parameters)： 告诉AI调用这个函数需要提供什么信息。
     • directory: 需要提供一个字符串，是“目录的绝对路径”。
   • （用JSON描述，如文章中所示）
2. 实现“专家”（函数）： 真正写代码实现这个函数的功能（如文章中用Flask写的 /files API）。
3. 告诉AI“通讯录”： 在提问时，把你定义好的“函数通讯录”（available_functions）一起发给AI。
4. 解析AI的“摇人请求”： AI输出的是文本，需要你的程序去解析这个文本，识别出它想调用哪个函数以及参数是什么。（如文章中用正则表达式提取 {"name":"files", "arguments":{"directory":"/tmp"}}）。
5. 真正“摇人”并拿结果： 根据AI的请求，执行对应的函数（如调用 /files?directory=/tmp API）。
6. 把结果告诉AI： 将函数执行的结果（如 [{'name': 'qq1.html'}]）传回给AI。
7. 让AI回答用户： AI拿到结果后，会生成最终的自然语言回答给用户（如“/tmp目录中的文件有：qq1.html”）。

不同模型的小差异（文章最后提到的）：

• 核心思想都一样：AI输出一个结构化的“函数调用请求”。
• 但不同模型（Llama 3, Qwen, Mistral）包裹这个请求的格式可能不同：
  • Llama 3: <|python_tag|>{... JSON ...}<|eom_id|>
  • Qwen: <|im_start|>assistant<tool_call>{... JSON ...}</tool_call><|im_end|>
  • Mistral (API): 返回一个结构化的对象（tool_calls），更容易解析。
• 你需要做的： 根据你用的模型，写对应的解析代码（如文章中用正则表达式匹配特定标签里的JSON）。这就像不同“专家”有不同的呼叫暗号，你需要听懂它用的暗号。

import json
import os
import sqlite3
import logging
from typing import List, Dict, Any, Optional

from dotenv import find_dotenv, load_dotenv
from openai import OpenAI, OpenAIError

# --- 日志配置 ---
# 1. 创建一个日志记录器，并将日志输出到文件而不是控制台
log_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
log_file = 'chatbot.log'

# 2. 设置文件处理器，使用 'a' 模式追加日志，并指定编码
file_handler = logging.FileHandler(log_file, mode='a', encoding='utf-8')
file_handler.setFormatter(log_formatter)

# 3. 获取根日志记录器，并添加文件处理器
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 4. 清除可能存在的默认控制台处理器，确保日志只输出到文件
if logger.hasHandlers():
    logger.handlers.clear()
logger.addHandler(file_handler)


# --- 日志配置结束 ---


class AppConfig:
    """
    管理应用程序的配置。
    从环境变量加载敏感信息，并定义常量。
    """

    def __init__(self):
        load_dotenv(find_dotenv())
        self.api_key: Optional[str] = os.getenv('GUIJI_API_KEY')
        self.base_url: str = os.getenv('OPENAI_API_BASE', "https://api.siliconflow.cn/v1")
        self.model_name: str = "Qwen/Qwen3-32B"
        self.db_path: str = 'SportsEquipment.db'
        self.promotions_path: str = 'store_promotions.txt'

        if not self.api_key:
            # 如果配置失败，记录到日志文件并抛出异常
            logging.critical("API key 'GUIJI_API_KEY' not found in environment variables.")
            raise ValueError("API key 'GUIJI_API_KEY' not found in environment variables.")


class DatabaseManager:
    """
    处理所有与 SQLite 数据库的交互。
    包括创建、填充和查询。
    """

    def __init__(self, db_path: str):
        self.db_path = db_path
        self._ensure_database_is_ready()

    def _get_connection(self) -> sqlite3.Connection:
        """建立并返回数据库连接。"""
        return sqlite3.connect(self.db_path)

    def _ensure_database_is_ready(self):
        """
        确保数据库和 'products' 表存在。如果不存在，则创建并填充数据。
        """
        logging.info("Checking database and table readiness...")
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='products';")
            if cursor.fetchone() is None:
                logging.info("'products' table not found. Creating and populating...")
                self._create_and_populate_table(cursor)
                conn.commit()
                logging.info("Database and table initialized successfully.")
            else:
                logging.info("Database and table already exist.")

    def _create_and_populate_table(self, cursor: sqlite3.Cursor):
        """创建 'products' 表并插入初始数据。"""
        cursor.execute('''
                       CREATE TABLE products
                       (
                           product_id     TEXT PRIMARY KEY,
                           product_name   TEXT NOT NULL,
                           description    TEXT,
                           specifications TEXT,
                           usage          TEXT,
                           brand          TEXT,
                           price          REAL,
                           stock_quantity INTEGER
                       )''')
        products_data = [
            ('001', '足球', '高品质职业比赛用球，符合国际标准', '圆形，直径22 cm', '职业比赛、学校体育课', '耐克', 120,
             50),
            ('002', '羽毛球拍', '轻量级，适合初中级选手', '碳纤维材质，重量85 g', '业余比赛、家庭娱乐', '尤尼克斯', 300,
             30),
            ('003', '篮球', '室内外可用，耐磨耐用', '皮质，标准7号球', '学校、社区运动场', '斯伯丁', 200, 40),
            ('004', '跑步鞋', '适合长距离跑步，舒适透气', '多种尺码，透气网布', '长跑、日常训练', '阿迪达斯', 500, 20),
            ('005', '瑜伽垫', '防滑材料，厚度适中', '长180cm，宽60cm，厚5mm', '瑜伽、普拉提', '曼达卡', 150, 25)
        ]
        cursor.executemany('INSERT INTO products VALUES (?,?,?,?,?,?,?,?)', products_data)

    def query_by_product_name(self, product_name: str) -> List[Dict[str, Any]]:
        logging.info(f"Querying database for product: '{product_name}'")
        try:
            with self._get_connection() as conn:
                conn.row_factory = sqlite3.Row
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM products WHERE product_name LIKE ?", ('%' + product_name + '%',))
                rows = cursor.fetchall()
                return [dict(row) for row in rows]
        except sqlite3.Error as e:
            logging.error(f"Database query failed for '{product_name}': {e}")
            return []


class ToolManager:
    """
    管理可供 LLM 调用的工具（函数）及其定义。
    """

    def __init__(self, db_manager: DatabaseManager, promotions_path: str):
        self.db_manager = db_manager
        self.promotions_path = promotions_path
        self.available_functions = {
            "query_by_product_name": self.db_manager.query_by_product_name,
            "read_store_promotions": self.read_store_promotions,
        }

    @property
    def tool_definitions(self) -> List[Dict[str, Any]]:
        return [
            {"type": "function", "function": {"name": "query_by_product_name",
                                              "description": "查询数据库以获取与指定产品名称匹配或包含该名称的产品列表。",
                                              "parameters": {"type": "object", "properties": {
                                                  "product_name": {"type": "string",
                                                                   "description": "要搜索的产品名称。支持模糊匹配。"}},
                                                             "required": ["product_name"]}}},
            {"type": "function", "function": {"name": "read_store_promotions",
                                              "description": "读取门店促销文档以查询与指定产品名称相关的促销活动。",
                                              "parameters": {"type": "object", "properties": {
                                                  "product_name": {"type": "string",
                                                                   "description": "要搜索的促销活动所关联的产品名称。"}},
                                                             "required": ["product_name"]}}}
        ]

    def read_store_promotions(self, product_name: str) -> str:
        logging.info(f"Reading promotions for product: '{product_name}'")
        try:
            if not os.path.exists(self.promotions_path):
                with open(self.promotions_path, 'w', encoding='utf-8') as f:
                    f.write("篮球: 本周购买篮球，赠送打气筒一个。\n")
                    f.write("跑步鞋: 凭学生证购买跑步鞋，可享8折优惠。\n")
                logging.warning(f"'{self.promotions_path}' not found, created a default one.")

            with open(self.promotions_path, 'r', encoding='utf-8') as file:
                promotions_content = file.readlines()

            filtered_content = [line.strip() for line in promotions_content if product_name in line]

            return '\n'.join(filtered_content) if filtered_content else "没有找到关于该产品的优惠政策。"
        except IOError as e:
            logging.error(f"Error reading promotions file: {e}")
            return f"读取优惠政策文档时发生错误: {e}"


class SalesChatbot:
    """
    一个集成了 LLM 和工具调用的销售聊天机器人。
    此版本优化了对话流程，并将日志输出到文件。
    """

    def __init__(self, config: AppConfig, db_manager: DatabaseManager, tool_manager: ToolManager):
        self.config = config
        self.client = OpenAI(api_key=config.api_key, base_url=config.base_url)
        self.tool_manager = tool_manager
        self.messages: List[Dict[str, Any]] = []

    def _get_llm_response(self, **kwargs) -> Optional[Any]:
        try:
            return self.client.chat.completions.create(model=self.config.model_name, messages=self.messages, **kwargs)
        except OpenAIError as e:
            logging.error(f"OpenAI API call failed: {e}")
            print(f"助理: 抱歉，与AI服务通信时发生错误，请稍后重试。")
            return None

    def _execute_tool_calls(self, tool_calls: List[Any]):
        for tool_call in tool_calls:
            function_name = tool_call.function.name
            function_to_call = self.tool_manager.available_functions.get(function_name)

            logging.info(f"LLM requests to call tool: '{function_name}'")
            tool_result = ""
            if not function_to_call:
                logging.error(f"Function '{function_name}' is not available.")
                tool_result = f"Error: Tool '{function_name}' not found."
            else:
                try:
                    function_args = json.loads(tool_call.function.arguments)
                    logging.info(f"Executing function: {function_name} with args: {function_args}")
                    function_response = function_to_call(**function_args)
                    tool_result = json.dumps(function_response, ensure_ascii=False, indent=2)
                except Exception as e:
                    logging.error(f"Error executing function '{function_name}': {e}")
                    tool_result = f"Error executing function: {e}"

            self.messages.append(
                {"role": "tool", "name": function_name, "content": tool_result, "tool_call_id": tool_call.id})

    def run(self):
        """启动与用户的交互循环，提供干净的对话界面。"""
        print("您好！我是您的智能体育用品销售助理。您可以问我关于产品或优惠的问题。输入 '退出' 即可结束对话。")
        while True:
            try:
                prompt = input('\n您: ')
                if prompt.lower() in ["退出", "exit", "quit"]:
                    print("助理: 感谢您的使用，再见！")
                    break

                self.messages.append({'role': 'user', 'content': prompt})

                while True:
                    completion = self._get_llm_response(tools=self.tool_manager.tool_definitions, tool_choice="auto")
                    if not completion:
                        self.messages.pop()  # API调用失败，移除最后一条用户消息
                        break

                    response_message = completion.choices[0].message
                    self.messages.append(response_message)

                    if response_message.tool_calls:
                        self._execute_tool_calls(response_message.tool_calls)
                        continue
                    else:
                        print(f"助理: {response_message.content}")
                        break

            except (KeyboardInterrupt, EOFError):
                print("\n助理: 对话已中断，再见！")
                break
            except Exception as e:
                logging.critical(f"An unexpected error occurred in the main loop: {e}", exc_info=True)
                print("助理: 抱歉，系统遇到未知错误，请重试。")


def main():
    try:
        config = AppConfig()
        db_manager = DatabaseManager(config.db_path)
        tool_manager = ToolManager(db_manager, config.promotions_path)
        chatbot = SalesChatbot(config, db_manager, tool_manager)
        chatbot.run()
    except ValueError as e:
        # 配置错误通常是致命的，直接打印到控制台
        print(f"错误：应用程序配置失败，请检查 .env 文件或环境变量 - {e}")
    except Exception as e:
        logging.critical("Application failed to start.", exc_info=True)
        print(f"错误：应用程序启动失败 - {e}")


if __name__ == "__main__":
    main()