[!info] related notes

• 所属 MOC: Tool Calling Engineering MOC
• 相关: Tool Runtime, Tool Registry

Tool Executor

一句话定义

Tool Executor 是真正执行工具逻辑的模块。Tool Runtime 负责校验和控制，Tool Executor 负责”调用函数并返回结果”。

核心原理

Python 实现

class ToolExecutor:
    async def execute(self, tool: Tool, args: dict, context: ExecutionContext) -> any:
        """执行工具并返回结果"""
        # 注入上下文参数（如 user_id）
        enriched_args = self.enrich_args(args, context)

        # 调用 handler
        if asyncio.iscoroutinefunction(tool.handler):
            result = await tool.handler(**enriched_args)
        else:
            result = tool.handler(**enriched_args)

        return result

    def enrich_args(self, args: dict, context: ExecutionContext) -> dict:
        """注入运行时参数"""
        enriched = args.copy()
        # 自动注入 user_id（模型不知道的参数）
        if "user_id" not in enriched:
            enriched["user_id"] = context.user_id
        # 自动注入 session_id
        if "session_id" not in enriched:
            enriched["session_id"] = context.session_id
        return enriched

不同类型的工具执行

# 1. 简单函数
async def search_handler(query: str, **kwargs):
    return await vector_store.search(query)

# 2. API 调用
async def weather_handler(city: str, **kwargs):
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.weather.com/{city}") as resp:
            return await resp.json()

# 3. 数据库查询
async def db_query_handler(sql: str, **kwargs):
    return await db.execute(sql)

# 4. 文件操作
async def read_file_handler(path: str, **kwargs):
    with open(path) as f:
        return f.read()

常见坑

1. 不做参数注入: 模型不知道 user_id，需要运行时注入
2. handler 不统一: 有的是 async，有的是 sync
3. 不做返回值归一化: 有的返回 dict，有的返回 str
4. 不做异常捕获: handler 抛异常导致 Agent 崩溃

参考资料

• Tool Runtime
• Tool Registry