企业官网建设流程全解析

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

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 助手。