进阶教程：在Kotaemon中添加自定义工具调用-平芜编程栈

进阶教程：在Kotaemon中添加自定义工具调用

在构建现代AI系统时，一个核心挑战是让大语言模型（LLM）不再局限于“说”，而是真正能够“做”。我们早已不满足于AI只是回答问题——用户更希望它能查订单、发邮件、调用API、操作数据库。这种从对话智能向行动智能的跃迁，正是当前Agent框架演进的关键方向。

Kotaemon 正是为此而生。作为一个轻量级、可扩展的智能体框架，它通过模块化设计将LLM与外部世界连接起来。其核心机制之一就是工具调用（Tool Calling）——允许开发者将自己的Python函数封装为AI可理解并调度的服务。这不仅提升了系统的实用性，也为企业私有系统的AI集成提供了灵活路径。

工具接口的本质：让AI“听懂”你的函数

要让LLM调用一个函数，首先要让它“理解”这个函数是干什么的、需要什么参数、返回什么结果。这就引出了Kotaemon中最重要的概念：工具描述 schema。

为什么是JSON Schema？

你可能已经注意到，主流平台如OpenAI、Anthropic都采用JSON Schema来描述工具。这不是偶然。Schema本质上是一种结构化元数据，它告诉模型：

函数叫什么名字？
它的功能是什么？（自然语言描述）
接受哪些参数？类型和含义分别是什么？
哪些是必填项？

更重要的是，这些格式已被大量训练数据覆盖，模型对它们有天然的“语感”。例如下面这个天气查询工具的定义：

{ "type": "function", "function": { "name": "get_weather", "description": "获取指定城市的实时天气情况，包括温度和天气状况。", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "城市中文或英文名称，例如'上海'或'Shanghai'" } }, "required": ["location"] } } }

你会发现，description写得越清晰，模型就越不容易误用。比如把“获取天气”写成“拉取气象数据”，虽然技术上没错，但模型可能无法准确关联到用户说的“今天热不热”。

💡 实践建议：不妨站在模型的角度思考——如果你只看这段schema，能不能猜出什么时候该调用它？

如何注册一个真正的可用工具？

光有schema还不够，还得有对应的执行逻辑。以下是一个完整的实现示例：

import requests from typing import Dict, Any def get_weather(location: str) -> Dict[str, Any]: """ 获取指定城市的天气数据 """ api_key = "your_openweather_api_key" # 生产环境应使用配置中心或密钥管理服务 url = f"http://api.openweathermap.org/data/2.5/weather?q={location}&appid={api_key}&units=metric" try: response = requests.get(url, timeout=5) if response.status_code == 200: data = response.json() return { "city": data["name"], "temperature": data["main"]["temp"], "condition": "晴" if "clear" in data["weather"][0]["description"].lower() else "多云/雨" } else: return {"error": f"无法获取 {location} 的天气信息"} except Exception as e: return {"error": str(e)}

注意这里的返回值设计：我们没有直接抛出异常，而是统一包装成包含error字段的对象。这是为了确保即使出错，也能被后续流程安全处理。

接下来，我们需要将函数和schema绑定到Kotaemon的运行时环境中：

from kotaemon.tools import Tool weather_tool = Tool( name="get_weather", description="获取城市天气", func=get_weather, parameters=WEATHER_TOOL_SCHEMA["function"]["parameters"] ) agent.add_tool(weather_tool)

有些开发者会问：“能不能不用手动构造schema？”当然可以！如果框架支持装饰器语法，还能进一步简化：

from kotaemon.decorators import tool @tool( description="获取指定城市的实时天气", parameters={ "location": {"type": "string", "description": "城市名称"} }, required=["location"] ) def get_weather(location: str): # 同上... pass

这种方式更符合Python开发者的直觉，同时也便于做自动化文档生成或测试注入。

模型是如何“决定”调用工具的？

很多人以为工具调用是靠关键词匹配触发的，比如听到“查天气”就去调get_weather。但实际上，现代LLM的能力远不止于此。

结构化输出：模型的新技能

GPT-3.5-turbo及以上版本经过专门训练，能够在特定提示下生成符合预定义结构的JSON输出，而不是自由文本。这就是所谓的structured output generation。

Kotaemon利用这一点，在system prompt中动态注入所有已注册工具的信息。例如：

You are a helpful assistant that can use tools. Available tools: - get_weather(location: str): 获取指定城市的实时天气情况...

当用户输入“北京现在冷吗？”时，模型不会直接回答“挺冷的”，而是判断：“这个问题需要实时数据 → 应该调用工具 → 参数是 location=’北京’”。

于是它输出一段特殊标记包裹的结构化请求：

`` {“name”: “get_weather”, “arguments”: {“location”: “北京”}}

这个过程不是随机的。为了让模型稳定输出这种格式，我们在推理时通常设置： | 参数 | 推荐值 | 说明 | |------|--------|------| | `temperature` | 0.0 ~ 0.3 | 降低随机性，保证一致性 | | `max_tokens` | ≥200 | 预留足够空间用于JSON输出 | | `stop_sequences` | `['<tool_call>']` | 遇到标记即停止，防止截断 | ### 解析与执行：别小看正则表达式 虽然听起来高大上，但最初的工具调用解析其实可以用几行代码完成： ```python import re import json from typing import Optional, Dict, Any def parse_tool_call(content: str) -> Optional[Dict[str, Any]]: pattern = r"<tool_call>(.*?)</tool_call>" match = re.search(pattern, content, re.DOTALL) if not match: return None try: call_data = json.loads(match.group(1)) return { "name": call_data["name"], "arguments": call_data.get("arguments", {}) } except json.JSONDecodeError: print("Invalid JSON in tool_call") return None

这看似简单，但在实际工程中非常有效。当然，生产级系统往往会升级为基于状态机或AST的解析器，以应对嵌套调用、流式输出等复杂场景。

一旦解析成功，Kotaemon就会查找对应函数并执行：

tool_request = parse_tool_call(model_output) if tool_request: result = get_weather(**tool_request["arguments"]) # 将结果回传给模型，用于生成最终回复

此时，整个流程形成了闭环：用户提问 → 模型识别意图 → 输出工具调用 → 执行函数 → 返回结果 → 生成自然语言响应。

更强大的地方在于，这个过程可以多轮进行。比如先查天气，再根据天气推荐穿衣，最后发送提醒邮件——这就是所谓的“工具链（Tool Chain）”。

真实场景落地：客服系统中的订单查询

让我们来看一个典型的企业应用案例：智能客服中的订单状态查询。

想象一位客户问：“我的订单ORD123456到哪儿了？”

如果没有工具调用，AI只能回答：“请登录官网查看。”
而有了工具能力后，它可以主动调用内部系统：

@tool( description="查询订单物流状态", parameters={"order_id": {"type": "string", "description": "订单编号"}}, required=["order_id"] ) def query_order_status(order_id: str): # 调用ERP系统API resp = requests.post("/api/order/status", json={"id": order_id}) if resp.status_code == 200: data = resp.json() return { "status": data["status"], "current_location": data["location"], "estimated_arrival": data["eta"] } else: return {"error": "订单不存在或系统繁忙"}

整个交互流程如下：

用户输入 → NLU识别意图 → 匹配query_order_status工具 ↓ 提取参数 order_id = "ORD123456" ↓ 调用后端服务，返回： { "status": "shipped", "current_location": "上海市分拣中心", "estimated_arrival": "2025-04-08" } ↓ 模型生成自然语言回复： “您的订单已发货，目前位于上海市分拣中心，预计4月8日送达。”

这套架构的优势显而易见：

传统痛点	Kotaemon解决方案
AI只能被动回答	主动调用系统获取真实数据
多个系统分散接入困难	统一抽象为工具接口
用户表达模糊导致错误操作	Schema强约束+参数校验拦截非法输入
故障难以追踪	每次调用都有完整日志，支持重放调试

设计哲学与避坑指南

当你开始编写自己的工具时，以下几个原则值得牢记。

✅ 推荐实践

1. 小颗粒度设计（SRP）

每个工具只做一件事。不要写一个万能函数handle_customer_issue(type, payload)，而是拆分为：

check_order_status
request_refund
create_support_ticket

这样模型更容易精准选择，也便于权限控制和单元测试。

2. 幂等性优先

对于涉及变更的操作（如退款），务必保证重复调用不会造成副作用。例如：

def refund_payment(order_id): if has_already_refunded(order_id): return {"message": "已退款，无需重复操作"} # 执行退款逻辑...

否则模型一旦重试，可能导致资金损失。

3. 错误透明化

永远不要让工具静默失败。返回值中应明确包含error字段，以便模型决定是否重试或提示用户。

4. 权限前置控制

敏感操作（如删除账户）应在工具层集成身份验证，而不是依赖模型“自觉不去调用”。

⚠️ 高危风险警示

❌ 绝对禁止的行为

注册os.system()、subprocess.run()、eval()等任意命令执行函数
暴露数据库原始查询接口（如sql_query("SELECT * FROM users")）
工具返回未脱敏的敏感信息（身份证、手机号、密码哈希）

这些相当于给AI一把万能钥匙，一旦被诱导滥用，后果不堪设想。

🔄 防止无限循环

模型有时会陷入自我调用陷阱。例如反复调用同一个工具却得不到满意结果。解决方法很简单：

# 设置最大调用次数 MAX_TOOL_CALLS = 3 call_count = 0 while call_count < MAX_TOOL_CALLS: output = model.generate(...) tool_call = parse_tool_call(output) if tool_call: result = execute_tool(tool_call) # 将结果加入上下文 conversation.append({"role": "tool", "content": result}) call_count += 1 else: break

超过阈值后强制终止，转由人工介入。

🔐 数据隐私保护

即使是合法调用，也要注意返回数据的最小化原则。比如查询用户信息时，自动过滤掉非必要的字段：

def get_user_profile(user_id): full_data = db.query("...") # 只暴露必要字段 return { "name": full_data["name"], "level": full_data["level"], # 不返回 phone, email, id_card 等 }