一、任务安排天核心任务产出目标完成标准Day1阅读 FC 文档、理解完整流程、创建代码文件、定义 3 个工具 JSON Schema理解 FC 机制能说清 "模型不执行代码，只决定调什么"Day2定义 inventory/order/notification 工具 Schema、实现 mock 执行逻辑、API 传入 tools 参数工具定义完成API 返回的 tool_calls 能正确解析Day3实现工具调度逻辑、回传 tool 执行结果、二次调用 API 获取最终回答单工具调用跑通"查 P001 库存"→ 调工具 → 自然语言回答Day4测试无工具调用场景、验证参数提取、添加工具调用日志边界场景处理不该调工具时不调，参数提取正确Day5周五演示、整理代码注释周五演示完成演示库存查询的完整 FC 流程二、核心代码# ================================================================ # Week 9 Day 1: Function Calling — 原理 + 工具调用 # ================================================================ # # API: 火山引擎 (ark.cn-beijing.volces.com) # 模型: deepseek-v3-2-251201 # # Function Calling 工作原理 (一句话): # 模型不执行代码，它只输出"我要调哪个函数，参数是什么"， # 由你的代码真正执行函数，再把结果还给模型，模型最终整合成自然语言回答。 # # 完整流程: # ┌──────────┐ ┌─────────────┐ ┌──────────┐ ┌─────────────┐ # │ 用户提问 │ ──→ │ LLM 返回 │ ──→ │ 代码执行 │ ──→ │ 结果回传LLM │ ──→ 最终回答 # │ │ │ tool_calls │ │ 真实函数 │ │ role:"tool" │ # └──────────┘ └─────────────┘ └──────────┘ └─────────────┘ # # 3个工具: # query_inventory — 查询商品库存 (product_id + warehouse) # create_order — 创建采购订单 (product_id + quantity + address) # send_notification — 发送通知 (recipient + message) # # 运行: # python 01_function_calling.py # # 依赖: # pip install httpx python-dotenv # ================================================================ import os import sys import json import time from datetime import datetime from pathlib import Path import httpx from dotenv import load_dotenv # Windows GBK 控制台兼容: 强制 UTF-8 输出 if sys.platform == "win32": sys.stdout.reconfigure(encoding="utf-8", errors="replace") # 加载 .env (当前目录优先，fallback 到 phase2 目录) env_path = os.path.join(os.path.dirname(__file__), ".env") if not os.path.exists(env_path): env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "phase2-deepseek-api-call", ".env") load_dotenv(env_path) # 火山引擎 API 配置 API_KEY = os.getenv("VOLCENGINE_API_KEY", "") BASE_URL = os.getenv("VOLCENGINE_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3").rstrip("/") MODEL = "deepseek-v3-2-251201" # 日志目录 LOG_DIR = Path(__file__).resolve().parent / "logs" LOG_DIR.mkdir(parents=True, exist_ok=True) # ================================================================ # 工具调用日志 (记录每次工具调用的 name、args、result) # ================================================================ class ToolCallLogger: """结构化记录每次工具调用，支持实时写入 JSON""" def __init__(self, session_name: str = None): self.session_id = session_name or datetime.now().strftime("tools_%Y%m%d_%H%M%S") self.calls: list[dict] = [] self._log_path = LOG_DIR / f"{self.session_id}.json" def record(self, name: str, args: dict, result: str, turn: int = 1, elapsed_ms: float = 0) - dict: """记录一次工具调用""" entry = { "timestamp": datetime.now().isoformat(), "turn": turn, "tool_name": name, "arguments": args, "result": json.loads(result) if isinstance(result, str) else result, "elapsed_ms": round(elapsed_ms, 1), } self.calls.append(entry) self._flush() return entry def _flush(self): """持久化到 JSON 文件""" with open(self._log_path, "w", encoding="utf-8") as f: json.dump({ "session_id": self.session_id, "total_calls": len(self.calls), "calls": self.calls, }, f, ensure_ascii=False, indent=2) def summary_table(self) - str: """生成调用摘要表格""" if not self.calls: return " (无工具调用)" sep = '-' * 22 + '┬' + '-' * 3
（十二）Function Calling原理与单工具调用
一、任务安排天核心任务产出目标完成标准Day1阅读 FC 文档、理解完整流程、创建代码文件、定义 3 个工具 JSON Schema理解 FC 机制能说清 "模型不执行代码，只决定调什么"Day2定义 inventory/order/notification 工具 Schema、实现 mock 执行逻辑、API 传入 tools 参数工具定义完成API 返回的 tool_calls 能正确解析Day3实现工具调度逻辑、回传 tool 执行结果、二次调用 API 获取最终回答单工具调用跑通"查 P001 库存"→ 调工具 → 自然语言回答Day4测试无工具调用场景、验证参数提取、添加工具调用日志边界场景处理不该调工具时不调，参数提取正确Day5周五演示、整理代码注释周五演示完成演示库存查询的完整 FC 流程二、核心代码# ================================================================ # Week 9 Day 1: Function Calling — 原理 + 工具调用 # ================================================================ # # API: 火山引擎 (ark.cn-beijing.volces.com) # 模型: deepseek-v3-2-251201 # # Function Calling 工作原理 (一句话): # 模型不执行代码，它只输出"我要调哪个函数，参数是什么"， # 由你的代码真正执行函数，再把结果还给模型，模型最终整合成自然语言回答。 # # 完整流程: # ┌──────────┐ ┌─────────────┐ ┌──────────┐ ┌─────────────┐ # │ 用户提问 │ ──→ │ LLM 返回 │ ──→ │ 代码执行 │ ──→ │ 结果回传LLM │ ──→ 最终回答 # │ │ │ tool_calls │ │ 真实函数 │ │ role:"tool" │ # └──────────┘ └─────────────┘ └──────────┘ └─────────────┘ # # 3个工具: # query_inventory — 查询商品库存 (product_id + warehouse) # create_order — 创建采购订单 (product_id + quantity + address) # send_notification — 发送通知 (recipient + message) # # 运行: # python 01_function_calling.py # # 依赖: # pip install httpx python-dotenv # ================================================================ import os import sys import json import time from datetime import datetime from pathlib import Path import httpx from dotenv import load_dotenv # Windows GBK 控制台兼容: 强制 UTF-8 输出 if sys.platform == "win32": sys.stdout.reconfigure(encoding="utf-8", errors="replace") # 加载 .env (当前目录优先，fallback 到 phase2 目录) env_path = os.path.join(os.path.dirname(__file__), ".env") if not os.path.exists(env_path): env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "phase2-deepseek-api-call", ".env") load_dotenv(env_path) # 火山引擎 API 配置 API_KEY = os.getenv("VOLCENGINE_API_KEY", "") BASE_URL = os.getenv("VOLCENGINE_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3").rstrip("/") MODEL = "deepseek-v3-2-251201" # 日志目录 LOG_DIR = Path(__file__).resolve().parent / "logs" LOG_DIR.mkdir(parents=True, exist_ok=True) # ================================================================ # 工具调用日志 (记录每次工具调用的 name、args、result) # ================================================================ class ToolCallLogger: """结构化记录每次工具调用，支持实时写入 JSON""" def __init__(self, session_name: str = None): self.session_id = session_name or datetime.now().strftime("tools_%Y%m%d_%H%M%S") self.calls: list[dict] = [] self._log_path = LOG_DIR / f"{self.session_id}.json" def record(self, name: str, args: dict, result: str, turn: int = 1, elapsed_ms: float = 0) - dict: """记录一次工具调用""" entry = { "timestamp": datetime.now().isoformat(), "turn": turn, "tool_name": name, "arguments": args, "result": json.loads(result) if isinstance(result, str) else result, "elapsed_ms": round(elapsed_ms, 1), } self.calls.append(entry) self._flush() return entry def _flush(self): """持久化到 JSON 文件""" with open(self._log_path, "w", encoding="utf-8") as f: json.dump({ "session_id": self.session_id, "total_calls": len(self.calls), "calls": self.calls, }, f, ensure_ascii=False, indent=2) def summary_table(self) - str: """生成调用摘要表格""" if not self.calls: return " (无工具调用)" sep = '-' * 22 + '┬' + '-' * 3
