輕松一刻來點(diǎn)注釋該寫的話def fix_bug(): try: do_something() except Exception as e: # 別問為什么捕獲所有異常問就是怕崩 print(我也不知道哪里錯(cuò)了但程序還在跑好耶) # TODO: 等我睡醒再修 passFunction Calling 實(shí)戰(zhàn)介紹起步文件中包含 3 個(gè)工具天氣查詢、計(jì)算器、時(shí)間查詢、完整的 Agent 循環(huán)以及 Gradio 界面。目標(biāo) 1跑通流程將天氣查詢功能優(yōu)化為爬取 wttr.in 網(wǎng)站再測試以下 4 個(gè)問題重點(diǎn)觀察終端輸出現(xiàn)在幾點(diǎn)了→ 單工具調(diào)用幫我算 (1527)*3→ 單工具調(diào)用北京天氣怎么樣→ 單工具調(diào)用北京和上海哪個(gè)更熱→多工具調(diào)用觀察 LLM 如何連續(xù)調(diào)用兩次 get_weather目標(biāo) 2a、新增一個(gè)工具search_wikipedia(keyword)用于搜索維基百科實(shí)操中改為爬取豆瓣網(wǎng)即 search_douban_movieb、新增一個(gè)工具read_file(filepath)用于讀取本地文件 → 這樣 Agent 就能看你的文件了離 RAG 更近一步目標(biāo) 3a、在 Gradio 界面增加一個(gè)思考過程展示區(qū)實(shí)時(shí)顯示 Agent 當(dāng)前正在調(diào)用哪個(gè)工具b、修改 system_prompt讓 Agent 更傾向于使用工具或更傾向于直接回答 → 觀察兩種 prompt 下 Agent 行為的差異流程圖 用戶提問 ↓ ┌─→ LLM 思考 │ ↓ │ 需要調(diào)工具 ──否──→ 輸出最終答案結(jié)束 │ ↓ 是 │ 執(zhí)行工具拿到結(jié)果 │ ↓ └── 把結(jié)果喂回 LLM回到循環(huán)開頭目標(biāo) 1 解決及測試將【Day4-6】中已實(shí)操過的天氣查詢工具代碼優(yōu)化過來測試結(jié)果如下單工具調(diào)用多工具調(diào)用目標(biāo) 2 解決及測試新增兩個(gè)工具完整代碼見文末。測試結(jié)果如下目標(biāo) 3 解決及測試a、給 Agent 返回值增加思考過程b、用列表記錄思考過程c、改動(dòng)三界面分欄布局with gr.Row(): with gr.Column(scale7): # 左側(cè)占 7/12 chatbot gr.Chatbot(label 對(duì)話區(qū)域) with gr.Column(scale5): # 右側(cè)占 5/12 thought_display gr.Textbox( label 思考過程, lines25, # 顯示 25 行 interactiveFalse, # 只讀模式 )測試結(jié)果如下項(xiàng)目源代碼整體項(xiàng)目代碼如下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. 工具函數(shù) —— Agent 能調(diào)用的手和眼 # # 每個(gè)工具就是一個(gè)普通的 Python 函數(shù)Agent 會(huì)根據(jù)用戶問題自動(dòng)決定調(diào)用哪個(gè) def get_weather(city: str) - str: 查詢指定城市的天氣 # # TODO 練習(xí): 你可以把這里改成調(diào)用真實(shí)天氣 API # weather_data { # 北京: 晴天, 28°C, 濕度45%, 北風(fēng)3級(jí), # 上海: 多云, 26°C, 濕度65%, 東南風(fēng)2級(jí), # 廣州: 陣雨, 30°C, 濕度80%, 南風(fēng)3級(jí), # 深圳: 晴轉(zhuǎn)多云, 29°C, 濕度70%, 東風(fēng)2級(jí), # 成都: 陰天, 24°C, 濕度75%, 微風(fēng), # } 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風(fēng)速:{current[windspeedKmph]}km/h ) except Exception as e: print(e) return f暫無 {city} 的天氣數(shù)據(jù) def calculate(expression: str) - str: 計(jì)算數(shù)學(xué)表達(dá)式 try: # 注意: 實(shí)際生產(chǎn)中不要用 eval, 這里為了演示簡化處理 # 只允許數(shù)字和基本運(yùn)算符 allowed set(0123456789-*/().% ) if not all(c in allowed for c in expression): return 錯(cuò)誤: 只支持基本數(shù)學(xué)運(yùn)算 (,-,*,/,%,括號(hào)) result eval(expression) return f{expression} {result} except Exception as e: return f計(jì)算錯(cuò)誤: {str(e)} def get_current_time() - str: 獲取當(dāng)前日期和時(shí)間 now datetime.datetime.now() return now.strftime(%Y年%m月%d日 %H:%M:%S 星期) \ [一, 二, 三, 四, 五, 六, 日][now.weekday()] #2026/08/19 加入新功能查詢豆瓣網(wǎng) def search_douban_movie(keyword): 爬取豆瓣電影信息 :param keyword: 搜索關(guān)鍵詞字符串 :return: 格式化后的電影信息字符串 # 設(shè)置請(qǐng)求頭模擬瀏覽器訪問 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 } # 構(gòu)建搜索URL search_url fhttps://movie.douban.com/subject_search?search_text{keyword} try: # 發(fā)送GET請(qǐng)求 response requests.get(search_url, headersheaders, timeout10) response.encoding utf-8 # 檢查響應(yīng)狀態(tài) if response.status_code ! 200: return f請(qǐng)求失敗狀態(tài)碼: {response.status_code} # 解析HTML soup BeautifulSoup(response.text, html.parser) # 查找電影列表 - 豆瓣搜索頁面的結(jié)構(gòu) movie_items soup.find_all(div, class_item) if not movie_items: return f未找到與 {keyword} 相關(guān)的電影信息 # 存儲(chǔ)結(jié)果 results [] results.append(f搜索結(jié)果: {keyword}) results.append(- * 60) # 遍歷前5個(gè)電影避免過多 for idx, item in enumerate(movie_items[:5], 1): try: # 提取電影標(biāo)題 title_elem item.find(a, class_title-text) title title_elem.text.strip() if title_elem else 未知標(biāo)題 # 提取評(píng)分 rating_elem item.find(span, class_rating_nums) rating rating_elem.text.strip() if rating_elem else 暫無評(píng)分 # 提取評(píng)分人數(shù) 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 暫無簡介 # 格式化結(jié)果 result f 【電影 {idx}】 標(biāo)題: {title} 評(píng)分: {rating} ({people}人評(píng)價(jià)) 鏈接: {link} 簡介: {desc[:50]}... results.append(result.strip()) except Exception as e: results.append(f【電影 {idx}】 解析失敗: {str(e)}) # 返回結(jié)果字符串 return \n\n.join(results) except requests.exceptions.Timeout: return 請(qǐng)求超時(shí)請(qǐng)稍后再試 except requests.exceptions.ConnectionError: return 網(wǎng)絡(luò)連接失敗請(qǐng)檢查網(wǎng)絡(luò) except Exception as e: return f爬取過程中出現(xiàn)錯(cuò)誤: {str(e)} def read_file(filepath): 簡化版只讀取第一個(gè)sheet輸出表格形式 :param filepath: 文件名 :return: 字符串 script_dir Path(__file__).parent # 構(gòu)建database文件夾路徑 database_dir script_dir / 數(shù)據(jù)倉庫 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]] # 構(gòu)建返回信息 result_parts [] result_parts.append(f? 文件 {filepath} 在database文件夾中不存在) result_parts.append( * 70) if not existing_files: result_parts.append( database文件夾為空請(qǐng)放入Excel文件) return \n.join(result_parts) if excel_files: result_parts.append(f 可用的Excel文件共{len(excel_files)}個(gè):) 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} 不存在數(shù)據(jù)倉庫中有以下數(shù)據(jù){result_parts} # 讀取第一個(gè)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)) # 數(shù)據(jù)行 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} 行數(shù)據(jù)) 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錯(cuò)誤: {str(e)} # # 2. 工具 Schema 定義 —— 告訴 LLM 你有哪些工具、怎么用 # # 這段 JSON 是 OpenAI Function Calling 的標(biāo)準(zhǔn)格式 # LLM 會(huì)根據(jù) name description parameters 來決定是否調(diào)用某個(gè)工具 TOOLS [ { type: function, function: { name: get_weather, description: 查詢指定城市的當(dāng)前天氣情況包括溫度、濕度、風(fēng)力等信息, parameters: { type: object, properties: { city: { type: string, description: 要查詢天氣的城市名稱如北京、上海、廣州 } }, required: [city] } } }, { type: function, function: { name: calculate, description: 計(jì)算數(shù)學(xué)表達(dá)式支持加減乘除、括號(hào)、取余等運(yùn)算, parameters: { type: object, properties: { expression: { type: string, description: 數(shù)學(xué)表達(dá)式如: 35*2, (1020)/3, 100%7 } }, required: [expression] } } }, { type: function, function: { name: get_current_time, description: 獲取當(dāng)前的日期和時(shí)間不需要任何參數(shù), parameters: { type: object, properties: {}, required: [] } } }, { type: function, function: { name: search_douban_movie, description: 對(duì)電影相關(guān)信息查詢豆瓣網(wǎng), parameters: { type: object, properties: { keyword: { type: string, description: 要查詢的keyword例如 我不是藥神 } }, required: [keyword] } } }, { type: function, function: { name: read_file, description: 讀取本地?cái)?shù)據(jù)倉庫文件獲取對(duì)應(yīng)文件信息, parameters: { type: object, properties: { filepath: { type: string, description: filepath例如 銷售數(shù)據(jù).xlsx } }, required: [filepath] } } }, ] # 工具名 → 函數(shù)的映射表, 執(zhí)行時(shí)用名字查找函數(shù) 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 核心循環(huán) —— 這是 Agent 的大腦循環(huán) # # # 流程圖: # 用戶提問 # ↓ # ┌─→ LLM 思考 # │ ↓ # │ 需要調(diào)工具? ──否──→ 輸出最終答案, 結(jié)束 # │ ↓ 是 # │ 執(zhí)行工具, 拿到結(jié)果 # │ ↓ # └── 把結(jié)果喂回 LLM (回到循環(huán)開頭) # # 關(guān)鍵理解: LLM 不是一次就給出答案, 而是可能經(jīng)過多輪 思考→調(diào)工具→看結(jié)果→再思考 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 ( 你是一個(gè)有用的AI助手。你可以使用以下工具來幫助用戶:\n 1. get_weather - 查詢城市天氣\n 2. calculate - 計(jì)算數(shù)學(xué)表達(dá)式\n 3. get_current_time - 獲取當(dāng)前時(shí)間\n 4. search_douban_movie - 獲取關(guān)鍵詞相關(guān)電影豆瓣網(wǎng)信息\n 5. read_file - 讀取本地?cái)?shù)據(jù)倉庫數(shù)據(jù),沒有對(duì)應(yīng)文件時(shí)返回所有已存在文件名\n 規(guī)則:\n - 如果用戶的問題需要使用工具才能回答請(qǐng)調(diào)用對(duì)應(yīng)的工具\(yùn)n - 如果不需要工具就能回答直接回答即可\n - 可以在一次對(duì)話中調(diào)用多個(gè)工具\(yùn)n - 調(diào)用工具后根據(jù)工具返回的結(jié)果給出最終回答\n - 用中文回答 ) def run(self, user_message: str, history: list None) - tuple: Agent 主循環(huán): 處理用戶消息, 返回最終回答和思考過程 返回: (最終回答, 思考過程文本) 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) # 調(diào)用 LLM response self.client.chat.completions.create( modelself.model, messagesmessages, toolsTOOLS, tool_choiceauto, ) msg response.choices[0].message # 檢查是否要調(diào)用工具 if msg.tool_calls: thought_process.append( LLM 決定調(diào)用工具:) 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 調(diào)用工具: {func_name}) thought_process.append(f 參數(shù): {json.dumps(func_args, ensure_asciiFalse)}) # 執(zhí)行工具 func TOOL_MAP.get(func_name) if func: result func(**func_args) else: result f錯(cuò)誤: 未知工具 {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 給出最終答案 (無需再調(diào)用工具)) thought_process.append( * 50) thought_process.append(f? 最終回答: {msg.content}) return msg.content, \n.join(thought_process) error_msg Agent 達(dá)到最大循環(huán)次數(shù)可能陷入了死循環(huán) 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 聊天接口: 返回回復(fù)、更新后的歷史和思考過程 # 轉(zhuǎn)換歷史格式 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]}) # 運(yùn)行 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 ### 你的第一個(gè)能自主調(diào)用工具的 AI Agent 試試問它: - 北京今天天氣怎么樣 - 幫我算一下 (15 27) * 3 - 現(xiàn)在幾點(diǎn)了 - 北京和上海哪個(gè)溫度更高 (多工具調(diào)用) - 查詢豆瓣電影我不是藥神 ) with gr.Row(): with gr.Column(scale7): chatbot gr.Chatbot( label 對(duì)話區(qū)域, height500, show_labelTrue, ) with gr.Column(scale5): thought_display gr.Textbox( label 思考過程, lines25, show_labelTrue, interactiveFalse, placeholderAgent 的思考過程會(huì)在這里顯示..., ) with gr.Row(): msg gr.Textbox( label輸入消息, placeholder問點(diǎn)什么..., lines2, scale4 ) send_btn gr.Button( 發(fā)送, variantprimary, scale1) with gr.Row(): clear_btn gr.Button(? 清空對(duì)話, variantsecondary, scale1) clear_thought_btn gr.Button( 清空思考過程, variantsecondary, scale1) # 狀態(tài) 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 # 調(diào)用 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] ) # 添加一個(gè)折疊面板顯示工具列表 with gr.Accordion( 可用工具列表, openFalse): gr.Markdown( | 工具名稱 | 功能描述 | 參數(shù) | |---------|---------|------| | get_weather | 查詢城市天氣 | city: 城市名稱 | | calculate | 計(jì)算數(shù)學(xué)表達(dá)式 | expression: 數(shù)學(xué)公式 | | get_current_time | 獲取當(dāng)前時(shí)間 | 無參數(shù) | | search_douban_movie | 查詢豆瓣電影 | keyword: 電影關(guān)鍵詞 | | read_file | 讀取本地?cái)?shù)據(jù)文件 | filepath: 文件名 | ) return demo # # 5. 啟動(dòng) # if __name__ __main__: if not os.getenv(DEEPSEEK_API_KEY): print(錯(cuò)誤: 請(qǐng)?jiān)?.env 文件中設(shè)置 DEEPSEEK_API_KEY) exit(1) demo create_interface() demo.launch(server_name127.0.0.1, server_port7860, themegr.themes.Soft())總結(jié)與展望本項(xiàng)目完整實(shí)現(xiàn)了一個(gè)基于 Function Calling 的 AI Agent 實(shí)戰(zhàn)案例核心成果包括將天氣查詢優(yōu)化為爬取 wttr.in 網(wǎng)站、新增豆瓣電影搜索與本地文件讀取兩個(gè)工具、在 Gradio 界面中加入思考過程展示區(qū)并驗(yàn)證了單工具調(diào)用與多工具調(diào)用的完整鏈路。通過本項(xiàng)目的實(shí)踐可以清晰理解 Agent 循環(huán)的核心機(jī)制LLM 思考、決定調(diào)用工具、執(zhí)行工具、將結(jié)果喂回 LLM 的閉環(huán)流程。不過當(dāng)前實(shí)現(xiàn)仍存在一些局限性值得在后續(xù)迭代中重點(diǎn)關(guān)注安全風(fēng)險(xiǎn)calculate工具使用了eval執(zhí)行表達(dá)式雖然做了字符白名單過濾但仍存在被繞過注入的風(fēng)險(xiǎn)生產(chǎn)環(huán)境應(yīng)改用安全的表達(dá)式解析庫。爬蟲穩(wěn)定性豆瓣與 wttr.in 的頁面結(jié)構(gòu)可能隨時(shí)變化且對(duì)請(qǐng)求頻率有限制當(dāng)前實(shí)現(xiàn)缺少重試機(jī)制、請(qǐng)求限速和異常降級(jí)策略長期運(yùn)行容易失效。上下文管理Agent 每次對(duì)話都會(huì)把完整歷史拼進(jìn) messages長對(duì)話下容易超出模型上下文窗口也缺少對(duì)歷史消息的裁剪與摘要壓縮。工具擴(kuò)展性工具注冊(cè)、參數(shù)校驗(yàn)和錯(cuò)誤處理都寫死在代碼里新增工具需要改動(dòng)多處維護(hù)成本較高。基于以上分析后續(xù)可以從以下幾個(gè)方向繼續(xù)完善工具管理框架將工具定義、注冊(cè)、執(zhí)行統(tǒng)一封裝為裝飾器或配置化機(jī)制降低新增工具的成本并統(tǒng)一處理參數(shù)校驗(yàn)與異常。記憶機(jī)制引入短期記憶會(huì)話內(nèi)摘要與長期記憶向量數(shù)據(jù)庫存儲(chǔ)歷史關(guān)鍵信息讓 Agent 在多輪對(duì)話中保持上下文連貫。RAG 集成將read_file工具升級(jí)為基于向量檢索的問答能力讓 Agent 能針對(duì)本地文檔進(jìn)行更精準(zhǔn)的檢索與回答。安全加固替換eval為安全計(jì)算庫為爬蟲增加重試與限速策略并對(duì)工具調(diào)用結(jié)果做長度截?cái)嗯c敏感信息過濾?？偟膩碚f本項(xiàng)目已經(jīng)搭建起一個(gè)功能完整的 Agent 雛形后續(xù)只要在安全、穩(wěn)定與智能化三個(gè)方向持續(xù)打磨就能逐步演進(jìn)為一個(gè)可投入實(shí)際業(yè)務(wù)使用的 AI 助手。