跳转到主要内容

快速理解

SimpleLLMFunc 的工具系统让 LLM 可以调用外部函数和 API,执行计算、查询、文件操作或多模态处理。框架会把 Python 函数自动转换成模型可理解的工具描述。

类型推断

自动从函数签名中提取参数类型和说明。

Docstring 解析

支持从 docstring 中抽取参数描述。

JSON Schema 生成

自动生成符合 Function Calling 的工具描述。

多返回类型

支持基本类型、结构化数据和多模态返回。

两种创建方式

支持装饰器方式和继承方式。

长文本截断

超长文本结果可自动落盘并截断回传。

支持的数据类型(参数类型)

  • 基本类型:strintfloatbool
  • 容器类型:List[T]Dict[K, V]
  • 多模态类型:TextImgPathImgUrl
  • 多模态列表:List[Text]List[ImgPath]List[ImgUrl]
  • Pydantic 模型
  • 带默认值的可选参数
  • 复杂嵌套结构
LLM 提供给工具的参数本质上仍然是 JSON 支持的数据类型,因此 TupleSet 这类容器通常不适合作为工具参数。

支持的返回类型

工具函数可以返回多种数据格式,框架会自动处理并传递给 LLM。
  • str
  • JSON 可序列化对象:dictlistintfloatboolNone
  • Pydantic 模型
  • ImgUrl
  • ImgPath
  • Tuple[str, ImgUrl]
  • Tuple[str, ImgPath]

返回类型示例

from SimpleLLMFunc import tool
from SimpleLLMFunc.type import ImgUrl, ImgPath
from typing import Dict, List, Tuple, Any

# 1. 基本类型返回
@tool(name="calculate", description="执行数学计算")
async def calculate(expression: str) -> float:
    """返回计算结果(浮点数)"""
    return eval(expression)

@tool(name="get_status", description="获取系统状态")
async def get_status() -> str:
    """返回状态信息(字符串)"""
    return "系统运行正常"

# 2. JSON 对象返回
@tool(name="get_user_info", description="获取用户信息")
async def get_user_info(user_id: int) -> Dict[str, Any]:
    """返回用户信息(字典)"""
    return {
        "id": user_id,
        "name": "张三",
        "age": 25,
        "skills": ["Python", "AI", "数据分析"]
    }

@tool(name="search_results", description="搜索并返回结果列表")
async def search_results(query: str) -> List[Dict[str, str]]:
    """返回搜索结果(字典列表)"""
    return [
        {"title": "结果1", "url": "https://example1.com"},
        {"title": "结果2", "url": "https://example2.com"}
    ]

# 3. 多模态返回 - 单独图片
@tool(name="get_chart", description="生成数据图表")
async def get_chart(data: List[float]) -> ImgPath:
    """返回图表图片(本地文件)"""
    # 在chart path下有一个对应的图片文件
    chart_path = "/path/to/generated/chart.png"
    return ImgPath(chart_path)

@tool(name="fetch_image", description="获取网络图片")
async def fetch_image(image_url: str) -> ImgUrl:
    """返回网络图片URL"""
    return ImgUrl(image_url)

# 4. 多模态返回 - 文本+图片组合
@tool(name="analyze_image", description="分析图片并生成报告")
async def analyze_image(image_path: str) -> Tuple[str, ImgPath]:
    """返回分析报告和标注后的图片"""
    analysis_text = "检测到3个对象:2个人、1辆汽车"
    annotated_image = ImgPath("/path/to/annotated_image.png")
    return (analysis_text, annotated_image)

返回类型处理机制

  1. 基本类型直接序列化回传
  2. ImgUrl 直接作为网络图片使用
  3. ImgPath 会转换为 base64 data URL
  4. 文本 + 图片会被组合成多模态消息
  5. 不支持的返回类型会回退为字符串
当工具返回图片或文本+图片组合时,框架会用 assistant + user 的多模态消息对替代标准 tool 消息,因为 OpenAI tool call 消息本身不能直接携带多模态内容。
  • 本地图片路径必须存在且可读
  • 网络图片 URL 应公开可访问
  • 组合类型必须是 (str, ImgPath)(str, ImgUrl)
  • 避免返回过大的数据结构

长文本截断功能

当工具可能返回非常长的文本结果时,例如代码执行输出或大规模搜索结果,可以启用 too_long_to_file 自动处理。
1

框架估算 token 数量

当工具返回 str 时,框架会估算其 token 大小。
2

超过阈值时自动落盘

超过 20000 tokens 后,完整结果会被写入临时文件,路径格式类似 /tmp/tool_result_{tool_name}_{random}.txt
3

只回传截断结果

返回内容只保留前 20000 tokens,并附带 <system-reminder> 提示,告诉模型完整文件路径。
启用方式:
@tool(name="execute_code", description="执行代码", too_long_to_file=True)
async def execute_code(code: str) -> str:
    """执行 Python 代码并返回输出"""
    # 当输出超过 20000 tokens 时,会自动截断
    result = run_python(code)
    return result
截断后的返回示例:
[前 20000 tokens 的内容...]

<system-reminder\>
tool return was too long. This is truncated result. Full result could be found in /tmp/tool_result_execute_code_a1b2c3d4.txt, you can use file operation toolkits or run code through pyrepl to progressivly reading necessary part of the result file. If you have no way to read file, you should warn the user and let they help you.
</system-reminder\>
@tool 装饰器要求被装饰的函数本身定义为 async def,以便在异步执行链路中无缝协作。
使用建议:
  • 仅在工具可能返回大量文本时启用
  • 启用后,agent 需要具备文件读取能力,例如 FileToolsetPyRepl
  • 对于结构化返回值(dictlist)该功能不会生效

内置工具集

FileToolset(文件工具)

提供安全的文件读取、搜索与编辑能力,并带哈希校验以避免陈旧写入。默认只允许访问工作区内的非隐藏文件。
  • read_file:读取文件内容,可指定起止行并附带行号
  • grep:在工作区内进行正则搜索,必须提供 path_pattern
  • sed:按行范围做正则替换,需先 read_file
  • echo_into:覆盖写入整个文件,需先 read_file
  • 修改前务必先 read_file
  • grep 必须提供 path_pattern
  • 隐藏文件和隐藏目录默认不可访问
使用示例:
from SimpleLLMFunc.builtin import FileToolset
from SimpleLLMFunc import llm_chat

file_tools = FileToolset("/path/to/workspace").toolset

@llm_chat(toolkit=file_tools)
async def agent(message: str, history=None):
    """文件操作助手"""

使用方法

基本语法

方式一:装饰器方式(推荐)

from SimpleLLMFunc import tool

@tool(name="工具名称", description="工具简短描述")
async def your_function(param1: Type1, param2: Type2 = default_value) -> ReturnType:
    """
    详细的函数说明,这部分会被包含在工具描述中
    
    Args:
        param1: 参数1的详细描述
        param2: 参数2的详细描述
        
    Returns:
        返回值描述
    """
    # 函数实现
    pass

方式二:继承方式(兼容旧版本)

from SimpleLLMFunc import Tool

class YourTool(Tool):
    def __init__(self):
        super().__init__(
            name="工具名称",
            description="工具描述"
        )
    
    async def run(self, *args, **kwargs):
        # 工具执行逻辑
        pass

参数说明

  • name:必需,工具名称
  • description:必需,工具简短描述
  • best_practices:可选,工具最佳实践提示
  • prompt_injection_builder:可选,动态注入器
  • too_long_to_file:可选,是否启用长文本截断
  • 建议补全类型标注
  • 建议编写详细 docstring,尤其是 Args
  • 建议显式声明返回类型

工具调用流程

1

注册工具

使用 @tool 或继承 Tool 类创建工具。
2

解析参数和类型

框架自动从函数签名和 docstring 中提取参数信息。
3

生成工具 schema

自动生成符合 OpenAI Function Calling 的工具描述。
4

执行工具

LLM 选择工具后,框架负责参数校验、函数执行和结果回传。

实现方法

核心类结构

Parameter 类

class Parameter:
    """工具参数的包装类"""
    def __init__(self, name, description, type_annotation, required, default=None, example=None):
        self.name = name                    # 参数名
        self.description = description      # 参数描述
        self.type_annotation = type_annotation  # Python 类型标注
        self.required = required            # 是否必需
        self.default = default             # 默认值
        self.example = example             # 示例值

Tool 类

class Tool(ABC):
    """抽象工具基类"""
    def __init__(self, name, description, func=None, too_long_to_file=False):
        self.name = name
        self.description = description
        self.func = func                   # 关联的函数
        self.too_long_to_file = too_long_to_file  # 是否启用长文本截断
        self.parameters = self._extract_parameters()  # 参数列表
    
    def run(self, *args, **kwargs):
        """执行工具"""
        
    def to_openai_tool(self):
        """转换为 OpenAI 工具格式"""
        
    @staticmethod
    def serialize_tools(tools):
        """批量序列化工具"""

关键实现细节

参数提取机制

系统通过以下步骤提取函数参数信息:
  1. 签名分析: 使用 inspect.signature() 获取函数签名
  2. 类型提示: 使用 get_type_hints() 获取类型标注
  3. 文档解析: 使用正则表达式解析 docstring 中的参数描述
  4. 默认值处理: 识别可选参数和默认值

类型转换规则

# 基本类型映射
str  -> {"type": "string"}
int  -> {"type": "integer"}
float -> {"type": "number"}
bool -> {"type": "boolean"}

# 容器类型
List[T] -> {"type": "array", "items": schema_of_T}
Dict[K, V] -> {"type": "object", "additionalProperties": schema_of_V}

# Pydantic 模型
BaseModel -> {"type": "object", "properties": model_json_schema}

文档字符串解析

系统支持标准的 docstring 格式:
def example_function(param1: str, param2: int = 10):
    """
    函数的主要描述
    
    Args:
        param1: 第一个参数的描述
        param2: 第二个参数的描述,可选
        
    Returns:
        返回值描述
    """

兼容写法

装饰器方式示例

from SimpleLLMFunc import tool
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field

# 示例1:基本数据类型
@tool(name="calculate", description="执行数学计算")
async def calculate(expression: str) -> float:
    """
    计算数学表达式的值
    
    Args:
        expression: 要计算的数学表达式,如 "2 + 3 * 4"
        
    Returns:
        计算结果
    """
    return eval(expression)

# 示例2:带可选参数
@tool(name="search_web", description="搜索网络信息")
async def search_web(query: str, max_results: int = 10, language: str = "zh") -> List[Dict[str, str]]:
    """
    在网络上搜索信息
    
    Args:
        query: 搜索关键词
        max_results: 最大返回结果数量,默认10条
        language: 搜索语言,默认中文
        
    Returns:
        搜索结果列表,每个结果包含标题和链接
    """
    # 模拟搜索实现
    return [
        {"title": f"搜索结果 {i}", "url": f"http://example.com/{i}"} 
        for i in range(max_results)
    ]

# 示例3:使用 Pydantic 模型
class Location(BaseModel):
    latitude: float = Field(..., description="纬度")
    longitude: float = Field(..., description="经度")
    name: Optional[str] = Field(None, description="位置名称")

@tool(name="get_weather", description="获取天气信息")
async def get_weather(location: Location, days: int = 1) -> Dict[str, Any]:
    """
    获取指定位置的天气预报
    
    Args:
        location: 位置信息,包含经纬度坐标
        days: 预报天数,1-7天,默认1天
        
    Returns:
        天气预报数据,包含温度、湿度、天气状况等
    """
    return {
        "location": location.name or f"{location.latitude},{location.longitude}",
        "forecast": [
            {
                "day": i + 1,
                "temperature": 25,
                "humidity": 60,
                "condition": "晴朗"
            } for i in range(days)
        ]
    }

# 示例4:复杂数据处理
@tool(name="analyze_data", description="分析数据集")
async def analyze_data(
    data: List[Dict[str, Any]], 
    analysis_type: str,
    include_charts: bool = False
) -> Dict[str, Any]:
    """
    对数据集进行统计分析
    
    Args:
        data: 要分析的数据集,每个元素是一个包含字段的字典
        analysis_type: 分析类型,可选值:summary, trend, correlation
        include_charts: 是否包含图表数据,默认否
        
    Returns:
        分析结果,包含统计信息和可选的图表数据
    """
    result = {
        "type": analysis_type,
        "record_count": len(data),
        "summary": "数据分析完成"
    }
    
    if include_charts:
        result["charts"] = {"type": "bar", "data": "chart_data"}
    
    return result

# 示例5:多模态返回类型
from SimpleLLMFunc.type import ImgUrl, ImgPath
from typing import Tuple

@tool(name="generate_chart", description="生成数据图表")
async def generate_chart(data: List[float], chart_type: str = "bar") -> ImgPath:
    """
    根据数据生成图表并保存为本地文件
    
    Args:
        data: 数据列表
        chart_type: 图表类型,如 bar、line、pie
        
    Returns:
        生成的图表文件路径
    """
    # 模拟图表生成
    import matplotlib.pyplot as plt
    import tempfile
    import os
    
    plt.figure(figsize=(10, 6))
    if chart_type == "bar":
        plt.bar(range(len(data)), data)
    elif chart_type == "line":
        plt.plot(data)
    
    # 保存到临时文件
    temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.png')
    plt.savefig(temp_file.name)
    plt.close()
    
    return ImgPath(temp_file.name)

@tool(name="fetch_web_image", description="获取网络图片")
async def fetch_web_image(image_url: str) -> ImgUrl:
    """
    验证并返回网络图片URL
    
    Args:
        image_url: 图片的网络地址
        
    Returns:
        验证后的图片URL对象
    """
    # 可以添加URL验证逻辑
    return ImgUrl(image_url, detail="high")

@tool(name="analyze_image_with_report", description="分析图片并生成详细报告")
async def analyze_image_with_report(image_path: str) -> Tuple[str, ImgPath]:
    """
    分析图片内容并生成带标注的图片
    
    Args:
        image_path: 要分析的图片路径
        
    Returns:
        分析报告文本和标注后的图片路径
    """
    # 模拟图像分析
    analysis_report = """
    图像分析报告:
    - 检测到 3 个对象
    - 主要颜色:蓝色、绿色
    - 场景类型:户外风景
    - 置信度:95%
    """
    
    # 模拟生成标注图片
    annotated_image_path = "/path/to/annotated_image.png"
    
    return (analysis_report.strip(), ImgPath(annotated_image_path))

@tool(name="create_data_visualization", description="创建在线数据可视化")
async def create_data_visualization(dataset: Dict[str, Any]) -> Tuple[str, ImgUrl]:
    """
    创建数据可视化并上传到云端
    
    Args:
        dataset: 包含数据和配置的字典
        
    Returns:
        可视化说明和在线图片URL
    """
    # 模拟数据可视化处理
    description = f"""
    数据可视化已创建:
    - 数据点数量:{len(dataset.get('data', []))}
    - 图表类型:{dataset.get('chart_type', '未指定')}
    - 创建时间:刚刚
    """
    
    # 模拟上传到云端并获取URL
    visualization_url = ImgUrl("https://example.com/visualizations/chart_12345.png")
    
    return (description.strip(), visualization_url)

继承方式示例

from SimpleLLMFunc import Tool
import requests
from typing import Dict, Any, List

class WebSearchTool(Tool):
    """网络搜索工具的继承实现方式"""
    
    def __init__(self):
        super().__init__(
            name="web_search",
            description="在网络上搜索信息并返回相关结果"
        )
    
    async def run(self, query: str, max_results: int = 5) -> List[Dict[str, str]]:
        """
        执行网络搜索
        
        Args:
            query: 搜索关键词
            max_results: 最大结果数量
            
        Returns:
            搜索结果列表
        """
        # 实际的搜索逻辑
        results = []
        for i in range(max_results):
            results.append({
                "title": f"搜索结果 {i+1}: {query}",
                "url": f"https://example.com/result/{i+1}",
                "snippet": f"关于{query}的相关信息..."
            })
        return results

class APICallTool(Tool):
    """API 调用工具"""
    
    def __init__(self, api_base_url: str):
        super().__init__(
            name="api_call",
            description="调用外部 API 获取数据"
        )
        self.api_base_url = api_base_url
    
    async def run(self, endpoint: str, method: str = "GET", data: Dict[str, Any] = None) -> Dict[str, Any]:
        """
        调用 API 端点
        
        Args:
            endpoint: API 端点路径
            method: HTTP 方法
            data: 请求数据
            
        Returns:
            API 响应数据
        """
        url = f"{self.api_base_url}/{endpoint.lstrip('/')}"
        
        if method.upper() == "GET":
            response = requests.get(url, params=data)
        elif method.upper() == "POST":
            response = requests.post(url, json=data)
        else:
            raise ValueError(f"不支持的 HTTP 方法: {method}")
        
        return response.json()

工具使用示例

import asyncio
from SimpleLLMFunc import llm_function, llm_chat, OpenAICompatible

# 初始化 LLM(从配置文件加载)
models = OpenAICompatible.load_from_json_file("provider.json")
llm = models["openai"]["gpt-3.5-turbo"]

# 在 llm_function 中使用工具
@llm_function(
    llm_interface=llm,
    toolkit=[calculate, search_web, get_weather]
)
async def intelligent_assistant(query: str) -> str:
    """
    智能助手,可以进行计算、搜索和查询天气。
    根据用户查询的内容,选择合适的工具来提供准确的答案。
    """
    pass

# 在 llm_chat 中使用工具
@llm_chat(
    llm_interface=llm,
    toolkit=[calculate, search_web, get_weather, analyze_data]
)
async def chat_with_tools(message: str, history: List[Dict[str, str]] | None = None):
    """
    支持工具调用的聊天助手。
    可以执行计算、搜索网络、查询天气和分析数据。
    """
    pass

# 使用示例
async def main():
    # 使用 llm_function
    result = await intelligent_assistant("帮我计算 25 * 4 + 18 的结果")
    print(f"计算结果: {result}")

    # 使用 llm_chat
    history = []
    async for response, updated_history in chat_with_tools("北京今天天气怎么样?", history):
        if response:
            print(response, end="")
        history = updated_history

if __name__ == "__main__":
    asyncio.run(main())

@llm_function 暴露为工具(Agent as a Tool)

当你希望一个上层 agent 把下层 specialist 当成工具调用时,可以把 @tool 叠加在 @llm_function 外层:
from typing import List, Dict
from SimpleLLMFunc import llm_function, llm_chat, tool


@tool(
    name="requirements_specialist",
    description="Delegate focused requirement analysis",
)
@llm_function(llm_interface=llm)
async def requirements_specialist(task: str) -> str:
    """Analyze the task and return a compact requirement breakdown.

    Args:
        task: The task to analyze.
    """
    pass


@llm_chat(
    llm_interface=llm,
    toolkit=[requirements_specialist],
    stream=True,
)
async def supervisor(
    message: str,
    history: List[Dict[str, str]] | None = None,
):
    """Use requirements_specialist for focused sub-analysis before answering."""
    pass
要点:
  • 推荐顺序写成 @tool 外层、@llm_function 内层。
  • 子 agent 推荐使用 @llm_function(enable_event=False),这样它表现为普通 async tool。
  • @llm_chat 返回的是异步生成器;如果一定要把它当工具复用,通常需要先写一个手动 wrapper。
  • 可运行完整示例见 examples/agent_as_tool_example.py

高级用法

工具序列化和检查

from SimpleLLMFunc import Tool
import json

# 创建工具列表
tools = [calculate, search_web, get_weather]

# 序列化为 OpenAI 格式
openai_tools = Tool.serialize_tools(tools)

# 查看生成的工具描述
for tool_spec in openai_tools:
    print(json.dumps(tool_spec, indent=2, ensure_ascii=False))

动态工具创建

def create_tool_from_config(config: Dict[str, Any]):
    """根据配置动态创建工具"""

    @tool(name=config["name"], description=config["description"])
    async def dynamic_tool(**kwargs):
        # 根据配置执行相应逻辑
        return config["handler"](kwargs)

    return dynamic_tool

# 配置示例
tool_config = {
    "name": "custom_processor",
    "description": "自定义数据处理器",
    "handler": lambda data: {"processed": True, "data": data}
}

custom_tool = create_tool_from_config(tool_config)

工具链组合

@tool(name="multi_step_analysis", description="多步骤数据分析")
async def multi_step_analysis(data: List[Dict[str, Any]], steps: List[str]) -> Dict[str, Any]:
    """
    执行多步骤数据分析流程
    
    Args:
        data: 原始数据
        steps: 分析步骤列表,如 ["clean", "analyze", "visualize"]
        
    Returns:
        分析结果
    """
    results = {"steps_completed": []}
    
    for step in steps:
        if step == "clean":
            # 数据清洗
            results["cleaned_records"] = len(data)
        elif step == "analyze":
            # 数据分析
            results["analysis"] = {"mean": 0, "std": 0}
        elif step == "visualize":
            # 数据可视化
            results["charts"] = ["bar", "line", "pie"]
        
        results["steps_completed"].append(step)
    
    return results

返回类型最佳实践

选择合适的返回类型

  1. 文本输出优先: 如果工具主要产生文本结果,使用 str 或结构化的 Dict
  2. 结构化数据: 复杂数据使用 DictList,便于 LLM 理解和处理
  3. 多模态内容: 需要展示图片时使用 ImgPathImgUrl 或组合类型
  4. 组合输出: 需要同时提供说明和图片时使用 Tuple[str, ImgPath/ImgUrl]

性能优化建议

# ✅ 推荐:结构化返回,便于LLM理解
@tool(name="search_products", description="搜索商品")
async def search_products(query: str) -> Dict[str, Any]:
    return {
        "total": 10,
        "products": [
            {"name": "商品1", "price": 99.9, "in_stock": True},
            {"name": "商品2", "price": 149.9, "in_stock": False}
        ],
        "query_time": "2024-01-01 12:00:00"
    }

# ✅ 推荐:多模态组合返回
@tool(name="generate_report", description="生成分析报告")
async def generate_report(data: List[Dict]) -> Tuple[str, ImgPath]:
    summary = f"分析了 {len(data)} 条记录,发现 3 个关键趋势"
    chart_path = ImgPath("/tmp/analysis_chart.png")
    return (summary, chart_path)

# ❌ 避免:返回过大的数据结构
def bad_example() -> Dict:
    return {
        "huge_data": list(range(10000)),  # 过大的数据
        "binary_content": b"..."  # 二进制数据无法JSON序列化
    }

错误处理模式

@tool(name="safe_division", description="安全除法运算")
async def safe_division(a: float, b: float) -> Dict[str, Any]:
    """
    安全的除法运算,包含错误处理
    
    Returns:
        包含结果或错误信息的字典
    """
    if b == 0:
        return {
            "success": False,
            "error": "除数不能为零",
            "result": None
        }
    
    return {
        "success": True,
        "error": None,
        "result": a / b
    }

@tool(name="robust_image_tool", description="鲁棒的图像处理工具")
async def robust_image_tool(image_path: str) -> Tuple[str, ImgPath]:
    """
    带错误处理的图像工具
    """
    try:
        # 图像处理逻辑
        processed_image = process_image(image_path)
        return ("图像处理成功", ImgPath(processed_image))
    except Exception as e:
        # 返回错误信息和默认图片
        error_msg = f"图像处理失败: {str(e)}"
        default_img = ImgPath("/path/to/error_placeholder.png")
        return (error_msg, default_img)

多模态类型详细说明

from SimpleLLMFunc.type import ImgUrl, ImgPath

# ImgPath 使用示例
img_local = ImgPath(
    path="/path/to/image.jpg",
    detail="high"  # 可选:图片细节级别 low/high
)

# ImgUrl 使用示例  
img_url = ImgUrl(
    url="https://example.com/image.jpg",
    detail="low"  # 网络图片建议使用low以节省token
)

# 组合类型使用
def complex_analysis() -> Tuple[str, ImgPath]:
    analysis = """
    检测结果:
    - 人数:3人
    - 车辆:1辆
    - 置信度:92%
    """
    annotated_img = ImgPath("/tmp/detection_result.jpg")
    return (analysis.strip(), annotated_img)

工具系统提供了强大而灵活的扩展机制,让 LLM 能够调用各种外部功能。通过装饰器方式,开发者可以轻松地将现有函数转换为 LLM 可用的工具,而继承方式则提供了更多的自定义控制。系统自动处理类型转换和参数验证,确保工具调用的安全性和准确性。 关键要点总结
  • 支持多种返回类型:基本类型、JSON对象、多模态内容
  • 多模态支持:单独图片或文本+图片组合
  • 自动类型转换:本地图片转base64,网络图片直接使用URL
  • 错误处理:推荐返回结构化的错误信息
  • 性能考虑:避免返回过大数据,网络图片使用低细节级别