第1章:Function Calling 基础概念
1.1 什么是 Function Calling
| 概念名称 | 说明 | 注意事项 |
|---|
| Function Calling(函数调用) | 指大语言模型在生成回复过程中,主动请求调用预定义的外部函数(工具),以获取实时数据、执行操作或扩展能力。模型不直接执行代码,而是输出结构化调用指令,由开发者解析并执行后将结果反馈给模型。 | 模型本身不具备执行能力,仅生成调用意图;需开发者实现函数逻辑和结果回传机制。 |
| Tool / Function Definition(工具/函数定义) | 开发者向模型提供的函数元数据,包括函数名称、描述、参数格式(通常为 JSON Schema),用于指导模型何时及如何调用该函数。 | 参数必须严格遵循 JSON Schema 规范;描述应清晰准确,避免歧义,否则影响模型调用准确性。 |
| Structured Output(结构化输出) | Function Calling 要求模型输出符合特定格式(如 OpenAI 的 tool_calls 字段),而非自由文本,便于程序解析。 | 不同模型厂商的结构化格式不同(如 OpenAI vs Qwen),需适配各自 API 规范。 |
| Execution Loop(执行循环) | 典型交互流程:用户输入 → 模型决定是否调用函数 → 返回函数调用请求 → 开发者执行函数 → 将结果作为新消息传回模型 → 模型生成最终回答。 | 需正确维护对话历史中的角色(如 tool、assistant)和消息顺序,否则模型可能混淆上下文。 |
1.2 Function Calling 的典型应用场景
| 应用场景 | 说明 | 注意事项 |
|---|
| 实时信息查询 | 如天气、股价、新闻、航班状态等动态数据获取,模型通过调用外部 API 获取最新信息后再作答。 | 需确保 API 可用性与响应速度;敏感数据需做权限控制和脱敏处理。 |
| 数据库交互 | 模型根据自然语言生成 SQL 查询或调用封装好的数据库接口,实现非技术人员的数据访问。 | 必须对用户输入进行严格校验,防止 SQL 注入等安全风险;建议使用只读接口。 |
| 系统操作自动化 | 如发送邮件、创建日历事件、控制智能家居设备等,模型作为自然语言入口触发本地或远程操作。 | 涉及高危操作时需加入二次确认机制;建议在沙箱或受限环境中运行。 |
| 多步骤任务分解 | 复杂任务(如”帮我订一张明天去上海的最便宜机票并通知我”)被拆解为多个函数调用(查航班、比价、下单、发通知)。 | 需设计状态管理机制,跟踪任务进度;部分模型支持并行调用,可提升效率。 |
| 知识增强与事实校验 | 模型调用知识库或搜索引擎验证自身知识,避免幻觉(hallucination)。 | 返回结果需经过可信度过滤;避免过度依赖外部源导致延迟过高。 |
1.3 支持 Function Calling 的主流大模型对比
| 模型/平台 | 所属类别 | Function Calling 接口名称 | 调用方式特点 | 注意事项 |
|---|
| OpenAI GPT-3.5/4 Turbo | 闭源商业模型 | tools + tool_choice | 使用 tools 参数传入函数列表,模型返回 tool_calls;支持并行调用;需指定 role=tool 回传结果。 | 需使用 Chat Completions API;旧版 function_call 已弃用,应迁移到 tools 格式。 |
| Anthropic Claude 3 | 闭源商业模型 | tool_use(Beta) | 在 system prompt 中定义工具,模型通过 content 中的 tool_use 块请求调用;需手动构造工具结果消息。 | 目前处于 Beta 阶段,API 可能变动;文档更新较慢,需关注官方公告。 |
| 阿里通义千问 Qwen-Max / Qwen-Plus | 闭源商业模型 | tool_calls(兼容 OpenAI 格式) | 支持 OpenAI-style tools 定义;返回 message 中包含 tool_calls 字段;可通过 dashscope SDK 调用。 | 需开通 DashScope 服务;部分版本(如 Qwen-Turbo)可能不支持工具调用。 |
| Meta Llama 3(配合 LangChain / LlamaIndex) | 开源基础模型 | 无原生支持,需框架封装 | 本身不支持结构化函数调用,需借助 LangChain 的 ToolCallingAgent 或自定义输出解析器实现。 | 需自行处理输出格式解析和错误恢复;性能依赖本地部署环境。 |
| Mistral AI Mixtral / Mistral Large | 闭源/开源混合 | tool use(通过特定 prompt 引导) | 官方未提供标准工具调用协议,但可通过精心设计的 system prompt 和 JSON 输出格式模拟。 | 稳定性不如原生支持方案;需大量测试调优 prompt。 |
| Google Gemini 1.5 | 闭源商业模型 | Function Calling(via GenerativeModel) | 通过 generative_models.FunctionDeclaration 定义工具,模型返回 function_call;支持多函数并行。 | 需使用 Google AI Python SDK;部分区域可能不可用。 |
第2章:OpenAI Function Calling 详解(以 GPT 系列为例)
| 名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|
| tools(主参数) | List[Dict],每个元素为一个工具定义 | 向模型提供可调用的函数列表,用于指导其生成结构化调用请求 | [{"type": "function", "function": {"name": "get_weather", "description": "获取指定城市的天气", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}}] | 必须是列表;每个工具必须包含 type: "function" 和 function 字段 |
| function.name | 字符串 | 函数的唯一标识符,模型调用时使用此名称 | "get_weather" | 命名应简洁、无空格、符合编程命名规范;避免与内置关键词冲突 |
| function.description | 字符串 | 描述函数用途,帮助模型理解何时调用 | "获取指定城市的当前天气状况" | 描述越清晰准确,模型调用成功率越高;建议包含输入输出说明 |
| function.parameters | JSON Schema 对象 | 定义函数参数的结构、类型和约束 | {"type": "object", "properties": {"city": {"type": "string", "description": "城市名称"}}, "required": ["city"]} | 必须是 valid JSON Schema;推荐使用 object 类型;required 字段用于指定必填参数 |
| parameters.type | 字符串(如 "object") | 指定参数整体类型 | "object" | 目前仅支持 "object" 作为顶层类型;不支持 array 或 primitive 作为根类型 |
| parameters.properties | 对象 | 定义每个参数的名称、类型和描述 | {"city": {"type": "string", "description": "中文或英文城市名"}} | 所有参数必须在此声明;未声明的参数将被模型忽略 |
| parameters.required | 字符串列表 | 指定哪些参数为必填项 | ["city"] | 若用户未提供必填参数,模型可能无法正确调用;可结合 description 引导用户提供 |
2.2 调用流程与消息交互机制
| 步骤名称 | 操作细节 | 注意事项 |
|---|
| 初始化对话 | 构造 messages 列表,包含 user 消息,并传入 tools 参数到 chat.completions.create() | messages 必须以 user 或 system 开始;tools 不放入 messages,而是作为独立参数传入 |
| 模型决策 | 模型分析用户意图,若需调用函数,则在响应中返回 tool_calls 字段,包含函数名和参数 | 模型可能选择不调用任何函数,直接回答;也可能调用多个函数(并行) |
| 构造工具调用消息 | 将模型返回的 tool_calls 中每个调用解析为函数执行,并构造新的 message,role 设为 “tool”,content 为执行结果,tool_call_id 必须匹配 | tool_call_id 是唯一标识,必须原样回传;否则模型无法关联调用与结果 |
| 回传结果 | 将工具执行结果作为新消息追加到 messages 列表,再次调用 API | 新消息的 role 必须为 “tool”;content 应为字符串(即使是 JSON,也需 json.dumps() 转为字符串) |
| 获取最终回答 | 模型基于工具结果生成自然语言回复,通常 role 为 “assistant” 且无 tool_calls | 若任务未完成,模型可能再次请求调用其他函数,需循环处理直至无 tool_calls |
2.3 处理模型返回的 function_call 响应
注:自 GPT-4 Turbo(2023年11月后),OpenAI 已弃用 function_call 字段,统一使用 tool_calls。本节以当前标准 tool_calls 为准。
| 方法/字段 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|
| response.choices[0].message.tool_calls | 列表,每个元素含 id、type、function | 获取模型请求调用的所有工具 | tool_calls = response.choices[0].message.tool_calls | 若为空或 None,表示无需调用函数,可直接输出 content |
| tool_call.id | 字符串 | 唯一标识一次函数调用,用于结果回传 | tool_call_id = tool_call.id | 必须在回传消息中作为 tool_call_id 字段值,不可修改或省略 |
| tool_call.function.name | 字符串 | 被调用的函数名称 | func_name = tool_call.function.name | 需与本地注册的函数名匹配,建议使用字典映射函数对象 |
| tool_call.function.arguments | 字符串(JSON 格式) | 函数调用所需的参数,以 JSON 字符串形式提供 | args = json.loads(tool_call.function.arguments) | 必须用 json.loads() 解析;若格式错误会抛出异常,需 try-except 处理 |
| 构造 tool 消息 | {"role": "tool", "tool_call_id": "...", "content": "..."} | 将函数执行结果封装为模型可识别的消息 | {"role": "tool", "tool_call_id": "call_abc123", "content": '{"temperature": 22}'} | content 必须是字符串;即使返回数字或布尔值,也需转为字符串 |
| 错误处理 | 捕获函数执行异常,返回错误信息作为 content | 避免因函数崩溃中断对话流 | content = json.dumps({"error": "City not found"}) | 错误信息应结构化,便于模型理解失败原因并重试或解释 |
2.4 并行函数调用与多轮对话控制
| 概念/操作 | 说明 | 代码示例 | 注意事项 |
|---|
| 并行函数调用 | 模型在单次响应中请求调用多个独立函数(如同时查天气和新闻) | tool_calls 包含两个元素:[{"function": {"name": "get_weather"}}, {"function": {"name": "get_news"}}] | 所有函数必须互不依赖;若存在依赖关系(如 B 需 A 的结果),模型通常不会并行调用 |
| 并行执行 | 开发者可同时执行多个函数(如用 asyncio.gather)以提升效率 | results = await asyncio.gather(func1(), func2()) | 需确保函数线程/异步安全;避免共享状态导致竞态条件 |
| 多轮对话控制 | 通过循环检查是否仍有 tool_calls,持续交互直至模型给出最终回答 | while True: resp = client.chat.completions.create(...); if not resp.tool_calls: break | 必须设置最大轮数(如 max_turns=5)防止无限循环;记录完整 messages 历史 |
| 消息历史维护 | 每次调用后将模型消息和工具结果消息追加到 messages 列表 | messages.append(assistant_msg); messages.append(tool_msg) | 顺序必须严格保持:user → assistant(含 tool_calls)→ tool → assistant… |
| 工具调用终止条件 | 当模型返回的 message.tool_calls 为 None 或空列表,且 message.content 非空 | if not message.tool_calls and message.content: return message.content | 有时模型会先返回空 content 再调用工具,需综合判断;避免过早终止 |
| 控制调用行为 | 使用 tool_choice 参数强制模型调用特定函数或禁止调用 | tool_choice={"type": "function", "function": {"name": "get_weather"}} 或 "none" | "auto" 为默认;"none" 可用于仅获取模型思考而不执行操作的场景 |
第3章:通义千问(Qwen)Function Calling 实践
| 名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|
| DashScope SDK 调用入口 | from dashscope import Generation; Generation.call() | 调用 Qwen 模型的官方 Python SDK 接口 | response = Generation.call(model='qwen-max', messages=messages, tools=tools) | 必须安装 dashscope 包(pip install dashscope);需配置 DASHSCOPE_API_KEY 环境变量 |
| tools 参数结构 | List[Dict],兼容 OpenAI 格式 | 向 Qwen 模型传入可调用的工具定义列表 | tools = [{"type": "function", "function": {"name": "get_time", "description": "获取当前时间", "parameters": {"type": "object", "properties": {}, "required": []}}}] | Qwen 支持 OpenAI-style tools 定义,字段完全一致;无需额外转换 |
| 模型响应中的 tool_calls | response.output.choices[0].message.tool_calls | 获取模型请求调用的工具列表(若存在) | tool_calls = resp.output.choices[0].message.tool_calls | 若未触发调用,该字段为 None 或空列表;需先判断是否存在 |
| 消息角色(role) | 字符串:"user" / "assistant" / "tool" | 标识消息来源,在多轮对话中维持上下文 | {"role": "tool", "content": "...", "tool_call_id": "..."} | 回传工具结果时,role 必须为 "tool";否则模型无法识别 |
| tool_call_id 字段 | 字符串,由模型生成 | 唯一标识一次工具调用,用于结果绑定 | tool_call_id = tool_call.tool_call_id | 必须原样回传至新消息的 tool_call_id 字段,不可修改或省略 |
| 名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|
| function.name | 字符串 | 工具的唯一标识名 | "search_stock_price" | 命名应语义清晰,避免中文或特殊符号;建议使用下划线命名法 |
| function.description | 字符串 | 描述工具功能,指导模型调用时机 | "查询指定股票代码的最新收盘价" | 描述越具体,模型调用准确率越高;可包含示例输入 |
| function.parameters | JSON Schema 对象 | 定义工具所需参数的结构和类型 | {"type": "object", "properties": {"symbol": {"type": "string", "description": "股票代码,如 'AAPL'"}}, "required": ["symbol"]} | 必须为 object 类型;支持 string、number、boolean、array 等子类型 |
| parameters.properties | 对象 | 声明每个参数的名称、类型和说明 | {"city": {"type": "string", "description": "城市中文名"}} | 所有参数必须在此声明;未声明的参数将被忽略 |
| parameters.required | 字符串列表 | 指定必填参数 | ["symbol"] | 若用户未提供必填项,模型可能无法正确调用;可结合 description 引导补充 |
| 多参数工具定义 | 在 properties 中定义多个字段 | 支持复杂工具调用 | {"properties": {"origin": {"type": "string"}, "destination": {"type": "string"}}, "required": ["origin", "destination"]} | 参数间逻辑关系需在 description 中说明(如”出发地和目的地不能相同”) |
3.3 解析模型响应并执行本地函数
| 操作步骤 | 操作细节 | 注意事项 |
|---|
| 判断是否需要工具调用 | 检查 response.output.choices[0].message.tool_calls 是否非空 | 若为空且 message.content 非空,可直接返回内容作为最终回答 |
| 提取工具调用信息 | 遍历 tool_calls,获取 tool_call_id、function.name 和 arguments | arguments 为 JSON 字符串,需用 json.loads() 解析 |
| 映射函数执行 | 使用字典将函数名映射到本地 Python 函数对象 | func_map = {"get_weather": get_weather}; func = func_map.get(name) |
| 执行函数并捕获异常 | 调用函数并处理可能的运行时错误 | try: result = func(**args); except Exception as e: result = {"error": str(e)} |
| 构造工具响应消息 | 创建 role="tool" 的消息,包含 tool_call_id 和字符串化结果 | {"role": "tool", "tool_call_id": tid, "content": json.dumps(result, ensure_ascii=False)} |
| 追加消息并继续对话 | 将工具消息加入 messages 列表,再次调用模型 | messages.append(tool_message); next_resp = Generation.call(...) |
3.4 多工具协同与对话状态管理
| 概念/机制 | 说明 | 代码示例 | 注意事项 |
|---|
| 并行工具调用 | Qwen 可在单次响应中返回多个独立 tool_calls | tool_calls = [call1, call2]; results = [func1(), func2()] | 仅适用于无依赖关系的工具;若 B 依赖 A 结果,模型通常不会并行调用 |
| 工具执行顺序控制 | 开发者可根据依赖关系决定执行顺序 | if 'get_user_id' in calls: exec get_user_id first | 需自行分析工具间依赖;Qwen 不提供依赖图信息 |
| 对话状态跟踪 | 使用变量记录已完成的工具调用或任务进度 | state = {"weather_done": True, "stock_queried": False} | 可避免重复调用;适用于多步骤任务(如订票流程) |
| 最大轮数限制 | 设置对话最大交互次数防止死循环 | max_turns = 5; for i in range(max_turns): ... | 建议设为 3~5 轮;超过后可返回”任务超时”提示 |
| 消息历史截断 | 当 messages 过长时,保留关键上下文(如 system + 最近几轮) | keep = [sys_msg] + messages[-6:] | Qwen 模型有上下文长度限制(如 qwen-max 为 32768 tokens);需监控 token 使用量 |
| 工具禁用机制 | 在特定场景下临时移除某些工具 | tools = [t for t in all_tools if t['function']['name'] != 'delete_account'] | 可用于权限控制或安全限制;需动态构造 tools 列表 |
第4章:Function Calling 高级技巧
4.1 动态工具注册与上下文感知
| 名称/机制 | 操作细节 | 用途 | 代码示例 | 注意事项 |
|---|
| 动态工具注册 | 根据用户身份、会话状态或上下文实时构造 tools 列表 | 限制用户可调用的工具范围,提升安全性与相关性 | if user.role == 'admin': tools = [delete_user, query_log] else: tools = [get_profile] | 每次 API 调用前重新生成 tools;避免缓存导致权限泄露 |
| 上下文感知工具过滤 | 基于对话历史判断是否启用某工具 | 避免在不相关场景中触发无关工具(如未提”订票”时不显示航班查询) | if 'flight' in last_user_msg: tools.append(search_flights) | 需结合 NLP 关键词或嵌入相似度判断;避免过度过滤导致功能缺失 |
| 工具元数据扩展 | 在 function 定义中添加自定义字段(如 permissions、category) | 支持更复杂的调度逻辑 | {"name": "pay_bill", "x-permission": "finance", "description": "..."} | OpenAI/Qwen 不解析 x- 前缀字段,但开发者可在本地使用 |
| 会话级工具缓存 | 将已注册工具与 session_id 绑定,避免重复构造 | 提升性能,尤其在高频对话场景 | tool_registry[session_id] = build_tools(user_context) | 需设置过期时间,防止内存泄漏;用户登出时应清除 |
| 条件化工具描述 | 根据上下文动态调整 function.description | 引导模型在特定条件下调用 | desc = f"仅当用户询问{topic}时使用:查询最新政策" | 描述变化需显著,否则模型可能忽略;建议配合 system prompt 使用 |
4.2 错误处理与重试机制
| 操作步骤 | 操作细节 | 注意事项 |
|---|
| 函数执行异常捕获 | 使用 try-except 包裹本地函数调用 | 必须捕获所有异常(包括网络超时、参数错误等),避免中断主流程 |
| 结构化错误返回 | 将错误信息封装为 JSON 并转为字符串回传 | content = json.dumps({"error": "API timeout", "code": 504}) |
| 模型重试引导 | 在 system prompt 中说明”若工具返回 error,可请求用户澄清或换方式” | system: "若工具失败,请询问用户是否提供更多信息或尝试其他方法" |
| 自动重试机制 | 对临时性错误(如网络抖动)自动重试 1~2 次 | for i in range(2): try: result = api_call(); break; except: time.sleep(1) |
| 用户介入提示 | 当多次失败后,返回自然语言提示请求人工干预 | "抱歉,无法获取航班信息,请确认城市拼写或稍后再试" |
| 最大重试轮次控制 | 在对话循环中记录工具调用失败次数 | fail_count += 1; if fail_count > 2: break |
4.3 安全性与输入验证
| 安全机制 | 操作细节 | 注意事项 |
|---|
| 参数白名单校验 | 对 function.arguments 中的每个字段进行类型和值域检查 | if not isinstance(args['city'], str) or len(args['city']) > 50: raise ValueError |
| 敏感操作二次确认 | 对高危工具(如删除、支付)要求用户显式确认 | if func_name == 'delete_account': return "请回复'确认删除'以继续" |
| SQL 注入防护 | 数据库查询类工具必须使用参数化查询 | cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,)) |
| 外部 API 调用限流 | 对每个会话或用户 ID 限制工具调用频率 | rate_limiter.check(session_id, 'weather_api', max=5/min) |
| 输出脱敏处理 | 工具返回结果中移除敏感字段(如身份证、手机号) | result.pop('phone', None); return json.dumps(result) |
| 工具权限隔离 | 基于 RBAC 控制哪些用户可调用哪些工具 | if user.role not in tool.permissions: raise PermissionError |
4.4 性能优化与缓存策略
| 优化策略 | 操作细节 | 代码示例 | 注意事项 |
|---|
| 工具结果缓存 | 对幂等且低频变更的工具(如天气、汇率)缓存结果 | cache_key = f"weather_{city}"; if key in cache: return cache[key] | 设置合理 TTL(如天气缓存 10 分钟);避免返回过期数据 |
| 并行执行异步工具 | 使用 asyncio.gather 同时调用多个无依赖工具 | results = await asyncio.gather(get_weather(), get_news()) | 仅适用于 I/O 密集型操作;CPU 密集型任务应使用线程池 |
| Token 使用监控 | 计算 messages 总 token 数,接近上限时截断历史 | from tiktoken import encoding_for_model; enc = encoding_for_model('gpt-4'); total = sum(len(enc.encode(msg['content'])) for msg in messages) | Qwen 和 GPT 均有上下文长度限制;超限将报错或截断 |
| 轻量级工具优先 | 在多个可选工具中优先注册响应快、成本低的 | tools = [fast_search, detailed_search] → 模型通常选前者 | 可通过 description 引导:“优先使用 fast_search 获取概要” |
| 预热与连接复用 | 对频繁调用的外部 API 使用连接池或预热 | session = requests.Session(); session.get('https://api.example.com/health') | 减少 TCP 握手和 TLS 开销;尤其适用于微服务架构 |
| 模型调用批处理(非实时场景) | 将多个独立请求合并为批量推理(若平台支持) | DashScope 支持 batch 推理(部分模型) | 实时对话不适用;适用于后台任务或日志分析等离线场景 |
第5章:实战项目开发
5.1 构建天气查询助手
| 组件/步骤 | 操作细节 | 用途 | 代码示例 | 注意事项 |
|---|
| 工具定义(get_weather) | 定义 function.name=“get_weather”,参数为 city(string) | 允许模型请求获取指定城市天气 | {"type": "function", "function": {"name": "get_weather", "description": "获取城市当前天气", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}} | city 应支持中英文;可扩展为包含国家字段避免歧义 |
| 天气 API 封装 | 调用如 OpenWeatherMap 或和风天气 API | 获取真实天气数据 | def get_weather(city): resp = requests.get(f"https://api.qweather.com/v7/weather/now?location={city}&key=YOUR_KEY"); return resp.json() | 需处理 API 错误(如城市不存在);建议使用 location ID 而非名称提升准确性 |
| 参数校验与标准化 | 将用户输入的城市名标准化(如”北京”→“Beijing”) | 提高 API 调用成功率 | city_map = {"北京": "Beijing"}; city = city_map.get(city, city) | 可集成地理编码服务(Geocoding)进行模糊匹配 |
| 工具执行与结果构造 | 执行函数后将结果转为自然语言摘要并回传 | 供模型生成最终回答 | result = get_weather(city); content = f"当前{city}气温{result['temp']}℃,天气{result['text']}" | 回传 content 必须是字符串;避免直接返回原始 JSON |
| 对话循环控制 | 主循环处理 tool_calls 直至获得最终回答 | 实现完整交互流程 | while True: resp = call_model(messages, tools); if not resp.tool_calls: break; else: exec tools and append results | 设置最大轮数(如 3 轮),防止无限循环 |
| 错误友好提示 | 当天气 API 返回错误时提供清晰提示 | 提升用户体验 | if "error" in result: content = "未找到该城市,请确认名称是否正确" | 避免暴露内部错误码或堆栈信息 |
5.2 开发数据库查询机器人
| 组件/步骤 | 操作细节 | 用途 | 代码示例 | 注意事项 |
|---|
| 工具定义(query_db) | 定义 function.name=“query_db”,参数为 natural_language_query | 允许用户用自然语言提问数据库 | {"name": "query_db", "description": "根据自然语言查询公司销售数据库", "parameters": {"type": "object", "properties": {"query": {"type": "string", "description": "如'上月销售额最高的产品'"}}}} | 不直接暴露 SQL,而是让模型描述意图 |
| NL2SQL 转换器 | 使用 LLM 或规则引擎将自然语言转为安全 SQL | 防止直接执行用户输入 | prompt = f"将以下问题转为 SQL:{nl_query}"; sql = llm_call(prompt, schema=schema) | 必须限定 SELECT 权限;禁止 DML/DDL 操作 |
| 参数化查询执行 | 使用数据库驱动的参数化接口执行 SQL | 防止 SQL 注入 | cursor.execute("SELECT product, SUM(sales) FROM sales WHERE month = %s GROUP BY product ORDER BY SUM(sales) DESC LIMIT 1", (last_month,)) | 绝对禁止字符串拼接 SQL;即使来自模型也不可信 |
| 结果结构化返回 | 将数据库结果转为 JSON 并摘要 | 便于模型理解并生成回答 | rows = cursor.fetchall(); content = json.dumps([dict(zip(columns, row)) for row in rows]) | 若结果为空,应明确返回”未找到相关数据” |
| Schema 描述注入 | 在 system prompt 中提供表结构说明 | 提高 NL2SQL 准确率 | system: "sales 表含字段:product(产品名)、month(YYYY-MM)、sales(销售额)" | 仅暴露必要字段;敏感字段(如成本价)应隐藏 |
| 查询审计日志 | 记录每次 query_db 调用的原始问题和生成 SQL | 用于安全审计与调试 | logger.info(f"User: {user_id}, NL: {nl}, SQL: {sql}") | 日志不得包含敏感数据;建议脱敏后存储 |
5.3 集成外部 API 的智能客服系统
| 组件/步骤 | 操作细节 | 用途 | 代码示例 | 注意事项 |
|---|
| 多工具注册 | 注册订单查询、物流跟踪、退款申请等工具 | 覆盖客服常见场景 | tools = [get_order_status, track_shipment, request_refund] | 每个工具需有清晰职责边界;避免功能重叠 |
| 用户身份绑定 | 从会话上下文提取 user_id 并透传至工具 | 实现个性化服务 | def get_order_status(user_id=None, order_id=None): ... | 需验证 user_id 合法性;防止越权访问他人订单 |
| 状态机控制对话流 | 使用有限状态机管理多步骤任务(如退款流程) | 确保操作合规 | state = "awaiting_reason"; if state == "awaiting_reason" and "reason" in args: proceed_to_next_step() | 状态应持久化(如存入 Redis);支持中断后恢复 |
| 敏感操作确认 | 退款、修改地址等操作需用户二次确认 | 降低误操作风险 | if func_name == "request_refund": return "请回复'确认退款'以继续处理订单 {order_id}" | 确认必须作为新 user 消息进入下一轮 |
| 外部 API 熔断机制 | 当第三方服务连续失败时临时禁用对应工具 | 提升系统鲁棒性 | if failure_count > 3: disable_tool("track_shipment") for 5 minutes | 可结合 circuit breaker 模式实现 |
| 多语言支持 | 根据用户语言返回对应语言的客服响应 | 提升国际化体验 | content = translate(result, target_lang=user.lang) | 翻译应在最终输出前进行;工具内部保持英文/结构化 |
5.4 多智能体协作中的工具调用
| 概念/机制 | 操作细节 | 用途 | 代码示例 | 注意事项 |
|---|
| 智能体角色定义 | 为每个 Agent 分配专属工具集(如 Planner、Executor、Reviewer) | 实现分工协作 | planner_tools = [decompose_task]; executor_tools = [call_api, run_code] | 工具不应跨角色共享,除非明确设计为公共能力 |
| 工具作为通信媒介 | Agent A 调用工具生成中间结果,Agent B 读取该结果继续处理 | 实现跨 Agent 数据传递 | planner calls plan_task → result stored in shared context → executor reads plan | 需设计共享上下文存储(如 dict 或消息队列) |
| 协作消息协议 | 定义 Agent 间消息格式(含 sender、intent、payload) | 规范交互行为 | {"from": "planner", "to": "executor", "action": "execute_plan", "data": {...}} | 可复用 tool_call 结构,但需扩展路由字段 |
| 主控调度器 | 中央控制器决定下一由哪个 Agent 响应 | 避免多 Agent 同时发言 | scheduler.next_agent = "reviewer" if last_output.needs_review else "user" | 调度逻辑可基于规则或小型 LLM 判断 |
| 工具调用链追踪 | 记录完整调用链路(Agent → Tool → Result → Next Agent) | 用于调试与审计 | trace_log.append({"agent": "executor", "tool": "search_web", "result": "...", "next": "reviewer"}) | 日志应包含时间戳和唯一会话 ID |
| 循环检测与终止 | 检测 Agent 间重复调用同一工具或无限协商 | 防止死锁 | if current_plan == last_two_plans[0] == last_two_plans[1]: break | 可设置最大协作轮数(如 8 轮)强制终止 |