Function Calling的核心价值

Function Calling让LLM能够调用外部工具和API，从"对话助手"升级为"行动助手"。但实践中的挑战在于：如何让模型可靠地选择正确的工具、生成正确的参数、处理工具执行失败的情况。

工具定义最佳实践

清晰的工具描述

tools = [
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": (
                "搜索企业知识库中的文档。当用户询问公司政策、产品信息、"
                "技术文档等内部知识时使用此工具。\n"
                "注意：不要用于搜索公开互联网信息。"
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词，使用自然语言描述要查找的内容"
                    },
                    "department": {
                        "type": "string",
                        "enum": ["engineering", "sales", "hr", "finance"],
                        "description": "限定搜索的部门范围，不指定则搜索全部"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "返回结果数量，默认5",
                        "default": 5,
                        "minimum": 1,
                        "maximum": 20
                    }
                },
                "required": ["query"]
            }
        }
    }
]

工具选择的引导

# 好的描述：明确什么场景用，什么场景不用
"description": "获取当前天气信息。当用户询问天气状况、温度、降水概率时使用。不要用于历史天气查询。"

# 坏的描述：模糊不清
"description": "天气工具"  # 模型不知道何时使用

参数验证

from pydantic import BaseModel, Field, validator

class SearchParams(BaseModel):
    query: str = Field(..., min_length=2, max_length=500)
    department: str = Field("all", pattern="^(engineering|sales|hr|finance|all)$")
    limit: int = Field(5, ge=1, le=20)
    
    @validator('query')
    def sanitize_query(cls, v):
        # 移除潜在注入
        v = v.replace('\n', ' ').replace('\r', ' ')
        return v.strip()

def validate_tool_args(func_name, args):
    """验证工具参数"""
    schema_map = {
        "search_knowledge_base": SearchParams,
    }
    
    if func_name in schema_map:
        return schema_map[func_name](**args)
    return args

执行与错误处理

class ToolExecutor:
    def __init__(self, tools_dict, timeout=30):
        self.tools = tools_dict
        self.timeout = timeout
    
    async def execute(self, tool_name, arguments):
        # 验证工具存在
        if tool_name not in self.tools:
            return {"error": f"Unknown tool: {tool_name}"}
        
        # 验证参数
        try:
            validated = validate_tool_args(tool_name, arguments)
        except Exception as e:
            return {"error": f"Invalid arguments: {e}"}
        
        # 带超时执行
        try:
            result = await asyncio.wait_for(
                self.tools[tool_name](**validated.dict()),
                timeout=self.timeout
            )
            return {"result": result}
        except asyncio.TimeoutError:
            return {"error": f"Tool execution timed out after {self.timeout}s"}
        except Exception as e:
            return {"error": f"Tool execution failed: {str(e)}"}

async def function_calling_loop(llm, messages, tools, executor, max_rounds=5):
    """Function Calling循环"""
    for round_idx in range(max_rounds):
        response = await llm.achat_completion(
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        msg = response.choices[0].message
        messages.append(msg)
        
        # 如果模型没有调用工具，返回最终回答
        if not msg.tool_calls:
            return msg.content
        
        # 执行所有工具调用
        for tool_call in msg.tool_calls:
            result = await executor.execute(
                tool_call.function.name,
                json.loads(tool_call.function.arguments)
            )
            
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })
    
    return "达到最大工具调用轮次限制。"

并行工具调用

async def parallel_tool_calls(llm, messages, tools, executor):
    """支持并行工具调用"""
    response = await llm.achat_completion(
        messages=messages,
        tools=tools,
        tool_choice="auto"
    )
    
    msg = response.choices[0].message
    messages.append(msg)
    
    if msg.tool_calls:
        # 并行执行所有工具调用
        tasks = []
        for tool_call in msg.tool_calls:
            task = executor.execute(
                tool_call.function.name,
                json.loads(tool_call.function.arguments)
            )
            tasks.append((tool_call.id, task))
        
        results = await asyncio.gather(*[t for _, t in tasks])
        
        for (tool_call_id, _), result in zip(tasks, results):
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call_id,
                "content": json.dumps(result, ensure_ascii=False)
            })
    
    return messages

工具调用日志

import structlog

logger = structlog.get_logger()

class LoggingToolExecutor:
    def __init__(self, executor):
        self.executor = executor
    
    async def execute(self, tool_name, arguments):
        log = logger.bind(tool=tool_name)
        log.info("tool_call_start", args=arguments)
        
        start = time.time()
        result = await self.executor.execute(tool_name, arguments)
        duration = time.time() - start
        
        if "error" in result:
            log.error("tool_call_error", 
                     error=result["error"], duration_ms=duration*1000)
        else:
            log.info("tool_call_success", duration_ms=duration*1000)
        
        return result

结语

Function Calling的可靠性来自清晰的工具描述、严格的参数验证、完善的错误处理和详尽的日志记录。将这些实践标准化，可以显著提升LLM Agent在生产环境中的稳定性。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。