LLM应用开发实战：从原型到生产部署的完整指南

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

"""
模型选择决策矩阵：

┌─────────────┬──────────┬──────────┬──────────┐
│    维度     │  商业API  │  开源模型 │  自部署   │
├─────────────┼──────────┼──────────┼──────────┤
│ 成本        │ 按token   │ 基础设施  │ 基础设施  │
│ 性能        │ 最优      │ 中等      │ 可定制    │
│ 数据隐私    │ 云端      │ 本地      │ 完全控制  │
│ 定制化      │ 有限      │ 微调      │ 全量训练  │
│ 延迟        │ 网络      │ 本地推理  │ 本地推理  │
│ 维护成本    │ 低        │ 中        │ 高        │
│ 适用场景    │ 快速原型  │ 敏感数据  │ 核心业务  │
└─────────────┴──────────┴──────────┴──────────┘
"""

class ModelSelector:
    """模型选择器"""

    def recommend_model(self, requirements: dict):
        """根据需求推荐模型"""

        # 成本敏感型
        if requirements.get("cost_sensitive"):
            return {
                "model": "gpt-4o-mini",
                "reason": "低成本，性价比高",
                "input_price": 0.15,
                "output_price": 0.60
            }

        # 性能优先型
        if requirements.get("performance_first"):
            return {
                "model": "claude-3-5-sonnet",
                "reason": "最强性能，复杂推理",
                "input_price": 3.0,
                "output_price": 15.0
            }

        # 数据隐私要求
        if requirements.get("data_privacy"):
            return {
                "model": "llama-3-70b",
                "reason": "本地部署，数据不出域",
                "deployment": "on-premise"
            }

        # 平衡型
        return {
            "model": "gpt-4o",
            "reason": "性能与成本平衡",
            "input_price": 2.5,
            "output_price": 10.0
        }

    def estimate_cost(self, model: str, tokens_per_day: int):
        """估算成本"""

        pricing = {
            "gpt-4o": {"input": 2.5, "output": 10.0},
            "gpt-4o-mini": {"input": 0.15, "output": 0.60},
            "claude-3-5-sonnet": {"input": 3.0, "output": 15.0},
            "claude-3-haiku": {"input": 0.25, "output": 1.25}
        }

        # 假设输入:输出 = 1:2
        input_tokens = tokens_per_day // 3
        output_tokens = tokens_per_day * 2 // 3

        daily_cost = (
            (input_tokens / 1_000_000) * pricing[model]["input"] +
            (output_tokens / 1_000_000) * pricing[model]["output"]
        )

        return {
            "daily": daily_cost,
            "monthly": daily_cost * 30,
            "yearly": daily_cost * 365
        }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

from anthropic import Anthropic
from openai import OpenAI
import tenacity

class LLMClient:
    """统一的LLM客户端"""

    def __init__(self, provider: str = "anthropic"):
        self.provider = provider

        if provider == "anthropic":
            self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        elif provider == "openai":
            self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        else:
            raise ValueError(f"Unsupported provider: {provider}")

    @tenacity.retry(
        stop=tenacity.stop_after_attempt(3),
        wait=tenacity.wait_exponential(multiplier=1, min=4, max=10)
    )
    def chat(self, messages: list, **kwargs) -> str:
        """聊天接口"""

        if self.provider == "anthropic":
            response = self.client.messages.create(
                model=kwargs.get("model", "claude-3-5-sonnet-20241022"),
                max_tokens=kwargs.get("max_tokens", 4096),
                temperature=kwargs.get("temperature", 0.7),
                messages=messages
            )
            return response.content[0].text

        elif self.provider == "openai":
            response = self.client.chat.completions.create(
                model=kwargs.get("model", "gpt-4o"),
                messages=messages,
                temperature=kwargs.get("temperature", 0.7)
            )
            return response.choices[0].message.content

    def stream_chat(self, messages: list, **kwargs):
        """流式聊天"""

        if self.provider == "anthropic":
            with self.client.messages.stream(
                model=kwargs.get("model", "claude-3-5-sonnet-20241022"),
                max_tokens=kwargs.get("max_tokens", 4096),
                messages=messages
            ) as stream:
                for text in stream.text_stream:
                    yield text

    async def achat(self, messages: list, **kwargs) -> str:
        """异步聊天"""
        # 实现异步版本...
        pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

from pydantic import BaseModel, Field
from typing import List, Optional

class Product(BaseModel):
    """产品信息模型"""
    name: str = Field(description="产品名称")
    price: float = Field(description="产品价格")
    category: str = Field(description="产品类别")
    description: str = Field(description="产品描述")
    tags: List[str] = Field(description="产品标签")

class ExtractionApp:
    """信息提取应用"""

    def __init__(self):
        self.client = LLMClient("anthropic")

    def extract_product(self, text: str) -> Product:
        """从文本中提取产品信息"""

        prompt = f"""
从以下文本中提取产品信息，以JSON格式输出：

{text}

请提取：
- name: 产品名称
- price: 价格（数字）
- category: 类别
- description: 描述
- tags: 标签列表
"""

        response = self.client.chat([
            {"role": "user", "content": prompt}
        ], temperature=0.3)

        # 解析JSON响应
        import json
        data = json.loads(response)
        return Product(**data)

    def batch_extract(self, texts: List[str]) -> List[Product]:
        """批量提取"""

        results = []
        for text in texts:
            try:
                product = self.extract_product(text)
                results.append(product)
            except Exception as e:
                print(f"Error extracting from text: {e}")
                continue

        return results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

class ChainOfThoughtApp:
    """思维链应用"""

    def __init__(self):
        self.client = LLMClient()

    def solve_math_problem(self, problem: str) -> str:
        """解决数学问题"""

        prompt = f"""
解决以下数学问题，请一步步思考：

{problem}

请按以下格式回答：
1. 理解问题
2. 制定计划
3. 分步计算
4. 得出答案
5. 验证结果
"""

        response = self.client.chat([
            {"role": "user", "content": prompt}
        ], temperature=0.5)

        return response

    def complex_reasoning(self, question: str) -> str:
        """复杂推理"""

        prompt = f"""
请仔细思考以下问题，给出详细的分析和结论：

{question}

要求：
1. 识别问题的核心要素
2. 分析各种可能的情况
3. 逐步推理得出结论
4. 检查结论的合理性
"""

        return self.client.chat([
            {"role": "user", "content": prompt}
        ])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

class FewShotApp:
    """少样本学习应用"""

    def __init__(self):
        self.client = LLMClient()

    def sentiment_analysis(self, text: str) -> str:
        """情感分析（少样本）"""

        examples = [
            {
                "text": "这个产品太棒了！非常推荐！",
                "sentiment": "positive"
            },
            {
                "text": "质量一般，价格偏高",
                "sentiment": "neutral"
            },
            {
                "text": "完全浪费钱，强烈不推荐",
                "sentiment": "negative"
            }
        ]

        prompt = "以下是文本的情感分类示例：\n\n"

        for ex in examples:
            prompt += f"文本: {ex['text']}\n"
            prompt += f"情感: {ex['sentiment']}\n\n"

        prompt += f"现在分析以下文本的情感：\n{text}\n"
        prompt += "情感："

        response = self.client.chat([
            {"role": "user", "content": prompt}
        ], temperature=0.3)

        return response.strip()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

class SimpleRAG:
    """简单的RAG系统"""

    def __init__(self):
        self.client = LLMClient()
        self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
        self.index = None
        self.documents = []

    def add_documents(self, docs: List[str]):
        """添加文档"""

        # 生成嵌入
        embeddings = self.embedder.encode(docs)

        # 归一化
        faiss.normalize_L2(embeddings)

        # 创建索引
        if self.index is None:
            dimension = embeddings.shape[1]
            self.index = faiss.IndexFlatIP(dimension)

        # 添加到索引
        self.index.add(embeddings.astype('float32'))

        # 存储文档
        self.documents.extend(docs)

    def retrieve(self, query: str, k: int = 5) -> List[str]:
        """检索相关文档"""

        # 生成查询嵌入
        query_embedding = self.embedder.encode([query])
        faiss.normalize_L2(query_embedding)

        # 搜索
        scores, indices = self.index.search(
            query_embedding.astype('float32'),
            k
        )

        # 返回文档
        results = []
        for score, idx in zip(scores[0], indices[0]):
            results.append({
                "document": self.documents[idx],
                "score": float(score)
            })

        return results

    def ask(self, question: str) -> str:
        """问答"""

        # 检索相关文档
        context_docs = self.retrieve(question, k=3)

        # 构建提示
        context = "\n\n".join([
            doc["document"] for doc in context_docs
        ])

        prompt = f"""
基于以下信息回答问题：

{context}

问题：{question}

如果信息不足以回答，请明确说明。
"""

        response = self.client.chat([
            {"role": "user", "content": prompt}
        ])

        return response

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List
import redis
import json

app = FastAPI(title="LLM API Service")

# 全局配置
llm_client = LLMClient("anthropic")
cache = redis.Redis(host='localhost', port=6379, db=0)

class ChatRequest(BaseModel):
    messages: List[dict]
    temperature: float = 0.7
    max_tokens: int = 4096
    stream: bool = False

class ChatResponse(BaseModel):
    content: str
    model: str
    tokens_used: int

@app.post("/api/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """聊天接口"""

    # 生成缓存key
    cache_key = f"chat:{hash(str(request.messages))}"

    # 尝试从缓存获取
    cached = cache.get(cache_key)
    if cached:
        return ChatResponse(**json.loads(cached))

    try:
        # 调用LLM
        response = llm_client.chat(
            request.messages,
            temperature=request.temperature,
            max_tokens=request.max_tokens
        )

        result = ChatResponse(
            content=response,
            model="claude-3-5-sonnet",
            tokens_used=0  # 需要从响应中获取
        )

        # 缓存结果（1小时）
        cache.setex(
            cache_key,
            3600,
            json.dumps(result.dict())
        )

        return result

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/api/chat/stream")
async def chat_stream(request: ChatRequest):
    """流式聊天"""

    async def generate():
        try:
            for chunk in llm_client.stream_chat(
                request.messages,
                temperature=request.temperature,
                max_tokens=request.max_tokens
            ):
                yield f"data: {json.dumps({'content': chunk})}\n\n"

        except Exception as e:
            yield f"data: {json.dumps({'error': str(e)})}\n\n"
            yield "data: [DONE]\n\n"

    from fastapi.responses import StreamingResponse
    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

# 启动命令
# uvicorn api:app --host 0.0.0.0 --port 8000 --workers 4

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

class PerformanceOptimizer:
    """性能优化器"""

    def implement_caching(self):
        """实现多层缓存"""

        cache_strategy = {
            "l1_memory": {
                "ttl": 300,  # 5分钟
                "max_size": 1000
            },
            "l2_redis": {
                "ttl": 3600,  # 1小时
                "max_size": 10000
            },
            "l3_database": {
                "ttl": 86400  # 1天
            }
        }

        return cache_strategy

    def batch_processing(self, prompts: List[str]) -> List[str]:
        """批量处理"""

        # 使用多线程并发请求
        from concurrent.futures import ThreadPoolExecutor

        with ThreadPoolExecutor(max_workers=5) as executor:
            results = list(executor.map(
                self.client.chat,
                [{"role": "user", "content": p} for p in prompts]
            ))

        return results

    def token_optimization(self, prompt: str) -> str:
        """Token优化"""

        # 移除冗余信息
        optimized = (
            prompt
            .replace("请", "")
            .replace("您", "")
            .replace("好的，", "")
        )

        return optimized.strip()