从零开始实践LLM：ChatGPT应用开发避坑指南

1次阅读

共计 3589 个字符，预计需要花费 9 分钟才能阅读完成。

大语言模型 (LLM, Large Language Model) 正在重塑软件开发范式：它能将自然语言转化为可执行逻辑，大幅降低开发门槛；其通用能力覆盖文本生成、代码补全等场景，显著提升生产效率；同时 API 化服务让集成成本趋近于零。对开发者而言，掌握 LLM 应用开发已成为必备技能。

开发前需根据业务需求选择服务商，以下是三大主流方案的横向对比：

OpenAI API
计费方式：按 token 数量计费（输入 + 输出累计）
响应延迟：200-800ms（gpt-3.5-turbo 模型）
上下文长度：16385 tokens（gpt-4-turbo）
特点：模型版本迭代快，工具链完善
Anthropic Claude
计费方式：按百万 token 阶梯定价
响应延迟：300-1200ms（Claude 2 模型）
上下文长度：200K tokens（行业领先）
特点：长文本处理优势明显
Google Bard
计费方式：免费（当前阶段）
响应延迟：500-1500ms
上下文长度：约 8K tokens
特点：谷歌生态集成便利

建议初创项目从 OpenAI API 起步，其稳定的 SDK 和丰富文档能降低学习曲线。

from typing import Optional, Dict, Any
import openai
from tenacity import retry, stop_after_attempt, wait_exponential

class LLMClient:
    def __init__(self, api_key: str):
        openai.api_key = api_key

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
    async def chat_completion(
        self,
        messages: list[dict[str, str]],
        model: str = "gpt-3.5-turbo",
        temperature: float = 0.7,
        max_tokens: Optional[int] = None,
    ) -> dict[str, Any]:
        try:
            return await openai.ChatCompletion.acreate(
                model=model,
                messages=messages,
                temperature=temperature,
                max_tokens=max_tokens,
                timeout=10  # 秒
            )
        except Exception as e:
            # 结构化异常处理
            if "rate limit" in str(e).lower():
                raise RateLimitError("API 调用频率超限")
            raise LLMError(f"API 调用失败: {str(e)}")

关键设计点：
1. 使用 tenacity 库实现指数退避重试
2. 异步接口提升并发性能
3. 自定义异常类型便于业务处理

def build_qa_prompt(question: str, context: Optional[str] = None) -> list[dict[str, str]]:
    messages = [{"role": "system", "content": "你是一个专业的技术顾问，用简洁准确的语言回答问题"},
        {"role": "user", "content": "请用中文回答以下问题"}
    ]

    if context:
        messages.append({"role": "assistant", "content": context})

    messages.append({"role": "user", "content": question})
    return messages

角色分工原则：
– system：定义 AI 行为范式
– user：实际用户输入
– assistant：多轮对话中的历史响应

async def stream_response(prompt: str):
    stream = await openai.ChatCompletion.acreate(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )

    async for chunk in stream:
        content = chunk["choices"][0].get("delta", {}).get("content")
        if content:
            yield content  # 逐块返回给前端

适用于需要实时显示生成结果的场景（如聊天界面）。

async def batch_completion(prompts: list[str], max_batch_size=5):
    """适合日志分析等非实时场景"""
    semaphore = asyncio.Semaphore(max_batch_size)

    async def process(prompt):
        async with semaphore:
            return await client.chat_completion(build_qa_prompt(prompt)
            )

    return await asyncio.gather(*[process(p) for p in prompts])

from datetime import timedelta
from redis import Redis

class DialogueCache:
    def __init__(self, redis_client: Redis):
        self.client = redis_client

    def get_session(self, session_id: str) -> list[dict]:
        if history := self.client.get(f"llm:{session_id}"):
            return json.loads(history)
        return []

    def update_session(self, session_id: str, messages: list[dict], ttl=timedelta(hours=2)):
        self.client.setex(f"llm:{session_id}",
            int(ttl.total_seconds()),
            json.dumps(messages)
        )

class FallbackLLM:
    """当主服务不可用时启用的本地轻量模型"""
    def __init__(self, local_model_path: str):
        self.model = load_local_model(local_model_path)  # 例如 llama.cpp

    def generate(self, prompt: str) -> str:
        return self.model.generate(prompt, max_length=512)

def sanitize_input(text: str) -> str:
    """移除 PII(个人身份信息)等敏感内容"""
    patterns = [(r"\b\d{4}[-]?\d{4}[-]?\d{4}\b", "[信用卡号]")
    ]
    for pattern, repl in patterns:
        text = re.sub(pattern, repl, text)
    return text

实现令牌桶算法控制请求速率
监控 429 状态码自动触发降速
重要业务设置专属 API-KEY

class CostMonitor:
    def __init__(self):
        self.counter = Counter()

    def track(self, model: str, input_tokens: int, output_tokens: int):
        rate = get_rate_card(model)  # 获取定价表
        cost = (input_tokens * rate["in"] + output_tokens * rate["out"]) / 1000
        self.counter[model] += cost

        if cost > DAILY_BUDGET * 0.8:
            alert_slack(f"预算预警: {model}消耗 {cost:.2f} 美元")