LLM 最佳实践与性能优化¶

本文档汇总了使用 TFRobot LLM 模块的最佳实践和性能优化技巧。

模型选择¶

按场景选择¶

场景	推荐模型	原因
通用对话	GPT-4o, Claude 3.5 Sonnet	综合能力最强
复杂推理	Claude 3.5 Sonnet (Extended Thinking)	推理过程可见
长文档分析	Gemini 1.5 Pro, Claude Opus	上下文窗口大
代码生成	GPT-4o, Claude 3.5 Sonnet	代码能力强
中文场景	DeepSeek, GLM	中文理解好
成本敏感	GPT-4o-mini, Claude Haiku	价格低
快速响应	GPT-4o-mini, Claude Haiku	延迟低
本地部署	Ollama (Llama 3)	数据隐私

成本对比¶

模型	输入价格	输出价格	性价比
GPT-4o	¥0.025/1M	¥0.05/1M	⭐⭐⭐⭐
Claude 3.5 Sonnet	¥0.015/1M	¥0.0375/1M	⭐⭐⭐⭐⭐
Gemini 1.5 Pro	¥0.00875/1M	¥0.02625/1M	⭐⭐⭐⭐⭐
DeepSeek-Chat	¥0.001/1M	¥0.002/1M	⭐⭐⭐⭐⭐
GPT-4o-mini	¥0.00075/1M	¥0.0015/1M	⭐⭐⭐⭐⭐

Prompt 工程¶

Prompt 位置策略¶

from tfrobot.brain.chain.llms import Claude
from tfrobot.brain.chain.prompt.knowledge_prompt import KnowledgePrompt
from tfrobot.brain.chain.prompt.memo_prompt import MemoPrompt
from tfrobot.brain.chain.prompt.template.jinja2_template import Jinja2PromptTemplate
from tfrobot.brain.chain.prompt.tool_prompt import ToolPrompt
from tfrobot.schema.types import Locale

llm = Claude(name="claude-3-5-sonnet-20241022")

# 推荐的组织方式
llm.system_msg_prompt = [
    MemoPrompt(
        template=Jinja2PromptTemplate(templates={Locale.DEFAULT: "你是一个专业的 Python 开发助手。"})
    ),  # 1. 系统角色
    ToolPrompt(),  # 2. 工具列表
    KnowledgePrompt(),  # 3. 知识库
]

llm.before_input_msg_prompt = [
    MemoPrompt(
        template=Jinja2PromptTemplate(templates={Locale.DEFAULT: "当前项目是一个 FastAPI Web 框架。"})
    ),  # 任务背景
]

llm.after_input_msg_prompt = [
    MemoPrompt(template=Jinja2PromptTemplate(templates={Locale.DEFAULT: "请使用 Markdown 格式回复。"})),  # 输出格式
]

llm.after_intermediate_msg_prompt = [
    MemoPrompt(
        template=Jinja2PromptTemplate(templates={Locale.DEFAULT: "请根据工具返回结果继续完成任务。"})
    ),  # 工具调用后引导
]

Prompt 模板选择¶

from tfrobot.brain.chain.prompt.template.jinja2_template import Jinja2PromptTemplate
from tfrobot.schema.types import Locale

# 推荐：使用 Jinja2 模板（支持条件、循环）
template = Jinja2PromptTemplate(
    templates={
        Locale.ZH: """
你是一个{{ role }}。

{% if tools %}
可用工具：
{% for tool in tools %}
- {{ tool.name }}: {{ tool.description }}
{% endfor %}
{% endif %}

用户问题：{{ user_input.input }}
"""
    }
)

Few-Shot 学习¶

from tfrobot.brain.chain.llms import GPT

llm = GPT(
    name="gpt-4o",
    response_format={
        "type": "text",
        "examples": '{"name": "张三", "age": 25}\n{"name": "李四", "age": 30}',  # Few-shot 示例
    },
)

Token 优化¶

上下文压缩策略¶

自动压缩：Chain 会自动触发压缩
手动压缩：调用 collapse_context 方法

from tfrobot.brain.chain.llms import Claude
from tfrobot.schema.exceptions import ContextTooLargeError
from tfrobot.schema.message.conversation.message_dto import TextMessage
from tfrobot.schema.users import BaseUser

llm = Claude(name="claude-3-5-sonnet-20241022")
user = BaseUser(name="用户", uid="1")
user_input = TextMessage(content="很长的问题...", creator=user)
long_conversation = [TextMessage(content=f"历史消息 {i}", creator=user) for i in range(1000)]

try:
    result = llm.complete(current_input=user_input, conversation=long_conversation)
except ContextTooLargeError as e:
    # 手动压缩
    compacted_conv, _, _, _, _ = llm.collapse_context(
        current_input=user_input, conversation=long_conversation, to_size=e.target_size
    )
    result = llm.complete(current_input=user_input, conversation=compacted_conv)

摘要模式¶

from tfrobot.brain.chain.llms import Claude
from tfrobot.schema.message.conversation.message_dto import TextMessage
from tfrobot.schema.users import BaseUser

llm = Claude(name="claude-3-5-sonnet-20241022")
user = BaseUser(name="用户", uid="1")

# 历史消息自动使用摘要模式
conversation = [TextMessage(content=f"历史消息 {i}", creator=user) for i in range(100)]
user_input = TextMessage(content="当前问题", creator=user)

result = llm.complete(current_input=user_input, conversation=conversation)  # 内部自动摘要

控制知识库大小¶

# 截断过长的知识库
max_knowledge_length = 5000  # tokens
knowledge = "这是一段很长的知识库内容..." * 1000  # 模拟长文本
if len(knowledge) > max_knowledge_length:
    knowledge = knowledge[:max_knowledge_length]

使用 smaller models¶

# 简单任务使用更小的模型
from tfrobot.brain.chain.llms import GPT

# 复杂任务
complex_llm = GPT(name="gpt-4o")

# 简单任务
simple_llm = GPT(name="gpt-4o-mini")

工具调用优化¶

工具描述优化¶

from tfrobot.drive.tool.tool import tool


@tool
def search_code(query: str, language: str = "python") -> str:
    """
    搜索代码示例

    Args:
        query: 搜索关键词（如 "快速排序"）
        language: 编程语言，支持 python、javascript、java 等

    Returns:
        代码示例和简要说明

    Examples:
        >>> search_code("快速排序", "python")
        "def quicksort(arr): ..."
    """
    # 工具实现
    ...

工具过滤¶

from tfrobot.brain.chain.llms import GPT

llm = GPT(name="gpt-4o", tool_filter="(&(Edit)(!(IDEWrite)))")  # 仅使用 Tag=Edit 且名称不为 IDEWrite 的工具

Map-Reduce 优化¶

from tfrobot.drive.tool.tool import tool


@tool
def batch_process(items: list[str]) -> str:
    """
    批量处理项目

    Args:
        items: 待处理的项目列表

    Returns:
        Map 模式：返回中间结果
        Reduce 模式：返回最终汇总
    """
    # Map 模式
    for item in items:
        yield f"处理结果: {item}"

    # Reduce 模式
    return f"汇总完成，共处理 {len(items)} 个项目"

错误处理¶

统一错误处理¶

from tfrobot.brain.chain.llms import Claude
from tfrobot.schema.exceptions import ContextTooLargeError
from tfrobot.schema.message.conversation.message_dto import TextMessage
from tfrobot.schema.users import BaseUser

llm = Claude(name="claude-3-5-sonnet-20241022")
user = BaseUser(name="用户", uid="1")
user_input = TextMessage(content="很长的问题...", creator=user)
conversation = [TextMessage(content=f"历史消息 {i}", creator=user) for i in range(1000)]

try:
    result = llm.complete(current_input=user_input)
except ContextTooLargeError as e:
    # 上下文超长
    print(f"上下文超长: {e.current_size} > {e.target_size}")
    # 触发压缩
    compacted, _, _, _, _ = llm.collapse_context(
        current_input=user_input, conversation=conversation, to_size=e.target_size
    )
    result = llm.complete(current_input=user_input, conversation=compacted)
except Exception as e:
    # 其他错误
    print(f"错误: {e}")
    raise

重试策略¶

from tenacity import retry, stop_after_attempt, wait_exponential


@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
)
def call_llm_with_retry(llm, user_input):
    return llm.complete(current_input=user_input)

性能优化¶

并发调用¶

import asyncio


async def batch_call_llm(llm, inputs):
    tasks = [llm.async_complete(current_input=inp) for inp in inputs]
    results = await asyncio.gather(*tasks)
    return results

缓存¶

import functools

from tfrobot.schema.message.conversation.message_dto import TextMessage
from tfrobot.schema.users import BaseUser

user = BaseUser(name="用户", uid="1")


@functools.lru_cache(maxsize=128)
def cached_llm_call(llm, query: str):
    return llm.complete(current_input=TextMessage(content=query, creator=user))

流式输出¶

from tfrobot.brain.chain.llms import GPT
from tfrobot.schema.message.conversation.message_dto import TextMessage
from tfrobot.schema.users import BaseUser

llm = GPT(name="gpt-4o", stream=True)
user = BaseUser(name="用户", uid="1")
user_input = TextMessage(content="你好", creator=user)

result = llm.complete(current_input=user_input)
# result.generations[0].text 会逐步生成

成本控制¶

设置价格¶

from tfrobot.brain.chain.llms import GPT

llm = GPT(
    name="gpt-4o",
    input_price=0.005,  # 5美元/百万tokens
    output_price=0.015,  # 15美元/百万tokens
)

统计成本¶

from tfrobot.brain.chain.llms import GPT
from tfrobot.schema.message.conversation.message_dto import TextMessage
from tfrobot.schema.users import BaseUser

llm = GPT(
    name="gpt-4o",
    input_price=0.005,
    output_price=0.015,
)
user = BaseUser(name="用户", uid="1")
user_input = TextMessage(content="你好", creator=user)

result = llm.complete(current_input=user_input)

# 查看成本
print(f"输入 tokens: {result.usage['prompt_tokens']}")
#> 输入 tokens: 10
print(f"输出 tokens: {result.usage['completion_tokens']}")
#> 输出 tokens: 11
print(f"总 tokens: {result.usage['total_tokens']}")
#> 总 tokens: 21

# 计算成本
input_cost = result.usage['prompt_tokens'] * llm.input_price / 1000
output_cost = result.usage['completion_tokens'] * llm.output_price / 1000
total_cost = input_cost + output_cost
print(f"总成本: ¥{total_cost:.4f}")
#> 总成本: ¥0.0002

使用更便宜的模型¶

from tfrobot.brain.chain.llms import GPT
from tfrobot.schema.message.conversation.message_dto import TextMessage
from tfrobot.schema.users import BaseUser

user = BaseUser(name="用户", uid="1")
user_input = TextMessage(content="简单问题", creator=user)


def is_simple_task(msg: TextMessage) -> bool:
    """判断是否为简单任务"""
    return len(msg.content) < 50


# 简单任务使用 mini 模型
if is_simple_task(user_input):
    llm = GPT(name="gpt-4o-mini")
else:
    llm = GPT(name="gpt-4o")

多模态优化¶

图片 Detail 级别¶

from tfrobot.schema.message.msg_part import ImagePart, ImgUrl

# 不需要细节时使用 low detail
ImagePart(image_url=ImgUrl(url="path/to/image.jpg", mime_type="image/jpeg"), detail="low")  # 仅 85 tokens

# 需要细节时使用 high detail
ImagePart(image_url=ImgUrl(url="path/to/image.jpg", mime_type="image/jpeg"), detail="high")  # 根据瓦片数计算

图片预压缩¶

from PIL import Image


def compress_image(image_path: str, max_size: int = 1024):
    """压缩图片以减少 token 消耗"""
    img = Image.open(image_path)

    # 等比缩放
    img.thumbnail((max_size, max_size))

    # 保存
    compressed_path = image_path.replace(".", "_compressed.")
    img.save(compressed_path, quality=85)
    return compressed_path

遥测与调试¶

查看实际 Prompt¶

from tfrobot.brain.chain.llms import GPT
from tfrobot.schema.message.conversation.message_dto import TextMessage
from tfrobot.schema.users import BaseUser

llm = GPT(name="gpt-4o")
user = BaseUser(name="用户", uid="1")
user_input = TextMessage(content="你好", creator=user)

result = llm.complete(current_input=user_input)

# 查看实际发送给 LLM 的 prompt
print(result.meta_info.get("prompt"))

查看工具调用¶

from tfrobot.brain.chain.llms import GPT
from tfrobot.schema.message.conversation.message_dto import TextMessage
from tfrobot.schema.users import BaseUser

llm = GPT(name="gpt-4o")
user = BaseUser(name="用户", uid="1")
user_input = TextMessage(content="你好", creator=user)

result = llm.complete(current_input=user_input)

# 查看工具调用
for gen in result.generations:
    if gen.tool_calls:
        for tc in gen.tool_calls:
            print(f"调用工具: {tc.function.name}")
            print(f"参数: {tc.function.parameters}")

OpenTelemetry 追踪¶

from tfrobot.brain.chain.llms import GPT
from tfrobot.schema.message.conversation.message_dto import TextMessage
from tfrobot.schema.users import BaseUser
from tfrobot.telemetry.tracer import tracer

llm = GPT(name="gpt-4o")
user = BaseUser(name="用户", uid="1")


@tracer.start_as_current_span("my_llm_operation")
def my_llm_function():
    user_input = TextMessage(content="你好", creator=user)
    result = llm.complete(current_input=user_input)
    return result

安全建议¶

API 密钥管理¶

import os

from tfrobot.brain.chain.llms import GPT

# 推荐：使用环境变量
api_key = os.getenv("OPENAI_API_KEY")
llm = GPT(name="gpt-4o", openai_api_key=api_key)

# 不推荐：硬编码密钥
# llm = GPT(name="gpt-4o", openai_api_key="sk-...")

输入验证¶

def validate_input(user_input: str) -> bool:
    """验证用户输入"""
    if len(user_input) > 10000:
        raise ValueError("输入过长")
    # 其他验证逻辑
    return True

输出过滤¶

from tfrobot.schema.message.llms.llm_response import LLMResult


def filter_output(result: LLMResult) -> str:
    """过滤输出内容"""
    text = result.generations[0].text
    # 过滤敏感词
    # ...
    return text

LLM 最佳实践与性能优化¶

模型选择¶

按场景选择¶

成本对比¶

Prompt 工程¶

Prompt 位置策略¶

Prompt 模板选择¶

Few-Shot 学习¶

Token 优化¶

上下文压缩策略¶

摘要模式¶

控制知识库大小¶

使用 smaller models¶

工具调用优化¶

工具描述优化¶

工具过滤¶

Map-Reduce 优化¶

错误处理¶

统一错误处理¶

重试策略¶

性能优化¶

并发调用¶

缓存¶

流式输出¶

成本控制¶

设置价格¶

统计成本¶

使用更便宜的模型¶

多模态优化¶

图片 Detail 级别¶

图片预压缩¶

遥测与调试¶

查看实际 Prompt¶

查看工具调用¶

OpenTelemetry 追踪¶

安全建议¶

API 密钥管理¶

输入验证¶

输出过滤¶

相关文档¶