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