轻松一刻来点注释该写的话def fix_bug(): try: do_something() except Exception as e: # 别问为什么捕获所有异常问就是怕崩 print(我也不知道哪里错了但程序还在跑好耶) # TODO: 等我睡醒再修 passFunction Calling 实战介绍起步文件中包含 3 个工具天气查询、计算器、时间查询、完整的 Agent 循环以及 Gradio 界面。目标 1跑通流程将天气查询功能优化为爬取 wttr.in 网站再测试以下 4 个问题重点观察终端输出现在几点了→ 单工具调用帮我算 (1527)*3→ 单工具调用北京天气怎么样→ 单工具调用北京和上海哪个更热→多工具调用观察 LLM 如何连续调用两次 get_weather目标 2a、新增一个工具search_wikipedia(keyword)用于搜索维基百科实操中改为爬取豆瓣网即 search_douban_movieb、新增一个工具read_file(filepath)用于读取本地文件 → 这样 Agent 就能看你的文件了离 RAG 更近一步目标 3a、在 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(scale7): # 左侧占 7/12 chatbot gr.Chatbot(label 对话区域) with gr.Column(scale5): # 右侧占 5/12 thought_display gr.Textbox( label 思考过程, lines25, # 显示 25 行 interactiveFalse, # 只读模式 )测试结果如下项目源代码整体项目代码如下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 fhttp://wttr.in/{city}?formatj1langzh 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,headersheaders) 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/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.8,en-US;q0.5,en;q0.3, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, Upgrade-Insecure-Requests: 1 } # 构建搜索URL search_url fhttps://movie.douban.com/subject_search?search_text{keyword} try: # 发送GET请求 response requests.get(search_url, headersheaders, timeout10) 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: 数学表达式如: 35*2, (1020)/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_keyapi_key or os.getenv(DEEPSEEK_API_KEY), base_urlbase_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( modelself.model, messagesmessages, toolsTOOLS, tool_choiceauto, ) 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_asciiFalse)}) # 执行工具 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(titleFunction Calling Agent, themegr.themes.Soft()) as demo: gr.Markdown( # Function Calling Agent ### 你的第一个能自主调用工具的 AI Agent 试试问它: - 北京今天天气怎么样 - 帮我算一下 (15 27) * 3 - 现在几点了 - 北京和上海哪个温度更高 (多工具调用) - 查询豆瓣电影我不是药神 ) with gr.Row(): with gr.Column(scale7): chatbot gr.Chatbot( label 对话区域, height500, show_labelTrue, ) with gr.Column(scale5): thought_display gr.Textbox( label 思考过程, lines25, show_labelTrue, interactiveFalse, placeholderAgent 的思考过程会在这里显示..., ) with gr.Row(): msg gr.Textbox( label输入消息, placeholder问点什么..., lines2, scale4 ) send_btn gr.Button( 发送, variantprimary, scale1) with gr.Row(): clear_btn gr.Button(️ 清空对话, variantsecondary, scale1) clear_thought_btn gr.Button( 清空思考过程, variantsecondary, scale1) # 状态 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( 可用工具列表, openFalse): 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_name127.0.0.1, server_port7860, themegr.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 助手。
