轻松一刻：来点注释该写的话

def fix_bug():
    try:
        do_something()
    except Exception as e:
        # 别问为什么捕获所有异常，问就是怕崩
        print("我也不知道哪里错了，但程序还在跑，好耶！")
        # TODO: 等我睡醒再修
        pass

Function Calling 实战

介绍

起步文件中包含 3 个工具（天气查询、计算器、时间查询）、完整的 Agent 循环以及 Gradio 界面。

目标 1：

跑通流程，将天气查询功能优化为爬取 wttr.in 网站，再测试以下 4 个问题（重点观察终端输出）：

1. "现在几点了" → 单工具调用
2. "帮我算 (15+27)*3" → 单工具调用
3. "北京天气怎么样" → 单工具调用
4. "北京和上海哪个更热" → 多工具调用，观察 LLM 如何连续调用两次 get_weather

目标 2：

a、新增一个工具：search_wikipedia(keyword)，用于搜索维基百科（实操中改为爬取豆瓣网，即 search_douban_movie）

b、新增一个工具：read_file(filepath)，用于读取本地文件 → 这样 Agent 就能"看"你的文件了，离 RAG 更近一步

目标 3：

a、在 Gradio 界面增加一个"思考过程"展示区，实时显示 Agent 当前正在调用哪个工具

b、修改 system_prompt，让 Agent 更倾向于使用工具，或更倾向于直接回答 → 观察两种 prompt 下 Agent 行为的差异

流程图：
   用户提问
      ↓
   ┌─→ LLM 思考
   │     ↓
   │   需要调工具？ ──否──→ 输出最终答案，结束
   │     ↓ 是
   │   执行工具，拿到结果
   │     ↓
   └── 把结果喂回 LLM（回到循环开头）

目标 1 解决及测试

将【Day4-6】中已实操过的天气查询工具代码优化过来，测试结果如下：

单工具调用

多工具调用

目标 2 解决及测试

新增两个工具，完整代码见文末。

测试结果如下：

目标 3 解决及测试

a、给 Agent 返回值增加"思考过程"

b、用列表记录思考过程

c、改动三：界面分栏布局

with gr.Row():
    with gr.Column(scale=7):  # 左侧占 7/12
        chatbot = gr.Chatbot(label="💬 对话区域")
    with gr.Column(scale=5):  # 右侧占 5/12
        thought_display = gr.Textbox(
            label="🧠 思考过程",
            lines=25,  # 显示 25 行
            interactive=False,  # 只读模式
        )

测试结果如下：

项目源代码

整体项目代码如下：

import os
import json
import datetime
import gradio as gr
from openai import OpenAI
from dotenv import load_dotenv
import requests
import re
from bs4 import BeautifulSoup
import pandas as pd
from pathlib import Path

load_dotenv()  # 读取 .env 里的 DEEPSEEK_API_KEY


# ============================================================
# 1. 工具函数 —— Agent 能调用的"手和眼"
# ============================================================
# 每个工具就是一个普通的 Python 函数，Agent 会根据用户问题自动决定调用哪个

def get_weather(city: str) -> str:
    """查询指定城市的天气 """
    # # TODO 练习: 你可以把这里改成调用真实天气 API
    # weather_data = {
    #     "北京": "晴天, 28°C, 湿度45%, 北风3级",
    #     "上海": "多云, 26°C, 湿度65%, 东南风2级",
    #     "广州": "阵雨, 30°C, 湿度80%, 南风3级",
    #     "深圳": "晴转多云, 29°C, 湿度70%, 东风2级",
    #     "成都": "阴天, 24°C, 湿度75%, 微风",
    # }
    try:
        url = f"http://wttr.in/{city}?format=j1&lang=zh"
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
                          ""
        }
        result = requests.get(url,headers=headers)
        result.raise_for_status()
        data = result.json()
        current = data["current_condition"][0]
        area = data["nearest_area"][0]
        return (
            f"{area['areaName'][0]['value']}天气\n"
            f"天气:{current['weatherDesc'][0]['value']}\n"
            f"温度:{current['temp_C']}°C（体感{current['FeelsLikeC']}°C）\n"
            f"湿度:{current['humidity']}%\n"
            f"风速:{current['windspeedKmph']}km/h"
        )
    except Exception as e:
        print(e)
        return f"暂无 {city} 的天气数据"


def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        # 注意: 实际生产中不要用 eval, 这里为了演示简化处理
        # 只允许数字和基本运算符
        allowed = set("0123456789+-*/().% ")
        if not all(c in allowed for c in expression):
            return "错误: 只支持基本数学运算 (+,-,*,/,%,括号)"
        result = eval(expression)
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"


def get_current_time() -> str:
    """获取当前日期和时间"""
    now = datetime.datetime.now()
    return now.strftime("%Y年%m月%d日 %H:%M:%S 星期") + \
           ["一", "二", "三", "四", "五", "六", "日"][now.weekday()]

#2026/08/19 加入新功能，查询豆瓣网
def search_douban_movie(keyword):
    """
    爬取豆瓣电影信息
    :param keyword: 搜索关键词（字符串）
    :return: 格式化后的电影信息字符串
    """
    # 设置请求头，模拟浏览器访问
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
        'Referer': 'https://movie.douban.com/',
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
        'Accept-Language': 'zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3',
        'Accept-Encoding': 'gzip, deflate, br',
        'Connection': 'keep-alive',
        'Upgrade-Insecure-Requests': '1'
    }

    # 构建搜索URL
    search_url = f'https://movie.douban.com/subject_search?search_text={keyword}'

    try:
        # 发送GET请求
        response = requests.get(search_url, headers=headers, timeout=10)
        response.encoding = 'utf-8'

        # 检查响应状态
        if response.status_code != 200:
            return f"请求失败，状态码: {response.status_code}"

        # 解析HTML
        soup = BeautifulSoup(response.text, 'html.parser')

        # 查找电影列表 - 豆瓣搜索页面的结构
        movie_items = soup.find_all('div', class_='item')

        if not movie_items:
            return f"未找到与 '{keyword}' 相关的电影信息"

        # 存储结果
        results = []
        results.append(f"搜索结果: '{keyword}'")
        results.append("-" * 60)

        # 遍历前5个电影（避免过多）
        for idx, item in enumerate(movie_items[:5], 1):
            try:
                # 提取电影标题
                title_elem = item.find('a', class_='title-text')
                title = title_elem.text.strip() if title_elem else "未知标题"

                # 提取评分
                rating_elem = item.find('span', class_='rating_nums')
                rating = rating_elem.text.strip() if rating_elem else "暂无评分"

                # 提取评分人数
                rating_people = item.find('span', class_='pl')
                if rating_people:
                    people_text = rating_people.text.strip()
                    people_match = re.search(r'(\d+)', people_text)
                    people = people_match.group(1) if people_match else "0"
                else:
                    people = "0"

                # 提取电影详情链接
                link_elem = item.find('a', class_='title-text')
                link = link_elem['href'] if link_elem else "#"

                # 提取简短简介
                desc_elem = item.find('span', class_='pl')
                if desc_elem and '简介' in desc_elem.text:
                    # 如果有简介信息，提取
                    desc = desc_elem.text.strip()
                else:
                    # 尝试获取其他描述信息
                    desc = "暂无简介"

                # 格式化结果
                result = f"""
【电影 {idx}】
标题: {title}
评分: {rating} ({people}人评价)
链接: {link}
简介: {desc[:50]}...
"""
                results.append(result.strip())

            except Exception as e:
                results.append(f"【电影 {idx}】 解析失败: {str(e)}")

        # 返回结果字符串
        return "\n\n".join(results)

    except requests.exceptions.Timeout:
        return "请求超时，请稍后再试"
    except requests.exceptions.ConnectionError:
        return "网络连接失败，请检查网络"
    except Exception as e:
        return f"爬取过程中出现错误: {str(e)}"


def read_file(filepath):
    """
    简化版：只读取第一个sheet，输出表格形式
    :param filepath: 文件名
    :return: 字符串
    """
    script_dir = Path(__file__).parent
    # 构建database文件夹路径
    database_dir = script_dir / "数据仓库"
    try:
        if not os.path.exists(database_dir/filepath):
            # 获取所有存在的文件
            existing_files = [f for f in database_dir.iterdir() if f.is_file()]

            # 分类文件
            excel_files = [f for f in existing_files if f.suffix.lower() in ['.xlsx', '.xls']]
            other_files = [f for f in existing_files if f.suffix.lower() not in ['.xlsx', '.xls']]

            # 构建返回信息
            result_parts = []
            result_parts.append(f"❌ 文件 '{filepath}' 在database文件夹中不存在")
            result_parts.append("=" * 70)

            if not existing_files:
                result_parts.append("📁 database文件夹为空，请放入Excel文件")
                return "\n".join(result_parts)
            if excel_files:
                result_parts.append(f"📊 可用的Excel文件（共{len(excel_files)}个）:")
                result_parts.append("-" * 70)
                for i, file in enumerate(sorted(excel_files), 1):
                    file_size = file.stat().st_size
                    # 判断是否为xlsx或xls
                    file_type = "📗" if file.suffix.lower() == '.xlsx' else "📘"
                    result_parts.append(f"  {i:2d}. {file_type} {file.name} ")
            return f"文件 '{filepath}' 不存在，数据仓库中有以下数据：{result_parts}"

        # 读取第一个sheet
        df = pd.read_excel(database_dir/filepath)

        result_parts = []
        result_parts.append(f"文件: {os.path.basename(database_dir/filepath)}")
        result_parts.append("=" * 60)

        # 获取列宽
        col_widths = []
        for col in df.columns:
            max_len = max(len(str(col)), df[col].astype(str).str.len().max())
            col_widths.append(min(max_len, 20))  # 限制最大宽度

        # 生成表格
        # 表头
        header = ""
        for col, width in zip(df.columns, col_widths):
            header += f"{str(col)[:width]:<{width}} | "
        result_parts.append(header)
        result_parts.append("-" * len(header))

        # 数据行
        for idx, row in df.iterrows():
            row_str = ""
            for col, width in zip(df.columns, col_widths):
                value = str(row[col])[:width] if not pd.isna(row[col]) else "空"
                row_str += f"{value:<{width}} | "
            result_parts.append(row_str)
            if idx >= 100:  # 最多显示100行
                result_parts.append(f"... 还有 {len(df) - 100} 行数据")
                break

        result_parts.append("=" * 60)
        result_parts.append(f"共 {len(df)} 行, {len(df.columns)} 列")

        return "\n".join(result_parts)

    except Exception as e:
        return f"错误: {str(e)}"

# ============================================================
# 2. 工具 Schema 定义 —— 告诉 LLM "你有哪些工具、怎么用"
# ============================================================
# 这段 JSON 是 OpenAI Function Calling 的标准格式
# LLM 会根据 name + description + parameters 来决定是否调用某个工具

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的当前天气情况，包括温度、湿度、风力等信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "要查询天气的城市名称，如：北京、上海、广州"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "计算数学表达式，支持加减乘除、括号、取余等运算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如: 3+5*2, (10+20)/3, 100%7"
                    }
                },
                "required": ["expression"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "获取当前的日期和时间，不需要任何参数",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_douban_movie",
            "description": "对电影相关信息查询豆瓣网",
            "parameters": {
                "type": "object",
                "properties": {
                    "keyword": {
                        "type": "string",
                        "description": "要查询的keyword，例如 我不是药神"
                    }
                },
                "required": ["keyword"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取本地数据仓库文件获取对应文件信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "filepath": {
                        "type": "string",
                        "description": "filepath，例如 销售数据.xlsx"
                    }
                },
                "required": ["filepath"]
            }
        }
    },
]

# 工具名 → 函数的映射表, 执行时用名字查找函数
TOOL_MAP = {
    "get_weather": get_weather,
    "calculate": calculate,
    "get_current_time": get_current_time,
    "search_douban_movie": search_douban_movie,
    "read_file":read_file,
}


# ============================================================
# 3. Agent 核心循环 —— 这是 Agent 的"大脑循环"
# ============================================================
#
# 流程图:
#   用户提问
#      ↓
#   ┌─→ LLM 思考
#   │     ↓
#   │   需要调工具? ──否──→ 输出最终答案, 结束
#   │     ↓ 是
#   │   执行工具, 拿到结果
#   │     ↓
#   └── 把结果喂回 LLM (回到循环开头)
#
# 关键理解: LLM 不是一次就给出答案, 而是可能经过多轮 "思考→调工具→看结果→再思考"

class FunctionAgent:
    def __init__(self, api_key: str = None, base_url: str = "https://api.deepseek.com"):
        self.client = OpenAI(
            api_key=api_key or os.getenv("DEEPSEEK_API_KEY"),
            base_url=base_url
        )
        self.model = "deepseek-chat"
        self.max_iterations = 10
        self.system_prompt = (
            "你是一个有用的AI助手。你可以使用以下工具来帮助用户:\n"
            "1. get_weather - 查询城市天气\n"
            "2. calculate - 计算数学表达式\n"
            "3. get_current_time - 获取当前时间\n"
            "4. search_douban_movie - 获取关键词相关电影豆瓣网信息\n"
            "5. read_file - 读取本地数据仓库数据,没有对应文件时返回所有已存在文件名\n"
            "规则:\n"
            "- 如果用户的问题需要使用工具才能回答，请调用对应的工具\n"
            "- 如果不需要工具就能回答，直接回答即可\n"
            "- 可以在一次对话中调用多个工具\n"
            "- 调用工具后，根据工具返回的结果给出最终回答\n"
            "- 用中文回答"
        )

    def run(self, user_message: str, history: list = None) -> tuple:
        """
        Agent 主循环: 处理用户消息, 返回最终回答和思考过程

        返回:
            (最终回答, 思考过程文本)
        """
        messages = [{"role": "system", "content": self.system_prompt}]
        if history:
            messages.extend(history)
        messages.append({"role": "user", "content": user_message})

        # 用于记录思考过程
        thought_process = []
        thought_process.append(f"📝 用户提问: {user_message}")
        thought_process.append("=" * 50)

        for i in range(self.max_iterations):
            step_info = f"\n🔄 第 {i + 1} 轮思考"
            thought_process.append(step_info)
            thought_process.append("-" * 40)

            # 调用 LLM
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=TOOLS,
                tool_choice="auto",
            )

            msg = response.choices[0].message

            # 检查是否要调用工具
            if msg.tool_calls:
                thought_process.append("🤔 LLM 决定调用工具:")
                messages.append(msg)

                for tool_call in msg.tool_calls:
                    func_name = tool_call.function.name
                    func_args = json.loads(tool_call.function.arguments)

                    thought_process.append(f"  📌 调用工具: {func_name}")
                    thought_process.append(f"     参数: {json.dumps(func_args, ensure_ascii=False)}")

                    # 执行工具
                    func = TOOL_MAP.get(func_name)
                    if func:
                        result = func(**func_args)
                    else:
                        result = f"错误: 未知工具 {func_name}"

                    thought_process.append(f"  ✅ 工具返回: {result[:200]}{'...' if len(result) > 200 else ''}")
                    thought_process.append("")

                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": str(result)
                    })

                continue

            else:
                # LLM 给出最终答案
                thought_process.append("💡 LLM 给出最终答案 (无需再调用工具)")
                thought_process.append("=" * 50)
                thought_process.append(f"✅ 最终回答: {msg.content}")
                return msg.content, "\n".join(thought_process)

        error_msg = "Agent 达到最大循环次数，可能陷入了死循环"
        thought_process.append(f"❌ {error_msg}")
        return error_msg, "\n".join(thought_process)

    def chat_with_thought(self, user_message: str, history: list) -> tuple:
        """
        Gradio 聊天接口: 返回回复、更新后的历史和思考过程
        """
        # 转换历史格式
        openai_history = []
        for h in history:
            if h["role"] == "user":
                openai_history.append({"role": "user", "content": h["content"]})
            elif h["role"] == "assistant":
                openai_history.append({"role": "assistant", "content": h["content"]})

        # 运行 Agent
        reply, thought = self.run(user_message, openai_history)

        # 更新历史
        history.append({"role": "user", "content": user_message})
        history.append({"role": "assistant", "content": reply})

        return history, history, thought


# ============================================================
# 4. Gradio 界面 —— 带思考过程展示
# ============================================================

def create_interface():
    agent = FunctionAgent()

    with gr.Blocks(title="Function Calling Agent", theme=gr.themes.Soft()) as demo:
        gr.Markdown("""
        # 🤖 Function Calling Agent
        ### 你的第一个能自主调用工具的 AI Agent

        试试问它:
        - "北京今天天气怎么样？"
        - "帮我算一下 (15 + 27) * 3"
        - "现在几点了？"
        - "北京和上海哪个温度更高？" (多工具调用)
        - "查询豆瓣电影：我不是药神"
        """)

        with gr.Row():
            with gr.Column(scale=7):
                chatbot = gr.Chatbot(
                    label="💬 对话区域",
                    height=500,
                    show_label=True,
                )
            with gr.Column(scale=5):
                thought_display = gr.Textbox(
                    label="🧠 思考过程",
                    lines=25,
                    show_label=True,
                    interactive=False,
                    placeholder="Agent 的思考过程会在这里显示...",
                )

        with gr.Row():
            msg = gr.Textbox(
                label="输入消息",
                placeholder="问点什么...",
                lines=2,
                scale=4
            )
            send_btn = gr.Button("🚀 发送", variant="primary", scale=1)

        with gr.Row():
            clear_btn = gr.Button("🗑️ 清空对话", variant="secondary", scale=1)
            clear_thought_btn = gr.Button("🧹 清空思考过程", variant="secondary", scale=1)

        # 状态
        chat_state = gr.State([])
        thought_state = gr.State("")

        # 事件绑定
        def clear_all():
            return [], [], ""

        def send_message(msg_text, history, thought_text):
            if not msg_text.strip():
                return history, history, thought_text

            # 调用 agent
            new_history, new_state, thought = agent.chat_with_thought(msg_text, history)
            return new_history, new_state, thought

        send_btn.click(
            send_message,
            inputs=[msg, chat_state, thought_state],
            outputs=[chatbot, chat_state, thought_display]
        ).then(
            lambda: "",
            outputs=[msg]
        )

        msg.submit(
            send_message,
            inputs=[msg, chat_state, thought_state],
            outputs=[chatbot, chat_state, thought_display]
        ).then(
            lambda: "",
            outputs=[msg]
        )

        clear_btn.click(
            clear_all,
            outputs=[chatbot, chat_state, thought_display]
        )

        clear_thought_btn.click(
            lambda: "",
            outputs=[thought_display]
        )

        # 添加一个折叠面板，显示工具列表
        with gr.Accordion("🔧 可用工具列表", open=False):
            gr.Markdown("""
            | 工具名称 | 功能描述 | 参数 |
            |---------|---------|------|
            | `get_weather` | 查询城市天气 | `city`: 城市名称 |
            | `calculate` | 计算数学表达式 | `expression`: 数学公式 |
            | `get_current_time` | 获取当前时间 | 无参数 |
            | `search_douban_movie` | 查询豆瓣电影 | `keyword`: 电影关键词 |
            | `read_file` | 读取本地数据文件 | `filepath`: 文件名 |
            """)

    return demo


# ============================================================
# 5. 启动
# ============================================================
if __name__ == "__main__":
    if not os.getenv("DEEPSEEK_API_KEY"):
        print("错误: 请在 .env 文件中设置 DEEPSEEK_API_KEY")
        exit(1)

    demo = create_interface()
    demo.launch(server_name="127.0.0.1", server_port=7860, theme=gr.themes.Soft())

总结与展望

本项目完整实现了一个基于 Function Calling 的 AI Agent 实战案例，核心成果包括：将天气查询优化为爬取 wttr.in 网站、新增豆瓣电影搜索与本地文件读取两个工具、在 Gradio 界面中加入思考过程展示区，并验证了单工具调用与多工具调用的完整链路。通过本项目的实践，可以清晰理解 Agent 循环的核心机制：LLM 思考、决定调用工具、执行工具、将结果喂回 LLM 的闭环流程。

不过，当前实现仍存在一些局限性，值得在后续迭代中重点关注：

• 安全风险：calculate 工具使用了 eval 执行表达式，虽然做了字符白名单过滤，但仍存在被绕过注入的风险，生产环境应改用安全的表达式解析库。
• 爬虫稳定性：豆瓣与 wttr.in 的页面结构可能随时变化，且对请求频率有限制，当前实现缺少重试机制、请求限速和异常降级策略，长期运行容易失效。
• 上下文管理：Agent 每次对话都会把完整历史拼进 messages，长对话下容易超出模型上下文窗口，也缺少对历史消息的裁剪与摘要压缩。
• 工具扩展性：工具注册、参数校验和错误处理都写死在代码里，新增工具需要改动多处，维护成本较高。

基于以上分析，后续可以从以下几个方向继续完善：

• 工具管理框架：将工具定义、注册、执行统一封装为装饰器或配置化机制，降低新增工具的成本，并统一处理参数校验与异常。
• 记忆机制：引入短期记忆（会话内摘要）与长期记忆（向量数据库存储历史关键信息），让 Agent 在多轮对话中保持上下文连贯。
• RAG 集成：将 read_file 工具升级为基于向量检索的问答能力，让 Agent 能针对本地文档进行更精准的检索与回答。
• 安全加固：替换 eval 为安全计算库，为爬虫增加重试与限速策略，并对工具调用结果做长度截断与敏感信息过滤。

总的来说，本项目已经搭建起一个功能完整的 Agent 雏形，后续只要在安全、稳定与智能化三个方向持续打磨，就能逐步演进为一个可投入实际业务使用的 AI 助手。