第02课：LLM API调用

🔌 LLM API：Agent的大脑

大语言模型(LLM)是现代AI Agent的"大脑"。掌握API调用是构建Agent的第一步。本课深入探索OpenAI兼容API的调用方法与最佳实践。

📖 主流LLM API对比

🏗️ API调用架构

┌──────────┐     ┌──────────────┐     ┌───────────┐
│  你的代码  │────→│  API Gateway  │────→│   LLM模型  │
│          │←────│  (认证/限流)   │←────│           │
└──────────┘     └──────────────┘     └───────────┘
     │                                       │
     │  请求: messages + params               │
     │  响应: completion + usage              │
     ↓                                       ↓
┌──────────┐                          ┌───────────┐
│  重试/缓存 │                          │  Token计算  │
└──────────┘                          └───────────┘

🔑 API调用核心参数

• model：模型名称，如"gpt-4o"、"gpt-4o-mini"
• messages：消息列表，包含system/user/assistant消息
• temperature：0-2，控制随机性（0=确定性，2=高随机）
• max_tokens：最大生成token数
• top_p：核采样概率阈值
• stop：停止生成的标记
• stream：是否流式返回

💡 Temperature对输出的影响

temperature = 0.0 → 确定性输出，适合代码生成、数据提取
temperature = 0.3 → 低随机性，适合分析、总结
temperature = 0.7 → 平衡模式，适合一般对话
temperature = 1.0 → 高创造性，适合创意写作
temperature = 1.5 → 非常随机，适合头脑风暴

Agent推荐: 0.0-0.3（需要确定性输出）
对话推荐: 0.5-0.8（需要自然感）

💻 代码实现：LLM客户端

# LLM API客户端 - 支持OpenAI兼容接口
import os, time, json
from typing import List, Dict, Optional
from dataclasses import dataclass

@dataclass
class LLMConfig:
    # LLM配置
    api_key: str = os.getenv("OPENAI_API_KEY", "sk-demo")
    base_url: str = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
    model: str = "gpt-4o-mini"
    temperature: float = 0.7
    max_tokens: int = 2048
    timeout: int = 30

class MessageBuilder:
    # 消息构建器
    @staticmethod
    def system(prompt):
        return {"role": "system", "content": prompt}
    @staticmethod
    def user(content):
        return {"role": "user", "content": content}
    @staticmethod
    def assistant(content):
        return {"role": "assistant", "content": content}
    @staticmethod
    def tool_result(tool_call_id, content):
        return {"role": "tool", "tool_call_id": tool_call_id, "content": content}

class MockLLMClient:
    # 模拟LLM客户端（无需真实API Key）
    def __init__(self, config=None):
        self.config = config or LLMConfig()
        self.usage = {"calls": 0, "tokens": 0}
        self.responses = {
            "你好": "你好！我是AI助手，有什么可以帮你的吗？",
            "1+1": "1+1等于2。",
            "写个函数": "好的，这是一个Python示例函数：\ndef hello():\n    print('Hello!')"
        }
    
    def chat(self, messages, **kwargs):
        self.usage["calls"] += 1
        self.usage["tokens"] += len(str(messages)) // 4
        user_msg = ""
        for m in messages:
            if m["role"] == "user":
                user_msg = m["content"]
        for key, resp in self.responses.items():
            if key in user_msg:
                return resp
        return f"收到：{user_msg[:50]}..."
    
    def chat_with_retry(self, messages, max_retries=3, **kwargs):
        # 带指数退避的重试
        for attempt in range(max_retries):
            try:
                return self.chat(messages, **kwargs)
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)
    
    def estimate_cost(self, input_tokens, output_tokens):
        # GPT-4o-mini定价估算
        return input_tokens * 0.15 / 1e6 + output_tokens * 0.60 / 1e6

# 使用示例
client = MockLLMClient()
messages = [MessageBuilder.system("你是友好的AI助手。"), MessageBuilder.user("你好！")]
result = client.chat(messages)
print(f"回复: {result}")

messages.append(MessageBuilder.assistant(result))
messages.append(MessageBuilder.user("1+1等于几？"))
result2 = client.chat(messages)
print(f"回复: {result2}")
print(f"统计: {client.usage}")
print(f"成本: ${client.estimate_cost(500, 200):.6f}")

🏋️ 实战练习

深入理解：LLM API全景

主流LLM API对比

API调用的底层机制

当你调用client.chat.completions.create()时，发生了什么？

客户端 -> API网关 -> 模型推理集群
  | HTTP POST     | 负载均衡    | GPU推理
  | Auth: Bearer  | 选节点      | Token流式返回
  | Content:JSON  |             |
  | <- SSE/JSON - | <- Token流 -|

关键概念：Token(LLM最小文本单位)、Temperature(随机性控制)、Top-p(核采样)、Max tokens(生成长度限制)

进阶实现：统一LLM调用框架

一个健壮的LLM调用框架需要支持：多Provider切换、自动重试、Token统计、错误分类处理。

# UnifiedLLM - 统一LLM调用框架
import time
import json
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum

class Provider(Enum):
    OPENAI = "openai"
    DEEPSEEK = "deepseek"
    ZHIPU = "zhipu"

@dataclass
class TokenUsage:
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0

@dataclass 
class LLMResponse:
    content: str
    model: str
    usage: TokenUsage
    latency_ms: float
    provider: str

class UnifiedLLM:
    # 统一LLM调用框架
    # 
    # 特性：
    # 1. 统一接口切换不同Provider
    # 2. 自动重试（网络错误/限流）
    # 3. Token统计与成本估算
    # 
    PRICING = {
        "gpt-4o": {"input": 0.0025, "output": 0.01},
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
        "deepseek-chat": {"input": 0.00014, "output": 0.00028},
        "glm-4": {"input": 0.001, "output": 0.001},
    }
    
    def __init__(self, provider: Provider = Provider.DEEPSEEK):
        self.provider = provider
        self.call_history: List[Dict] = []
        self.total_usage = TokenUsage()
    
    def _estimate_tokens(self, text: str) -> int:
        cn = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
        en = len(text) - cn
        return int(cn / 1.5 + en / 4) + 1
    
    def chat(self, messages: List[Dict], model: str = None,
             temperature: float = 0.7, max_retries: int = 3) -> LLMResponse:
        # 统一聊天接口 - 自动重试 + Token统计
        config = {
            Provider.OPENAI: {"model": "gpt-4o-mini"},
            Provider.DEEPSEEK: {"model": "deepseek-chat"},
            Provider.ZHIPU: {"model": "glm-4"},
        }
        if model is None:
            model = config[self.provider]["model"]
        
        for attempt in range(max_retries):
            start = time.time()
            try:
                last_msg = messages[-1]["content"]
                usage = TokenUsage(
                    prompt_tokens=self._estimate_tokens(
                        json.dumps(messages, ensure_ascii=False)),
                    completion_tokens=self._estimate_tokens(last_msg) * 2,
                )
                usage.total_tokens = usage.prompt_tokens + usage.completion_tokens
                latency = (time.time() - start) * 1000
                
                response = LLMResponse(
                    content=f"[模拟] 针对'{last_msg[:20]}'的回答，"
                            f"来自{self.provider.value}。",
                    model=model, usage=usage, latency_ms=latency,
                    provider=self.provider.value
                )
                self.total_usage.prompt_tokens += usage.prompt_tokens
                self.total_usage.completion_tokens += usage.completion_tokens
                self.total_usage.total_tokens += usage.total_tokens
                self.call_history.append({
                    "model": model, "tokens": usage.total_tokens,
                    "latency_ms": round(latency, 1), "status": "success"
                })
                return response
            except ConnectionError:
                wait = 2 ** attempt
                print(f"  连接失败(尝试{attempt+1}/{max_retries})，{wait}s后重试...")
                time.sleep(min(wait, 0.5))
        raise RuntimeError(f"API调用失败，已重试{max_retries}次")
    
    def get_stats(self) -> Dict:
        total = len(self.call_history)
        success = sum(1 for h in self.call_history if h["status"] == "success")
        cost = sum(h.get("tokens", 0) / 1000 * 
                   self.PRICING.get(h.get("model", ""), 
                                   {"input": 0, "output": 0}).get("output", 0)
                   for h in self.call_history)
        return {
            "provider": self.provider.value,
            "total_calls": total,
            "success_rate": f"{success/total*100:.1f}%" if total else "N/A",
            "total_tokens": self.total_usage.total_tokens,
            "total_cost_usd": round(cost, 6)
        }

# 测试
llm = UnifiedLLM(Provider.DEEPSEEK)
resp = llm.chat([{"role": "user", "content": "什么是AI Agent？"}])
print(f"响应: {resp.content}")
print(f"Token: {resp.usage.total_tokens}, 耗时: {resp.latency_ms:.1f}ms")

llm2 = UnifiedLLM(Provider.OPENAI)
resp2 = llm2.chat([{"role": "user", "content": "Python是什么？"}])
print(f"响应(OpenAI): {resp2.content}")
print(f"统计: {llm.get_stats()}")

常见问题FAQ

LLM API调用最佳实践

1. 密钥安全 - API Key存环境变量绝不硬编码，使用.env文件
2. 超时设置 - 务必设置timeout推荐30-60秒避免无限等待
3. 重试策略 - 指数退避+最大重试次数，区分可重试错误和不可重试
4. 输入校验 - 发送前检查prompt长度是否在context window内
5. 日志记录 - 记录每次调用的model/tokens/latency/cost
6. 降级方案 - 主Provider不可用时自动切换到备用Provider

设计格言：LLM API调用的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。