大模型应用概述¶

"""Token 计数示例"""
import tiktoken

# GPT-4o 使用 o200k_base 编码
enc = tiktoken.encoding_for_model("gpt-4o")

texts = [
    "Hello, world!",
    "大语言模型应用开发",
    "The quick brown fox jumps over the lazy dog.",
    "def fibonacci(n): return n if n <= 1 else fibonacci(n-1) + fibonacci(n-2)",
]

for text in texts:
    tokens = enc.encode(text)
    print(f"文本: {text}")
    print(f"  Token数: {len(tokens)}")
    print(f"  Token列表: {[enc.decode([t]) for t in tokens[:10]]}")
    print()

# 成本估算函数
def estimate_cost(
    input_tokens: int,
    output_tokens: int,
    model: str = "gpt-4o-mini"
) -> dict:
    """估算API调用成本"""
    pricing = {
        "gpt-4o":      {"input": 2.50,  "output": 10.00},
        "gpt-4o-mini": {"input": 0.15,  "output": 0.60},
        "o3":          {"input": 10.00, "output": 40.00},
        "claude-sonnet-4": {"input": 3.00, "output": 15.00},
        "deepseek-v3": {"input": 0.27,  "output": 1.10},
    }

    if model not in pricing:
        return {"error": f"未知模型: {model}"}

    p = pricing[model]
    input_cost = input_tokens / 1_000_000 * p["input"]
    output_cost = output_tokens / 1_000_000 * p["output"]

    return {
        "model": model,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "input_cost": f"${input_cost:.6f}",
        "output_cost": f"${output_cost:.6f}",
        "total_cost": f"${input_cost + output_cost:.6f}",
    }

# 示例：一次典型对话(500输入 + 1000输出)
print(estimate_cost(500, 1000, "gpt-4o-mini"))
# {'model': 'gpt-4o-mini', 'input_tokens': 500, 'output_tokens': 1000,
#  'input_cost': '$0.000075', 'output_cost': '$0.000600', 'total_cost': '$0.000675'}

print(estimate_cost(500, 1000, "gpt-4o"))
# {'model': 'gpt-4o', 'total_cost': '$0.011250'}  —— 贵了约16倍

"""不同 temperature 的效果对比"""
from openai import OpenAI
import os

client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

prompt = "用一句话描述人工智能的未来"

for temp in [0.0, 0.5, 1.0, 1.5]:
    responses = []
    for _ in range(3):  # 每个temperature生成3次
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=temp,
            max_tokens=100,
        )
        responses.append(resp.choices[0].message.content)

    print(f"\n--- temperature={temp} ---")
    for i, r in enumerate(responses, 1):  # enumerate同时获取索引和元素
        print(f"  [{i}] {r}")
    # temperature=0.0 时三次结果几乎相同
    # temperature=1.5 时三次结果差异很大

messages = [
    # system: 设定AI的行为和角色(只需一条，放在最前)
    {
        "role": "system",
        "content": "你是一位资深Python工程师，回答简洁、准确，必要时给出代码示例。"
    },
    # user: 用户输入
    {
        "role": "user",
        "content": "如何实现一个线程安全的单例模式？"
    },
    # assistant: AI的历史回复(用于多轮对话)
    {
        "role": "assistant",
        "content": "可以使用 `threading.Lock` 配合双重检查锁定..."
    },
    # user: 用户的后续问题
    {
        "role": "user",
        "content": "能给一个完整的代码示例吗？"
    },
]

"""安全加载 API Key 的标准做法"""
import os
from dotenv import load_dotenv

load_dotenv()  # 自动读取 .env 文件

# 封装为统一的配置获取函数
def get_api_key(provider: str) -> str:
    """获取指定提供商的 API Key，缺失时给出清晰错误"""
    key_map = {
        "openai": "OPENAI_API_KEY",
        "anthropic": "ANTHROPIC_API_KEY",
        "deepseek": "DEEPSEEK_API_KEY",
    }
    env_var = key_map.get(provider.lower())
    if not env_var:
        raise ValueError(f"不支持的提供商: {provider}")

    key = os.environ.get(env_var)
    if not key:
        raise EnvironmentError(
            f"缺少环境变量 {env_var}，请在 .env 文件中配置"
        )
    return key

"""统一的多提供商LLM客户端"""
from openai import OpenAI
from anthropic import Anthropic
import os
from dotenv import load_dotenv
from dataclasses import dataclass

load_dotenv()

@dataclass  # @dataclass自动生成__init__等方法
class LLMResponse:
    """统一的响应格式"""
    content: str
    model: str
    input_tokens: int
    output_tokens: int
    total_tokens: int

    @property  # @property将方法变为属性访问
    def cost_estimate(self) -> float:
        """粗略成本估算(美元)"""
        pricing = {
            "gpt-4o":          (2.50, 10.00),
            "gpt-4o-mini":     (0.15, 0.60),
            "claude-sonnet-4-20250514": (3.00, 15.00),
            "deepseek-chat":   (0.27, 1.10),
        }
        if self.model in pricing:
            ip, op = pricing[self.model]  # 元组解包：pricing的值是(输入价, 输出价)元组，一行赋值给ip和op两个变量
            return (self.input_tokens * ip + self.output_tokens * op) / 1_000_000
        return 0.0

class UnifiedLLM:
    """支持 OpenAI / Anthropic / DeepSeek 的统一客户端"""

    def __init__(self):
        # OpenAI
        if os.environ.get("OPENAI_API_KEY"):
            self.openai = OpenAI()
        else:
            self.openai = None

        # Anthropic
        if os.environ.get("ANTHROPIC_API_KEY"):
            self.anthropic = Anthropic()
        else:
            self.anthropic = None

        # DeepSeek (兼容OpenAI接口)
        if os.environ.get("DEEPSEEK_API_KEY"):
            self.deepseek = OpenAI(
                api_key=os.environ["DEEPSEEK_API_KEY"],
                base_url="https://api.deepseek.com"
            )
        else:
            self.deepseek = None

    def chat(
        self,
        message: str,
        model: str = "gpt-4o-mini",
        system: str = "You are a helpful assistant.",
        temperature: float = 0.7,
        max_tokens: int = 1024,
    ) -> LLMResponse:
        """统一的对话接口"""

        # Anthropic Claude
        if model.startswith("claude"):
            if not self.anthropic:
                raise RuntimeError("未配置 ANTHROPIC_API_KEY")
            resp = self.anthropic.messages.create(
                model=model,
                max_tokens=max_tokens,
                system=system,
                messages=[{"role": "user", "content": message}],
                temperature=temperature,
            )
            return LLMResponse(
                content=resp.content[0].text,
                model=model,
                input_tokens=resp.usage.input_tokens,
                output_tokens=resp.usage.output_tokens,
                total_tokens=resp.usage.input_tokens + resp.usage.output_tokens,
            )

        # DeepSeek
        elif model.startswith("deepseek"):
            if not self.deepseek:
                raise RuntimeError("未配置 DEEPSEEK_API_KEY")
            client = self.deepseek

        # OpenAI (默认)
        else:
            if not self.openai:
                raise RuntimeError("未配置 OPENAI_API_KEY")
            client = self.openai

        # OpenAI 兼容接口（OpenAI / DeepSeek）
        resp = client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": system},
                {"role": "user", "content": message},
            ],
            temperature=temperature,
            max_tokens=max_tokens,
        )
        usage = resp.usage
        return LLMResponse(
            content=resp.choices[0].message.content,
            model=model,
            input_tokens=usage.prompt_tokens,
            output_tokens=usage.completion_tokens,
            total_tokens=usage.total_tokens,
        )

# ---------- 使用示例 ----------
llm = UnifiedLLM()

# OpenAI
r1 = llm.chat("什么是Transformer?", model="gpt-4o-mini")
print(f"[GPT-4o-mini] {r1.content[:100]}...")
print(f"  Tokens: {r1.total_tokens}, 成本: ${r1.cost_estimate:.6f}")

# DeepSeek (如果配置了key)
# r2 = llm.chat("什么是Transformer?", model="deepseek-chat")

"""使用 Ollama 本地模型 —— 兼容 OpenAI 接口"""
from openai import OpenAI

# Ollama 默认在 localhost:11434 提供 OpenAI 兼容接口
local_client = OpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama",  # Ollama 不需要真实key，但字段必须提供
)

# 与云端API完全相同的调用方式
response = local_client.chat.completions.create(
    model="qwen3:8b",
    messages=[
        {"role": "system", "content": "你是一位AI助手，用中文回答。"},
        {"role": "user", "content": "什么是注意力机制？请简要说明。"}
    ],
    temperature=0.7,
)

print(response.choices[0].message.content)
# 优势：零成本、数据不出本地、无网络延迟
# 劣势：需要GPU(8B模型约需6GB显存)、质量低于顶级闭源模型

"""生产级的基础对话函数"""
from openai import OpenAI, APIError, RateLimitError, APIConnectionError
import os
import time
from dotenv import load_dotenv

load_dotenv()
client = OpenAI()

def chat(
    message: str,
    system: str = "You are a helpful assistant.",
    model: str = "gpt-4o-mini",
    temperature: float = 0.7,
    max_retries: int = 3,
) -> str:
    """
    带重试和错误处理的对话函数

    Args:
        message: 用户消息
        system: 系统提示词
        model: 模型名称
        temperature: 生成温度
        max_retries: 最大重试次数
    Returns:
        模型回复文本
    """
    for attempt in range(max_retries):
        try:  # try/except捕获异常，防止程序崩溃
            response = client.chat.completions.create(
                model=model,
                messages=[
                    {"role": "system", "content": system},
                    {"role": "user", "content": message},
                ],
                temperature=temperature,
            )
            return response.choices[0].message.content

        except RateLimitError:
            wait = 2 ** attempt  # 指数退避: 1s, 2s, 4s
            print(f"[Rate limit] 等待 {wait}s 后重试...")
            time.sleep(wait)

        except APIConnectionError:
            print(f"[连接错误] 第 {attempt+1}/{max_retries} 次重试...")
            time.sleep(1)

        except APIError as e:
            print(f"[API错误] {e.status_code}: {e.message}")
            raise  # 非临时性错误，直接抛出

    raise RuntimeError(f"API调用失败，已重试 {max_retries} 次")

# 使用
print(chat("用三句话解释什么是大语言模型"))

"""流式输出示例 —— 实现打字机效果"""
from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()
client = OpenAI()

def stream_chat(
    message: str,
    system: str = "You are a helpful assistant.",
    model: str = "gpt-4o-mini",
) -> str:
    """流式对话，实时打印每个token"""
    stream = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": message},
        ],
        stream=True,  # 开启流式输出
    )

    full_response = []
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            print(delta.content, end="", flush=True)
            full_response.append(delta.content)

    print()  # 换行
    return "".join(full_response)

# 效果：文字逐字出现，而非等待数秒后一次性显示
response = stream_chat("写一首关于编程的五言绝句")

"""多轮对话管理器 —— 支持历史管理和token限制"""
from openai import OpenAI
import tiktoken
import os
from dotenv import load_dotenv

load_dotenv()

class ChatSession:
    """多轮对话会话管理"""

    def __init__(
        self,
        model: str = "gpt-4o-mini",
        system: str = "You are a helpful assistant.",
        max_history_tokens: int = 4000,
    ):
        self.client = OpenAI()
        self.model = model
        self.system = system
        self.max_history_tokens = max_history_tokens
        self.history: list[dict] = []
        self.enc = tiktoken.encoding_for_model(model)

    def _count_tokens(self, messages: list[dict]) -> int:
        """估算消息列表的token数"""
        total = 0
        for msg in messages:
            total += len(self.enc.encode(msg["content"])) + 4  # 角色标记开销
        return total

    def _trim_history(self):
        """当历史超出token限制时，移除最早的对话轮次"""
        while (
            len(self.history) > 2  # 至少保留最近一轮
            and self._count_tokens(self.history) > self.max_history_tokens
        ):
            # 移除最早的一轮(user + assistant)
            self.history.pop(0)
            if self.history and self.history[0]["role"] == "assistant":
                self.history.pop(0)

    def send(self, message: str, temperature: float = 0.7) -> str:
        """发送消息并获取回复"""
        self.history.append({"role": "user", "content": message})
        self._trim_history()

        messages = [
            {"role": "system", "content": self.system},
            *self.history,
        ]

        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
        )

        reply = response.choices[0].message.content
        self.history.append({"role": "assistant", "content": reply})

        return reply

    def reset(self):
        """清空对话历史"""
        self.history.clear()

    @property
    def total_tokens_used(self) -> int:
        """当前历史的总token数"""
        return self._count_tokens(self.history)

# ---------- 使用示例 ----------
session = ChatSession(
    model="gpt-4o-mini",
    system="你是一位Python编程导师，用简洁易懂的方式教学。",
    max_history_tokens=4000,
)

# 多轮对话
print("Q1:", r1 := session.send("什么是装饰器？"))
print(f"[历史token: {session.total_tokens_used}]\n")

print("Q2:", r2 := session.send("能给个实际的例子吗？"))
print(f"[历史token: {session.total_tokens_used}]\n")

print("Q3:", r3 := session.send("如何用装饰器实现缓存？"))
print(f"[历史token: {session.total_tokens_used}]\n")

# 模型能理解上下文：第3个问题不需要重新解释"装饰器"是什么

"""LLM驱动的智能文本摘要器"""
from openai import OpenAI
import tiktoken
import os
from dotenv import load_dotenv

load_dotenv()
client = OpenAI()

def summarize(
    text: str,
    style: str = "brief",
    model: str = "gpt-4o-mini",
    language: str = "中文",
) -> str:
    """
    智能文本摘要

    Args:
        text: 要摘要的文本
        style: 摘要风格 - "brief"(简要), "detailed"(详细), "bullets"(要点)
        model: 使用的模型
        language: 输出语言
    """
    style_prompts = {
        "brief": f"用{language}写一段简要摘要(50-100字)",
        "detailed": f"用{language}写一段详细摘要(200-300字)，保留关键细节",
        "bullets": f"用{language}列出5-7个关键要点，每个要点一句话",
    }

    # Token 检查
    enc = tiktoken.encoding_for_model(model)
    input_tokens = len(enc.encode(text))

    if input_tokens > 100000:
        return "错误：文本过长，请分段处理"

    prompt = f"""{style_prompts.get(style, style_prompts['brief'])}

原文：
{text}"""

    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "你是一个专业的文本摘要助手。"},
            {"role": "user", "content": prompt},
        ],
        temperature=0.3,  # 摘要任务用低temperature保证准确性
    )

    return response.choices[0].message.content

# 测试
sample_text = """
大语言模型(Large Language Model, LLM)是一种基于深度学习的自然语言处理模型，
通常具有数十亿甚至数千亿个参数。这些模型通过在海量文本数据上进行预训练,
学习了语言的统计规律和语义知识。LLM的核心架构通常基于Transformer,
利用自注意力机制来捕捉文本中的长距离依赖关系。
近年来,GPT系列、Claude系列和开源的LLaMA、Qwen系列等模型不断刷新各项基准测试的记录,
展示了在文本生成、推理、编程、翻译等任务上的卓越能力。
LLM的应用已经从学术研究扩展到了商业领域,包括智能客服、内容创作、
代码辅助、数据分析等场景,正在深刻改变各行各业的工作方式。
"""

print("=== 简要摘要 ===")
print(summarize(sample_text, style="brief"))

print("\n=== 要点摘要 ===")
print(summarize(sample_text, style="bullets"))