1. 项目概述:当AI学会"动手动脚"
去年在开发一个智能客服系统时,我遇到了一个典型问题:当用户询问"帮我查下订单12345的物流状态"时,AI只能礼貌地回复"我理解您想查询物流,但很抱歉我无法直接操作系统"。这种无力感促使我开始研究Function Calling技术——它能让大语言模型不再只是"动嘴",而是真正具备"动手"能力。
Function Calling本质上是大模型与外部工具的连接协议。通过这项技术,AI可以:
- 理解用户请求中的操作意图(如"查物流")
- 自动生成规范的API调用参数
- 将原始API响应转化为自然语言回复
以DeepSeek/OpenAI为代表的现代大模型,其Function Calling能力已经可以处理复杂的工作流。比如用户说"下周二下午3点约张总在星巴克开会",AI能自动完成:
- 解析时间语义(下周二15:00)
- 查询联系人数据库(张总)
- 调用日历API创建事件
- 预订会议室/场地
- 发送确认邮件
这种"思考-行动-反馈"的闭环,正是AI智能体(Agent)的核心能力。本文将基于最新DeepSeek API(兼容OpenAI格式),手把手实现一个具备真实操作能力的AI助手。
2. 环境准备与API配置
2.1 开发环境搭建
推荐使用Python 3.10+环境,这是目前大模型开发最稳定的版本。我习惯用conda管理环境:
bash复制conda create -n ai_agent python=3.10
conda activate ai_agent
核心依赖库安装(注意版本兼容性):
bash复制pip install openai==1.12.0 httpx==0.25.0 python-dotenv==1.0.0
重要提示:不要直接
pip install openai,新版本(v1.0+)的API设计与旧版(v0.28)有重大变化。本文示例基于v1.12.0编写。
2.2 获取API密钥
DeepSeek目前提供与OpenAI兼容的API服务,申请流程:
- 访问DeepSeek官网
- 注册账号并完成企业认证(个人开发者也可申请)
- 在控制台"API Keys"页面创建新密钥
将密钥保存在项目根目录的.env文件中:
ini复制DEEPSEEK_API_KEY=sk-your-key-here
BASE_URL=https://api.deepseek.com
2.3 初始化客户端
创建client.py作为基础模块:
python复制import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(
api_key=os.getenv('DEEPSEEK_API_KEY'),
base_url=os.getenv('BASE_URL')
)
测试连接(使用最新deepseek-v4-pro模型):
python复制response = client.chat.completions.create(
model="deepseek-v4-pro",
messages=[{"role": "user", "content": "Hello"}]
)
print(response.choices[0].message.content)
3. Function Calling核心实现
3.1 定义工具函数
我们先实现一个查询天气的示例。创建weather.py:
python复制import httpx
from typing import Literal
async def get_current_weather(
location: str,
unit: Literal['celsius', 'fahrenheit'] = 'celsius'
) -> dict:
"""获取指定城市的实时天气
Args:
location: 城市名称(如"北京")
unit: 温度单位(摄氏度/华氏度)
Returns:
{
"location": "北京",
"temperature": 25,
"unit": "celsius",
"forecast": ["晴朗", "微风"]
}
"""
# 这里使用模拟数据,实际项目可接入气象API
return {
"location": location,
"temperature": 25 if unit == 'celsius' else 77,
"unit": unit,
"forecast": ["晴朗", "微风"]
}
关键设计要点:
- 使用Python类型注解(
Literal)明确参数可选值 - 函数docstring要详细描述输入输出,这将成为AI理解功能的依据
- 实际项目中建议添加错误处理和缓存机制
3.2 工具注册与调用
创建agent.py实现核心逻辑:
python复制from typing import List, Dict, Any
import inspect
import json
from client import client
tools = [{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "获取指定城市的当前天气情况",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市名称,如'北京市'"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
"required": ["location"]
}
}
}]
async def run_conversation(messages: List[Dict[str, Any]]) -> str:
# 首次调用,让模型决定是否使用工具
response = client.chat.completions.create(
model="deepseek-v4-pro",
messages=messages,
tools=tools,
tool_choice="auto"
)
tool_calls = response.choices[0].message.tool_calls
if not tool_calls:
return response.choices[0].message.content
# 准备工具调用
available_functions = {
"get_current_weather": get_current_weather
}
messages.append(response.choices[0].message)
# 执行所有需要的工具调用
for tool_call in tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
function_to_call = available_functions[function_name]
function_response = await function_to_call(**function_args)
messages.append({
"tool_call_id": tool_call.id,
"role": "tool",
"name": function_name,
"content": json.dumps(function_response)
})
# 第二次调用,让模型总结工具返回结果
second_response = client.chat.completions.create(
model="deepseek-v4-pro",
messages=messages
)
return second_response.choices[0].message.content
3.3 实战测试
创建main.py进行端到端测试:
python复制import asyncio
from agent import run_conversation
async def main():
messages = [{
"role": "user",
"content": "北京现在天气怎么样?用摄氏度表示"
}]
response = await run_conversation(messages)
print("AI回复:", response)
asyncio.run(main())
预期输出:
code复制AI回复: 北京当前天气晴朗,有微风,气温25摄氏度。
4. 高级应用场景
4.1 多工具协同工作
现实场景往往需要组合多个工具。例如电商客服系统可能需要:
python复制tools = [
{ # 订单查询
"type": "function",
"function": {
"name": "get_order_status",
"description": "查询订单状态及物流信息",
"parameters": {...}
}
},
{ # 退款申请
"type": "function",
"function": {
"name": "create_refund",
"description": "发起退款流程",
"parameters": {...}
}
},
{ # 人工转接
"type": "function",
"function": {
"name": "transfer_to_human",
"description": "转接人工客服",
"parameters": {...}
}
}
]
当用户说"我的订单12345还没收到,想退款",AI会自动:
- 调用
get_order_status确认物流状态 - 如果确实超期,调用
create_refund发起退款 - 遇到复杂情况时调用
transfer_to_human
4.2 动态参数验证
在工具函数中添加参数校验逻辑:
python复制async def create_refund(
order_id: str,
reason: str,
amount: float = None
):
if not order_id.startswith('DD-') and len(order_id) != 12:
raise ValueError("订单号格式错误")
if amount and amount <= 0:
raise ValueError("退款金额必须大于0")
# 实际业务逻辑...
在run_conversation中捕获异常并让AI重新提问:
python复制try:
function_response = await function_to_call(**function_args)
except Exception as e:
messages.append({
"role": "tool",
"content": f"Error: {str(e)}",
"name": function_name
})
# 让AI处理错误
return await run_conversation(messages)
4.3 流式响应优化
对于耗时操作,使用流式传输提升用户体验:
python复制stream = client.chat.completions.create(
model="deepseek-v4-pro",
messages=messages,
tools=tools,
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.tool_calls:
# 处理工具调用
elif chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="")
5. 避坑指南与性能优化
5.1 常见问题排查
-
工具不被调用
- 检查函数描述是否清晰(AI靠description理解功能)
- 确认用户提问包含足够触发信息(如"天气"触发天气查询)
- 测试时设置
tool_choice={"type": "function", "function": {"name": "xxx"}}强制调用
-
参数解析错误
- 确保parameters定义与函数签名一致
- 复杂参数使用
$ref定义JSON Schema - 在API测试平台验证参数结构
-
认证失败
- 检查BASE_URL是否正确(DeepSeek使用专用端点)
- 确认API密钥未过期(控制台显示剩余额度)
- 企业账号需绑定IP白名单
5.2 性能优化技巧
-
上下文管理
- 对长时间会话,定期总结上下文避免token超限
- 使用
context_length=8192参数(DeepSeek支持8k上下文)
-
缓存策略
python复制from functools import lru_cache @lru_cache(maxsize=100) async def get_weather(location: str): # 实现带缓存的查询 -
批量处理
python复制# 同时处理多个工具调用 tasks = [ function_to_call(**json.loads(tool.function.arguments)) for tool in tool_calls ] await asyncio.gather(*tasks)
5.3 安全注意事项
-
输入过滤
python复制import html def sanitize_input(text: str) -> str: return html.escape(text.strip()) -
权限控制
- 为不同功能设置权限等级
- 敏感操作需二次确认:
python复制if "删除" in user_input: confirm = await ask_user("确认要删除数据吗?") if not confirm: return -
API调用限制
python复制from tenacity import retry, stop_after_attempt @retry(stop=stop_after_attempt(3)) async def call_api(): # 带重试的逻辑
6. 项目扩展思路
6.1 与现有系统集成
-
企业微信/钉钉机器人
- 通过webhook接收消息
- 处理后将结果以卡片消息形式返回
-
数据库对接
python复制from sqlalchemy import create_engine engine = create_engine("postgresql://user:pass@localhost/db") @tool def query_db(query: str) -> list: with engine.connect() as conn: return [dict(row) for row in conn.execute(text(query))] -
低代码平台
- 将工具函数封装为可视化节点
- 通过拖拽构建AI工作流
6.2 监控与数据分析
-
日志记录
python复制import logging logging.basicConfig( filename='ai_agent.log', level=logging.INFO, format='%(asctime)s - %(message)s' ) def log_interaction(user_input, ai_response): logging.info(f"Q: {user_input} | A: {ai_response}") -
效果评估
- 记录工具调用准确率
- 人工标注bad case用于模型微调
-
成本分析
python复制def calculate_cost(response): input_tokens = response.usage.prompt_tokens output_tokens = response.usage.completion_tokens # DeepSeek定价: $0.01/千token return (input_tokens + output_tokens) * 0.00001
在实际项目中,我发现最影响用户体验的不是技术实现,而是交互设计。比如当AI需要更多信息时,与其直接说"请提供订单号",不如说"为了帮您查询订单,需要您提供订单号(通常以DD开头)"。这种细节处理能让AI显得更专业、更人性化。
