Function Calling 模式选择指南同步、异步和流式的场景决策树一、从能调用到会调用Function Calling 的模式选择困境2026 年上半年某 AI 助手平台在处理用户请求时遇到了性能瓶颈平均响应时间 8 秒用户流失率激增。分析原因发现他们的 Function Calling 实现采用了单一的同步模式导致所有工具调用都必须等待前一个完成。经过重构引入异步和流式模式后平均响应时间降到 1.5 秒用户体验显著提升。关键是不同的场景需要不同的调用模式。本文将系统对比 Function Calling 的三大模式并给出场景选择决策树。二、模式一同步 Function Calling最常用适用场景典型场景工具执行时间短 100ms需要保证执行顺序工具之间有依赖关系示例查询订单状态、获取用户信息、简单计算实现与优化from typing import Dict, List, Any import json import time class SyncFunctionCaller: 同步 Function Calling 执行器 def __init__(self): self.tools self._register_tools() def _register_tools(self) - Dict: 注册工具 return { get_user: self.get_user, get_order: self.get_order, calculate_discount: self.calculate_discount, } def execute(self, tool_name: str, params: Dict) - Any: 同步执行工具 if tool_name not in self.tools: raise ValueError(fTool {tool_name} not found) tool_func self.tools[tool_name] # 执行工具 start time.time() result tool_func(**params) elapsed time.time() - start # 记录执行时间 print(fTool {tool_name} executed in {elapsed:.3f}s) return result def execute_with_llm(self, user_query: str) - str: 配合大模型执行 Function Calling # 1. 调用大模型获取工具调用计划 response self._call_llm(user_query) # 2. 解析工具调用 tool_calls self._parse_tool_calls(response) # 3. 同步执行所有工具按顺序 results [] for tool_call in tool_calls: result self.execute(tool_call[name], tool_call[parameters]) results.append(result) # 4. 将结果返回给大模型生成最终回答 final_response self._call_llm_with_results(user_query, results) return final_response def _call_llm(self, query: str) - str: 调用大模型简化 # 实际应该调用 OpenAI API 等 return json.dumps({ tool_calls: [ {name: get_user, parameters: {user_id: 123}}, {name: get_order, parameters: {order_id: 456}}, ] }) def _parse_tool_calls(self, response: str) - List[Dict]: 解析工具调用 data json.loads(response) return data.get(tool_calls, []) def _call_llm_with_results(self, query: str, results: List[Any]) - str: 将工具执行结果传给大模型 prompt f 用户问题: {query} 工具执行结果: {json.dumps(results, indent2, ensure_asciiFalse)} 请基于以上结果回答用户问题。 return self._call_llm(prompt) # 工具实现 def get_user(self, user_id: str) - Dict: 获取用户信息模拟 time.sleep(0.05) # 50ms return {id: user_id, name: 张三, level: VIP} def get_order(self, order_id: str) - Dict: 获取订单信息模拟 time.sleep(0.08) # 80ms return {id: order_id, amount: 99.9, status: shipped} def calculate_discount(self, amount: float, user_level: str) - float: 计算折扣模拟 time.sleep(0.01) # 10ms if user_level VIP: return amount * 0.9 return amount # 使用示例 def main(): caller SyncFunctionCaller() # 用户提问我是 VIP订单 456 多少钱 user_query 我是 VIP订单 456 多少钱 answer caller.execute_with_llm(user_query) print(f回答: {answer}) # 问题如果工具执行时间是 5 秒 5 秒 10 秒用户要等很久同步模式的性能问题解决方案并发执行无依赖的工具# 优化并发执行无依赖的工具 def execute_parallel(self, tool_calls: List[Dict]) - List[Any]: 并发执行工具无依赖 import concurrent.futures with concurrent.futures.ThreadPoolExecutor(max_workers10) as executor: futures [] for tool_call in tool_calls: future executor.submit( self.execute, tool_call[name], tool_call[parameters] ) futures.append(future) results [f.result() for f in futures] return results # 效果5 秒 5 秒并行 5 秒三、模式二异步 Function Calling适合长时间任务适用场景典型场景工具执行时间长 3 秒用户不需要立即得到结果可以接受任务已提交的回复示例生成报告、批量处理数据、调用慢速 API实现与优化from typing import Dict, List, Any import asyncio import uuid import time from datetime import datetime class AsyncFunctionCaller: 异步 Function Calling 执行器 def __init__(self): self.tasks {} # 存储任务状态 self.tools self._register_tools() def _register_tools(self) - Dict: return { generate_report: self.generate_report, batch_process: self.batch_process, } async def execute_async(self, tool_name: str, params: Dict) - str: 异步执行工具返回任务 ID task_id str(uuid.uuid4()) # 创建任务 self.tasks[task_id] { status: running, created_at: datetime.now(), result: None, error: None } # 启动异步任务 asyncio.create_task(self._execute_and_update(task_id, tool_name, params)) return task_id async def _execute_and_update(self, task_id: str, tool_name: str, params: Dict): 执行工具并更新状态 try: tool_func self.tools[tool_name] result await tool_func(**params) self.tasks[task_id][status] completed self.tasks[task_id][result] result except Exception as e: self.tasks[task_id][status] failed self.tasks[task_id][error] str(e) def get_task_status(self, task_id: str) - Dict: 查询任务状态 if task_id not in self.tasks: return {error: Task not found} task self.tasks[task_id] return { task_id: task_id, status: task[status], created_at: task[created_at].isoformat(), result: task[result], error: task[error] } # 异步工具实现 async def generate_report(self, user_id: str, report_type: str) - Dict: 生成报告模拟耗时操作 await asyncio.sleep(10) # 模拟 10 秒 return { report_id: report_123, user_id: user_id, type: report_type, download_url: https://example.com/report_123.pdf } async def batch_process(self, items: List[str]) - Dict: 批量处理模拟 results [] for item in items: await asyncio.sleep(0.5) # 每个 item 耗时 0.5 秒 results.append(fProcessed: {item}) return {processed: len(results), results: results} # 配合大模型使用 class AsyncFunctionCallingWithLLM: 带异步 Function Calling 的 LLM 系统 def __init__(self): self.caller AsyncFunctionCaller() async def handle_query(self, user_query: str) - Dict: 处理用户查询 # 1. 调用大模型判断是否要异步执行 response await self._call_llm(user_query) if response[type] async: # 异步执行 tool_calls response[tool_calls] task_ids [] for tool_call in tool_calls: task_id await self.caller.execute_async( tool_call[name], tool_call[parameters] ) task_ids.append(task_id) return { type: async, message: 任务已提交请稍后查询, task_ids: task_ids, query_url: f/api/tasks?ids{,.join(task_ids)} } else: # 同步执行 return { type: sync, answer: response[answer] } async def _call_llm(self, query: str) - Dict: 调用大模型简化 # 实际应该根据 query 判断是否需要异步 if 生成报告 in query or 批量处理 in query: return { type: async, tool_calls: [ {name: generate_report, parameters: {user_id: 123, report_type: monthly}} ] } else: return { type: sync, answer: 同步回答 } # 使用示例 async def main(): system AsyncFunctionCallingWithLLM() # 用户提问帮我生成上个月的报告 response await system.handle_query(帮我生成上个月的报告) print(f响应: {response}) # 输出: {type: async, message: 任务已提交请稍后查询, ...} # 轮询任务状态 task_id response[task_ids][0] while True: status system.caller.get_task_status(task_id) print(f任务状态: {status[status]}) if status[status] in [completed, failed]: print(f结果: {status.get(result) or status.get(error)}) break await asyncio.sleep(2) # 运行 # asyncio.run(main())四、模式三流式 Function Calling最佳用户体验适用场景典型场景工具执行时间长但能产生中间结果用户需要实时看到进度生成的内容是流式的如长文本生成示例流式生成文章、逐步展示数据分析结果实现与优化from typing import Generator, Dict import json class StreamingFunctionCaller: 流式 Function Calling 执行器 def __init__(self): self.tools self._register_tools() def execute_streaming(self, tool_name: str, params: Dict) - Generator[Dict, None, None]: 流式执行工具yield 中间结果 if tool_name not in self.tools: raise ValueError(fTool {tool_name} not found) tool_func self.tools[tool_name] # 执行工具应该是生成器函数 yield from tool_func(**params) def _register_tools(self) - Dict: return { stream_generate_article: self.stream_generate_article, stream_data_analysis: self.stream_data_analysis, } # 流式工具实现 def stream_generate_article(self, topic: str, max_length: int 1000) - Generator[Dict, None, None]: 流式生成文章 # 模拟流式生成 paragraphs [ f# {topic}\n\n, ## 引言\n\n这是引言部分...\n\n, ## 正文\n\n这是正文部分逐步生成...\n\n, ## 结论\n\n这是结论部分...\n\n, ] for i, para in enumerate(paragraphs): # 模拟生成延迟 import time time.sleep(0.5) yield { type: progress, progress: (i 1) / len(paragraphs) * 100, content: para, is_final: (i len(paragraphs) - 1) } def stream_data_analysis(self, data_source: str) - Generator[Dict, None, None]: 流式数据分析 # 模拟数据分析步骤 steps [ {step: load_data, message: 加载数据中...}, {step: clean_data, message: 清洗数据中...}, {step: analyze, message: 分析中...}, {step: visualize, message: 生成图表中...}, ] for step_info in steps: import time time.sleep(1) yield { type: step, step: step_info[step], message: step_info[message], progress: 25 # 每步 25% } # 最终返回结果 yield { type: result, data: { summary: 数据分析完成, chart_url: https://example.com/chart.png }, is_final: True } # 配合 Web 服务器使用SSE from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() caller StreamingFunctionCaller() app.post(/api/stream_function) async def stream_function(tool_name: str, params: Dict): 流式 Function Calling API def generate(): for chunk in caller.execute_streaming(tool_name, params): # SSE 格式 yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n return StreamingResponse(generate(), media_typetext/event-stream) # 前端接收流式响应 async function callStreamFunction() { const response await fetch(/api/stream_function, { method: POST, body: JSON.stringify({ tool_name: stream_generate_article, params: {topic: AI 发展趋势} }) }); const reader response.body.getReader(); const decoder new TextDecoder(); while (true) { const {done, value} await reader.read(); if (done) break; const chunk decoder.decode(value); const lines chunk.split(\\n); for (const line of lines) { if (line.startsWith(data: )) { const data JSON.parse(line.slice(6)); if (data.type progress) { // 更新进度条 updateProgressBar(data.progress); // 显示内容 appendContent(data.content); } } } } } 五、总结Function Calling 模式选择决策树选择同步模式✅ 工具执行快 100ms✅ 工具之间有依赖✅ 需要保证顺序❌ 不适合长时间任务选择异步模式✅ 工具执行慢 3s✅ 用户可以接受延迟✅ 需要执行后台任务❌ 不适合需要即时反馈选择流式模式✅ 工具能产生中间结果✅ 用户体验优先✅ 生成内的长内容❌ 实现复杂度高混合使用策略先调用大模型判断任务类型短时间任务 → 同步长时间任务 → 异步生成任务 → 流式性能优化建议无依赖的工具并发执行设置工具超时避免无限等待流式响应设置心跳避免超时异步任务要持久化避免重启丢失记住用户体验优先技术选型服务于体验。下一篇我们将深入探讨 Python 异步编程的选型矩阵。