llm_function 装饰器 - SimpleLLMFunc

本文档介绍 SimpleLLMFunc 库中的核心装饰器 llm_function。该装饰器能够将异步 Python 函数的执行委托给大语言模型（LLM），开发者只需要定义函数签名（参数和返回类型）并在文档字符串中描述函数的执行策略，LLM 就会根据描述自动完成函数的实际执行。

快速理解

llm_function 是 SimpleLLMFunc 的核心入口之一。它把一个异步 Python 函数包装成“由 LLM 执行的函数”，你只需要定义函数签名、返回类型和 docstring，框架就会负责构造 prompt、调用模型、解析结果。

智能参数传递

自动把函数参数转换成 LLM 能理解的输入。

类型安全

按返回类型自动解析成 str、dict、Pydantic 模型等。

工具集成

可为函数挂载工具，让模型调用外部能力。

模板扩展

支持自定义 prompt 模板与动态模板参数。

事件流

支持 enable_event=True 观察调用过程和 Token 使用量。

重要说明

llm_function 只能装饰 async def 定义的异步函数，返回的也是可 await 的协程；请在异步上下文中调用，或在脚本入口使用 asyncio.run()。

装饰器用法

基本语法

from SimpleLLMFunc import llm_function

@llm_function(
    llm_interface=llm_interface,
    toolkit=None,
    max_tool_calls=None,
    system_prompt_template=None,
    user_prompt_template=None,
    **llm_kwargs
)
async def your_function(param1: Type1, param2: Type2) -> ReturnType:
    """在这里描述函数的功能和执行策略"""
    pass

函数体不会被执行，DocString 才是 Prompt；建议直接使用 pass。

参数说明

基础参数

llm_interface：必需，LLM 接口实例
toolkit：可选，工具列表，可以是 Tool 对象或被 @tool 装饰的函数
max_tool_calls：可选，最大工具调用次数。默认 None，表示框架不主动限制

Prompt 模板参数

system_prompt_template：自定义系统提示模板
user_prompt_template：自定义用户提示模板

事件流与 LLM 参数

enable_event=False：直接返回解析后的结果
enable_event=True：返回异步生成器，yield ReactOutput
其他 llm_kwargs 会直接传给底层模型接口，例如 temperature、top_p
可传入 retry_times 控制空响应重试次数，默认 2

运行时中断（AbortSignal）

通过 _abort_signal 传入 AbortSignal，可在运行中中断当前回合（停止流式输出并取消正在执行的工具调用）。当 enable_event=True 时可以用 async for 消费输出并随时触发中断；非事件流模式也可传入 _abort_signal 来提前终止 await。

import asyncio
from SimpleLLMFunc.hooks import AbortSignal, ABORT_SIGNAL_PARAM

abort_signal = AbortSignal()

async def run():
    async def abort_later():
        await asyncio.sleep(1.0)
        abort_signal.abort("timeout")

    asyncio.create_task(abort_later())

    # 注意：your_function 需要 enable_event=True 才能用 async for 消费
    async for output in your_function(
        param1,
        **{ABORT_SIGNAL_PARAM: abort_signal},
    ):
        ...

asyncio.run(run())

当 enable_event=True 时，ReactEndEvent.extra 会包含 aborted: true 和可选的 abort_reason。详见中断与取消。

自定义提示模板

系统提示模板占位符

{function_description}：函数文档字符串内容
{parameters_description}：函数参数及其类型描述
{return_type_description}：返回值类型描述

用户提示模板占位符

{parameters}：格式化后的参数名称和值

动态模板参数

SimpleLLMFunc 支持在函数调用时通过 _template_params 参数动态设置 DocString 模板参数。这个功能让同一个函数可以适应不同的使用场景，大大提高了代码的复用性。

使用方法

在 DocString 中使用占位符：

@llm_function(llm_interface=llm)
async def flexible_function(text: str) -> str:
    """作为{role}，请{action}以下文本，输出风格为{style}。"""
    pass

调用时传入模板参数：

import asyncio


async def main():
    # 编辑角色
    result1 = await flexible_function(
        text,
        _template_params={
            'role': '专业编辑',
            'action': '润色',
            'style': '学术'
        }
    )

    # 翻译角色
    result2 = await flexible_function(
        text,
        _template_params={
            'role': '翻译专家',
            'action': '翻译',
            'style': '商务'
        }
    )

    return result1, result2


result1, result2 = asyncio.run(main())

核心特性

动态角色切换

同一个函数可以扮演不同角色。

灵活任务适配

根据调用上下文调整任务类型。

透明处理

_template_params 不会传给模型，只参与模板渲染。

错误兜底

模板参数不完整时，会回退到原始 docstring。

装饰器行为

数据流程

捕获函数调用

装饰器接收实际参数，并解析函数签名。

构造提示词

docstring 作为系统提示核心，参数会被格式化成用户输入；如果提供了模板，会在这一步应用。

调用 LLM 和工具

框架把 prompt 发给模型；如果模型触发工具调用，框架会自动执行工具并继续 ReAct 流程。

解析返回值

把模型输出解析为指定的返回类型，再返回给调用方。

内置功能

类型转换支持

基本类型：str、int、float、bool
容器类型：List、Dict
Pydantic 模型
其他类型默认回退到文本结果

输出格式约束

简单返回类型通常使用纯文本约束
复杂返回类型（Pydantic、List、Dict、Union）会使用更严格的结构化约束

错误处理机制

空响应自动重试
调用异常捕获和日志记录
类型转换失败时提供兜底处理

日志记录

记录参数、提示内容和响应
支持 trace_id 追踪
区分调试、信息、警告和错误级别

常见示例

下面保留原始代码示例，便于直接复制运行。你可以先看标题，再跳到自己最接近的场景。

示例 1: 基本文本处理

import asyncio
from SimpleLLMFunc import llm_function, OpenAICompatible

# 初始化 LLM 接口（推荐方式：从配置文件加载）
models = OpenAICompatible.load_from_json_file("provider.json")
llm = models["openai"]["gpt-3.5-turbo"]

@llm_function(llm_interface=llm)
async def summarize_text(text: str, max_words: int = 100) -> str:
    """根据输入文本生成一个简洁的摘要，摘要不超过指定的词数。"""
    pass

# 使用函数
long_text = "这是一段很长的文本..."


async def main():
    summary = await summarize_text(long_text, max_words=50)
    print(summary)


asyncio.run(main())

示例 2: 结构化数据返回

import asyncio
from typing import Dict, Any, List

@llm_function(llm_interface=llm)
async def analyze_sentiment(text: str) -> Dict[str, Any]:
    """
    分析文本的情感倾向，返回包含以下字段的字典：
    - sentiment: 情感标签（positive/negative/neutral）
    - confidence: 置信度（0-1之间的浮点数）
    - keywords: 关键词列表
    """
    pass

# 使用函数


async def main():
    result = await analyze_sentiment("我今天心情很好，天气也很棒！")
    print(result)


asyncio.run(main())
# 输出: {'sentiment': 'positive', 'confidence': 0.95, 'keywords': ['心情', '好', '天气', '棒']}

示例 3: 使用工具集

import asyncio
from SimpleLLMFunc import tool, llm_function, OpenAICompatible

# 加载模型接口
models = OpenAICompatible.load_from_json_file("provider.json")
llm = models["openai"]["gpt-3.5-turbo"]

@tool
async def search_web(query: str) -> str:
    """在网络上搜索信息"""
    # 实现网络搜索逻辑
    return f"搜索结果: {query}"

@tool
async def calculate(expression: str) -> float:
    """计算数学表达式"""
    return eval(expression)

@llm_function(
    llm_interface=llm,
    toolkit=[search_web, calculate]
)
async def research_and_calculate(topic: str, calculation: str) -> str:
    """
    根据主题搜索相关信息，并进行指定的计算，
    最后整合信息给出综合报告。
    """
    pass

# 使用函数
async def main():
    result = await research_and_calculate(
        topic="Python编程语言",
        calculation="2023 - 1991"
    )
    print(result)


asyncio.run(main())

示例 4: 自定义提示模板

import asyncio

# 自定义系统提示模板
custom_system_template = """
你是一名专业的数据分析师，请根据以下信息执行任务:

参数信息:
{parameters_description}

返回类型: {return_type_description}

任务描述:
{function_description}

请确保分析结果准确、客观，并提供数据支持。
"""

# 自定义用户提示模板
custom_user_template = """
请分析以下数据:
{parameters}

请直接提供分析结果，格式要求为指定的返回类型。
"""

@llm_function(
    llm_interface=llm,
    system_prompt_template=custom_system_template,
    user_prompt_template=custom_user_template,
    temperature=0.7,  # 通过 llm_kwargs 传递模型参数
    top_p=0.9
)
async def analyze_data(data: List[Dict[str, Any]], analysis_type: str) -> Dict[str, Any]:
    """
    对给定的数据集进行指定类型的分析，
    支持的分析类型包括：趋势分析、异常检测、统计摘要等。
    """
    pass

# 使用函数
sample_data = [
    {"date": "2023-01-01", "value": 100},
    {"date": "2023-01-02", "value": 120},
    {"date": "2023-01-03", "value": 95}
]

async def main():
    analysis_result = await analyze_data(sample_data, "趋势分析")
    print(analysis_result)


asyncio.run(main())

示例 5: Pydantic 模型返回

import asyncio
from pydantic import BaseModel
from typing import List

class TaskResult(BaseModel):
    success: bool
    message: str
    tasks: List[str]
    estimated_time: int

@llm_function(llm_interface=llm)
async def create_project_plan(project_description: str, deadline_days: int) -> TaskResult:
    """
    根据项目描述和截止时间，制定详细的项目计划。
    返回包含任务列表、预估时间和执行建议的结构化结果。
    """
    pass

# 使用函数


async def main():
    plan = await create_project_plan(
        project_description="开发一个简单的待办事项应用",
        deadline_days=30
    )

    print(f"计划制定成功: {plan.success}")
    print(f"建议: {plan.message}")
    print(f"任务列表: {plan.tasks}")
    print(f"预估时间: {plan.estimated_time}天")


asyncio.run(main())

示例 6: 动态模板参数

这个示例展示了如何使用动态模板参数让同一个函数适应不同的使用场景：

import asyncio


@llm_function(llm_interface=llm)
async def analyze_code(code: str) -> str:
    """以{style}的方式分析{language}代码，重点关注{focus}。"""
    pass


@llm_function(llm_interface=llm)
async def process_text(text: str) -> str:
    """作为{role}，请{action}以下文本，输出风格为{style}。"""
    pass


# 使用示例
python_code = """
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)
"""


async def main():
    # 不同的分析方式
    performance_analysis = await analyze_code(
        python_code,
        _template_params={
            'style': '详细',
            'language': 'Python',
            'focus': '性能优化'
        }
    )

    code_style_analysis = await analyze_code(
        python_code,
        _template_params={
            'style': '简洁',
            'language': 'Python',
            'focus': '代码规范'
        }
    )

    # 不同的文本处理角色
    sample_text = "人工智能技术正在快速发展，对各行各业产生深远影响。"

    edited_text = await process_text(
        sample_text,
        _template_params={
            'role': '专业编辑',
            'action': '润色',
            'style': '学术'
        }
    )

    translated_text = await process_text(
        sample_text,
        _template_params={
            'role': '翻译专家',
            'action': '翻译成英文',
            'style': '商务'
        }
    )

    print("性能分析结果:", performance_analysis)
    print("代码规范分析:", code_style_analysis)
    print("编辑润色结果:", edited_text)
    print("翻译结果:", translated_text)


asyncio.run(main())

这个示例展示了动态模板参数的强大功能：

一个函数，多种场景：analyze_code 可以用于性能分析、规范检查等不同目的
动态角色切换：process_text 可以扮演编辑、翻译等不同角色
灵活任务适配：根据调用时的参数动态调整任务类型和输出风格

通过这些示例可以看出，llm_function 装饰器提供了一种简洁而强大的方式来利用 LLM 的能力，同时保持了 Python 代码的类型安全性和可读性。

异步使用示例

llm_function 自身即为原生异步实现。以下示例演示如何在不同场景下使用它：

示例 1: 基本异步用法

import asyncio
from SimpleLLMFunc import llm_function, OpenAICompatible

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


@llm_function(llm_interface=llm)
async def summarize_text_async(text: str, max_words: int = 100) -> str:
    """根据输入文本生成一个简洁的摘要，摘要不超过指定的词数。"""
    pass


async def main():
    long_text = "这是一段很长的文本..."
    summary = await summarize_text_async(long_text, max_words=50)
    print(summary)


asyncio.run(main())

示例 2: 并发处理多个请求

import asyncio


@llm_function(llm_interface=llm)
async def translate_text_async(text: str, target_language: str = "English") -> str:
    """将输入文本翻译成指定语言。"""
    pass


@llm_function(llm_interface=llm)
async def analyze_sentiment_async(text: str) -> str:
    """分析文本的情感倾向。"""
    pass


async def process_texts_concurrently():
    texts = [
        "今天天气很好",
        "我感到很沮丧",
        "这个产品质量很棒"
    ]

    translation_tasks = [translate_text_async(text) for text in texts]
    sentiment_tasks = [analyze_sentiment_async(text) for text in texts]

    translations, sentiments = await asyncio.gather(
        asyncio.gather(*translation_tasks),
        asyncio.gather(*sentiment_tasks)
    )

    for i, (text, translation, sentiment) in enumerate(zip(texts, translations, sentiments)):
        print(f"文本 {i + 1}: {text}")
        print(f"翻译: {translation}")
        print(f"情感: {sentiment}")


asyncio.run(process_texts_concurrently())

示例 3: 与其他异步操作配合

import aiohttp
import asyncio


@llm_function(llm_interface=llm)
async def process_content_async(content: str) -> str:
    """处理从网络获取的内容。"""
    pass


async def fetch_and_process_url(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as response:
        content = await response.text()

    processed = await process_content_async(content[:1000])
    return processed


async def process_multiple_urls():
    urls = [
        "https://example1.com",
        "https://example2.com",
        "https://example3.com"
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_and_process_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        for url, result in zip(urls, results):
            if isinstance(result, Exception):
                print(f"处理 {url} 时出错: {result}")
            else:
                print(f"{url}: {result}")


asyncio.run(process_multiple_urls())

这些示例展示了如何使用 llm_function 在异步环境中构建高并发的 LLM 调用逻辑。

事件流使用

llm_function 支持事件流，允许你实时观察函数执行过程中的 LLM 调用、Token 用量等信息。这对于性能监控、成本追踪和调试非常有用。

llm_function 本身不提供流式文本输出。即使启用事件流，ResponseYield 也只会在最终解析完成后返回一次结果。

基本用法

from SimpleLLMFunc import llm_function
from SimpleLLMFunc.hooks import ResponseYield, EventYield
from SimpleLLMFunc.hooks.events import LLMCallEndEvent

@llm_function(
    llm_interface=llm,
    enable_event=True,  # 🔑 启用事件流
)
async def analyze_text(text: str) -> str:
    """分析文本并提供见解"""
    pass

# 处理事件和响应
async for output in analyze_text("Sample text"):
    if isinstance(output, ResponseYield):
        # 获取最终结果（已解析为指定的返回类型）
        print(f"分析结果: {output.response}")
    elif isinstance(output, EventYield):
        event = output.event
        # 处理各种事件
        if isinstance(event, LLMCallEndEvent):
            usage = event.usage
            if usage:
                print(f"Token 用量: {usage.total_tokens}")

Token 用量监控示例

参考完整示例：examples/llm_function_token_usage.py

total_tokens = 0

async for output in summarize_text(text="长文本..."):
    if isinstance(output, EventYield):
        event = output.event
        if isinstance(event, LLMCallEndEvent) and event.usage:
            total_tokens += event.usage.total_tokens
    elif isinstance(output, ResponseYield):
        print(f"摘要: {output.response}")

print(f"总计使用 Token: {total_tokens}")

详细文档：请参考事件流系统文档了解完整的事件类型、使用示例和最佳实践。

最佳实践

1. 错误处理

async def robust_llm_call():
    try:
        result = await your_llm_function("input")
        return result
    except Exception as e:
        print(f"LLM 调用失败: {e}")
        return "默认值"

2. 超时控制

async def llm_call_with_timeout():
    try:
        result = await asyncio.wait_for(
            your_llm_function("input"),
            timeout=30.0  # 30秒超时
        )
        return result
    except asyncio.TimeoutError:
        print("LLM 调用超时")
        return "超时默认值"

通过这些示例可以看出，llm_function 装饰器在异步场景下同样能够提供高性能的 LLM 调用能力，并保持了良好的易用性与功能完整性。

概览

入门

基础设施

开发体验

Agent 主体逻辑

工具与运行时

UI 与交互

集成与示例

​快速理解

智能参数传递

类型安全

工具集成

模板扩展

事件流

​重要说明

​装饰器用法

​基本语法

​参数说明

​运行时中断（AbortSignal）

​自定义提示模板

​动态模板参数

​使用方法

​核心特性

动态角色切换

灵活任务适配

透明处理

错误兜底

​装饰器行为

​数据流程

​内置功能

​常见示例

​示例 1: 基本文本处理

​示例 2: 结构化数据返回

​示例 3: 使用工具集

​示例 4: 自定义提示模板

​示例 5: Pydantic 模型返回

​示例 6: 动态模板参数

​异步使用示例

​示例 1: 基本异步用法

​示例 2: 并发处理多个请求

​示例 3: 与其他异步操作配合

​事件流使用

​基本用法

​Token 用量监控示例

​最佳实践

​1. 错误处理

​2. 超时控制

快速理解

重要说明

装饰器用法

基本语法

参数说明

运行时中断（AbortSignal）

自定义提示模板

动态模板参数

使用方法

核心特性

装饰器行为

数据流程

内置功能

常见示例

示例 1: 基本文本处理

示例 2: 结构化数据返回

示例 3: 使用工具集

示例 4: 自定义提示模板

示例 5: Pydantic 模型返回

示例 6: 动态模板参数

异步使用示例

示例 1: 基本异步用法

示例 2: 并发处理多个请求

示例 3: 与其他异步操作配合

事件流使用

基本用法

Token 用量监控示例

最佳实践

1. 错误处理

2. 超时控制