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.memo_prompt import MemoPrompt
from tfrobot.brain.chain.prompt.tool_prompt import ToolPrompt
from tfrobot.brain.chain.prompt.knowledge_prompt import KnowledgePrompt

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

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

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

llm.after_input_msg_prompt = [
    MemoPrompt(template="请使用 Markdown 格式回复。"),        # 输出格式
]

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

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": "json_schema",
        "json_schema": {...},
        "examples": [  # Few-shot 示例
            {"name": "张三", "age": 25},
            {"name": "李四", "age": 30},
        ]
    }
)

Token 优化

上下文压缩策略

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

from tfrobot.schema.exceptions import ContextTooLargeError

try:
    result = llm.complete(
        current_input=user_input,
        conversation=long_conversation
    )
except ContextTooLargeError:
    # 手动压缩
    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
    )

摘要模式

# 历史消息自动使用摘要模式
conversation = [...]  # 长对话历史

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

控制知识库大小

# 截断过长的知识库
max_knowledge_length = 5000  # tokens
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 优化

# 工具支持 Map-Reduce 模式
from tfrobot.schema.drive.tool.base import MessageCategory

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

    Returns:
        Map 模式：返回中间结果
        Reduce 模式：返回最终汇总
    """
    # Map 模式
    yield from process_items(items)

    # Reduce 模式
    return summarize_results()

错误处理

统一错误处理

from tfrobot.schema.exceptions import ContextTooLargeError, TFLLMError

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 TFLLMError as e:
    # 其他 LLM 错误
    print(f"LLM 错误: {e}")
    raise
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

缓存

from functools import lru_cache

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

流式输出

llm = GPT(name="gpt-4o", stream=True)

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

成本控制

设置价格

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

统计成本

result = llm.complete(current_input=user_input)

# 查看成本
print(f"输入 tokens: {result.usage.prompt_tokens}")
print(f"输出 tokens: {result.usage.completion_tokens}")
print(f"总 tokens: {result.usage.total_tokens}")

# 计算成本
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}")

使用更便宜的模型

# 简单任务使用 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

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

# 需要细节时使用 high detail
ImagePart(
    image_url=ImgUrl(url="path/to/image.jpg"),
    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

result = llm.complete(current_input=user_input)

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

查看工具调用

result = llm.complete(current_input=user_input, tools=[...])

# 查看工具调用
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.telemetry.tracer import tracer

@tracer.start_as_current_span("my_llm_operation")
def my_llm_function():
    result = llm.complete(current_input=user_input)
    return result

安全建议

API 密钥管理

# 推荐：使用环境变量
import os

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

输出过滤

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

相关文档

• BaseLLM 基类详解
• ChatLLM 对话模型
• 提供商指南