Chaos and Order

💡 왼쪽 원문을 읽으면서 오른쪽에 따라 써보세요. Tab 키로 힌트를 받을 수 있습니다.

1. LLM 应用开发概述

1.1 什么是 LLM 应用？

LLM 应用是以大语言模型作为核心组件，用于处理自然语言、生成内容、对信息进行推理或执行动作的任何软件系统。与所有行为都被显式编程的传统软件不同，LLM 应用把相当一部分逻辑委托给了预训练模型。

常见的 LLM 应用类别：

类别	示例	核心挑战
聊天机器人与助手	客户支持、个人助理	上下文管理、语气一致性
文档问答	合同审查、内部检索	检索准确性、幻觉
代码生成	自动补全、PR 审查、编写测试	正确性、安全性
内容生成	营销文案、摘要	质量控制、品牌语调
数据抽取	表单解析、结构化输出	模式遵从性、鲁棒性
自主智能体	研究智能体、任务自动化	可靠性、成本控制

1.2 开发技术栈

现代 LLM 应用通常由以下几层组成：

┌─────────────────────────────────────────┐
│              用户界面                    │
│  (Web, Mobile, API, Slack, CLI)         │
├─────────────────────────────────────────┤
│            应用逻辑层                    │
│  (编排, 业务规则)                        │
├─────────────────────────────────────────┤
│          LLM 编排层                      │
│  (LangChain, LlamaIndex, 原始 SDK)      │
├─────────────────────────────────────────┤
│           LLM 提供方                     │
│  (OpenAI, Anthropic, Google, 本地)      │
├─────────────────────────────────────────┤
│           支撑服务                       │
│  (向量数据库, 缓存, 检索, 工具)          │
└─────────────────────────────────────────┘

1.3 核心原则

1. 从简单开始，只在需要时增加复杂度。 一次精心设计提示词的直接 API 调用，往往比复杂的编排框架表现更好。等有了明确验证过的使用场景之后，再添加抽象层。

2. 把提示词当作代码来对待。 对提示词做版本管理，编写测试，谨慎追踪每一处改动。提示词的回归和代码回归一样致命。

3. 上线前先做评估。 LLM 的输出是非确定性的。没有系统性的评估，你就无法知道某次改动到底是提升了质量，还是让质量变差了。

4. 为失败而设计。 LLM 会产生幻觉、会超时、会返回意料之外的格式。从一开始就要构建重试逻辑、兜底方案和校验机制。

2. 提示词工程基础

2.1 提示词的结构

生产环境提示词由四个可选部分组成：

[系统指令]
你是 Acme Corp 的一位友善的客户支持坐席。
使用用户所使用的语言作答。
始终保持礼貌和简洁。绝不提及竞争对手。

[上下文 / 检索到的文档]
订单 #12345，创建于 2026-03-10。状态：已发货。
运单号：1Z999AA10123456784

[示例（few-shot）]
用户：我的订单在哪里？
助手：您的订单 #99999 已于 3 月 5 日发货，正在配送中。
预计送达日期：3 月 12 日。

[用户消息]
我上周下的单，到现在还没收到。

2.2 系统指令最佳实践

编写具备以下特征的系统指令：

角色特化: 精确定义模型是谁、目的是什么。
约束明确: 说明模型应该做什么、不应该做什么。
格式指定: 当输出格式很重要时，清楚地描述它。
语气设定: 指定正式程度、语言、长度预期。

SYSTEM_PROMPT = """你是一家金融科技公司的资深 Python 代码审查员。

职责:
- 审查正确性、安全漏洞、性能问题
- 提出具体的改进建议, 并附上代码示例
- 识别违反 GDPR/CCPA 的 PII 处理方式

输出格式:
- 以一句话的总体评价开头
- 按严重程度列出问题: [CRITICAL], [WARNING], [SUGGESTION]
- 如需修改, 以修改后的代码块结尾

不生成新功能。只审查给定的内容。"""

2.3 Few-Shot 提示

Few-shot 示例向模型展示了期望的输入-输出模式。以下场景中尤其有效：

自定义输出格式
领域专属的语气或术语
使用非标准标签的分类

FEW_SHOT_EXAMPLES = """
从下面的会议记录中提取行动项。
以 JSON 数组形式输出。

会议记录: John 将在周五前更新部署指南。
Sarah 需要在董事会会议前审查 Q1 预算。
行动项: [
  {"owner": "John", "task": "更新部署指南", "due": "周五"},
  {"owner": "Sarah", "task": "审查 Q1 预算", "due": "董事会会议前"}
]

会议记录: API 团队同意在本次冲刺中增加速率限制。
文档更新未指定负责人。
行动项: [
  {"owner": "API team", "task": "增加速率限制", "due": "本次冲刺"},
  {"owner": null, "task": "文档更新", "due": null}
]

会议记录: {meeting_text}
行动项:"""

2.4 Chain-of-Thought(CoT)

对于复杂的推理任务，要求模型在给出最终答案前先展示思考过程。

COT_PROMPT = """请逐步解决以下问题。
在每一步展示你的推理过程, 然后给出最终答案。

问题: 一位客户有 500 美元的信用额度。他下了一笔 320 美元的订单,
又退货了一件 80 美元的商品。剩余信用额度是多少?

让我们一步步思考:"""

Zero-shot CoT 触发方式：在没有示例的情况下，在提示词末尾加上"让我们一步步思考。"。这个简单的追加，就能在许多模型上显著提升多步推理的表现。

2.5 结构化输出

使用 JSON 模式或模式(schema)约束，强制模型产生可解析的输出。

from openai import OpenAI
import json

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    response_format={"type": "json_object"},
    messages=[
        {"role": "system", "content": "提取实体。只输出有效的 JSON。"},
        {"role": "user", "content": "Apple 于 2024 年 9 月 9 日在库比蒂诺发布了 iPhone 16。"}
    ]
)

data = json.loads(response.choices[0].message.content)
# {"company": "Apple", "product": "iPhone 16", "location": "库比蒂诺", "date": "2024-09-09"}

Pydantic 与 OpenAI SDK 的结构化输出功能：

from pydantic import BaseModel
from openai import OpenAI

class NewsEvent(BaseModel):
    company: str
    product: str
    location: str
    date: str

client = OpenAI()
response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "提取事件详情。"},
        {"role": "user", "content": "Apple 于 2024 年 9 月 9 日在库比蒂诺发布了 iPhone 16。"}
    ],
    response_format=NewsEvent,
)
event = response.choices[0].message.parsed
print(event.company)  # Apple

3. LLM API 与 SDK

3.1 OpenAI SDK

from openai import OpenAI

client = OpenAI(api_key="sk-...")  # 或设置 OPENAI_API_KEY 环境变量

# 基础聊天补全
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一个乐于助人的助手。"},
        {"role": "user", "content": "用 3 句话总结 Transformer 架构。"}
    ],
    temperature=0.7,
    max_tokens=200,
)

print(response.choices[0].message.content)
print(f"使用的 token 数: {response.usage.total_tokens}")

3.2 Anthropic SDK

import anthropic

client = anthropic.Anthropic(api_key="sk-ant-...")

message = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    system="你是一个乐于助人的助手。",
    messages=[
        {"role": "user", "content": "解释一下 Transformer 中的注意力机制。"}
    ]
)

print(message.content[0].text)
print(f"输入 token 数: {message.usage.input_tokens}")
print(f"输出 token 数: {message.usage.output_tokens}")

3.3 用 LiteLLM 构建统一接口

LiteLLM 为 100 多个 LLM 提供方提供了单一接口：

from litellm import completion

# OpenAI
response = completion(
    model="gpt-4o",
    messages=[{"role": "user", "content": "你好"}]
)

# Anthropic(同样的接口)
response = completion(
    model="anthropic/claude-opus-4-5",
    messages=[{"role": "user", "content": "你好"}]
)

# 本地 Ollama 模型(同样的接口)
response = completion(
    model="ollama/llama3",
    messages=[{"role": "user", "content": "你好"}]
)

print(response.choices[0].message.content)

3.4 管理对话历史

class ConversationManager:
    def __init__(self, system_prompt: str, max_history: int = 20):
        self.system_prompt = system_prompt
        self.max_history = max_history
        self.history: list[dict] = []
        self.client = OpenAI()

    def chat(self, user_message: str) -> str:
        self.history.append({"role": "user", "content": user_message})

        # 裁剪历史以避免超出上下文窗口
        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history:]

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": self.system_prompt},
                *self.history
            ]
        )

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

4. 检索增强生成(RAG)

4.1 为什么需要 RAG?

LLM 有两个根本性的局限，RAG 正是为了解决这两点而生：

知识截止日期: 模型只知道训练数据中出现过的内容。
上下文窗口限制: 模型无法一次性"知道"你所有的文档。

RAG 在推理时检索相关信息并注入到提示词中，同时解决了这两个问题。

用户查询
    │
    ▼
[查询嵌入] ──► [向量检索] ──► 前 K 个相关分块
                      │
                      ▼
            [构建增强后的提示词]
            系统: 你是一个乐于助人的助手。
            上下文: {检索到的分块}
            用户: {原始查询}
                      │
                      ▼
               [LLM 生成答案]

4.2 文档摄取流水线

from langchain.document_loaders import PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 1. 加载文档
loader = DirectoryLoader("./docs", glob="**/*.pdf", loader_cls=PyPDFLoader)
documents = loader.load()

# 2. 切分为分块
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""]
)
chunks = splitter.split_documents(documents)

# 3. 嵌入并存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)
print(f"已从 {len(documents)} 份文档中索引了 {len(chunks)} 个分块")

4.3 检索与生成

from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# 加载已有的向量存储
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small")
)

# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="mmr",          # 为提升多样性使用 Maximal Marginal Relevance
    search_kwargs={"k": 5, "fetch_k": 20}
)

# 自定义提示词
QA_PROMPT = PromptTemplate(
    template="""使用以下上下文回答问题。
如果上下文中没有答案, 就说"没有相关信息。"
不要编造信息。

上下文:
{context}

问题: {question}

答案:""",
    input_variables=["context", "question"]
)

# 链
llm = ChatOpenAI(model="gpt-4o", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    chain_type_kwargs={"prompt": QA_PROMPT},
    return_source_documents=True
)

result = qa_chain.invoke({"query": "退款政策是什么?"})
print(result["result"])
for doc in result["source_documents"]:
    print(f"出处: {doc.metadata['source']}, 第 {doc.metadata.get('page', 'N/A')} 页")

4.4 提升检索质量

混合检索 结合了密集(语义)检索与稀疏(关键词)检索：

from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

# 密集检索器(语义)
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 稀疏检索器(BM25 关键词)
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# 等权重集成
ensemble_retriever = EnsembleRetriever(
    retrievers=[dense_retriever, bm25_retriever],
    weights=[0.6, 0.4]
)

在初次检索之后，使用交叉编码器进行重排序:

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query: str, docs: list, top_k: int = 3) -> list:
    pairs = [(query, doc.page_content) for doc in docs]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, docs), key=lambda x: x[0], reverse=True)
    return [doc for _, doc in ranked[:top_k]]

5. 工具使用与函数调用

5.1 定义工具

工具让 LLM 能够调用外部 API、检索数据库或执行代码：

import json
import requests
from openai import OpenAI

client = OpenAI()

# 定义工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取某个城市当前的天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名, 例: '首尔'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "为获取最新信息而搜索网页",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                },
                "required": ["query"]
            }
        }
    }
]

5.2 处理工具调用

def get_weather(city: str, unit: str = "celsius") -> dict:
    # 实际实现会调用天气 API
    return {"city": city, "temp": 18, "unit": unit, "condition": "sunny"}

def search_web(query: str) -> str:
    # 实际实现会调用搜索 API
    return f"搜索结果: {query}"

TOOL_MAP = {
    "get_weather": get_weather,
    "search_web": search_web,
}

def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )

        message = response.choices[0].message

        # 没有工具调用 → 最终答案
        if not message.tool_calls:
            return message.content

        # 将助手带工具调用的响应加入历史
        messages.append(message)

        # 执行每一个工具调用
        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            result = TOOL_MAP[func_name](**func_args)

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            })

answer = run_agent("东京的天气怎么样? 我需要带伞吗?")
print(answer)

5.3 并行工具调用

GPT-4o 和 Claude 3 以上版本支持并行工具调用，大幅降低了独立操作的延迟：

# 模型可以一次调用多个工具
# 上面的循环已经能处理这种情况 —— message.tool_calls 是一个列表
# 天气和搜索都可以在单次模型轮次中被调用

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "比较一下东京和巴黎的天气, 再搜索一些旅行建议。"}
    ],
    tools=tools,
    parallel_tool_calls=True  # GPT-4o 中默认为 True
)

# 响应中可能同时包含 3 个工具调用: weather(东京), weather(巴黎), search(旅行建议)

6. 流式传输与异步模式

6.1 流式响应

流式传输能显著改善用户感知到的延迟，因为文本会随着生成即时显示出来：

from openai import OpenAI

client = OpenAI()

# 同步流式传输
with client.chat.completions.stream(
    model="gpt-4o",
    messages=[{"role": "user", "content": "写一个关于机器人的短篇故事。"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

6.2 用 FastAPI 实现异步流式传输

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

@app.post("/chat")
async def chat(body: dict):
    async def generate():
        async with client.chat.completions.stream(
            model="gpt-4o",
            messages=body["messages"]
        ) as stream:
            async for text in stream.text_stream:
                yield f"data: {text}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

6.3 异步批处理

在处理大量项目时，异步并发能大幅提升吞吐量：

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def classify_one(text: str, semaphore: asyncio.Semaphore) -> str:
    async with semaphore:
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "分类为 POSITIVE, NEGATIVE, 或 NEUTRAL。"},
                {"role": "user", "content": text}
            ],
            max_tokens=10
        )
        return response.choices[0].message.content.strip()

async def classify_batch(texts: list[str], max_concurrent: int = 20) -> list[str]:
    semaphore = asyncio.Semaphore(max_concurrent)
    tasks = [classify_one(text, semaphore) for text in texts]
    return await asyncio.gather(*tasks)

# 用法示例
texts = ["我真的很喜欢这个产品!", "糟糕的体验。", "还行吧。"] * 100
results = asyncio.run(classify_batch(texts))

7. 评估与测试

7.1 为什么 LLM 评估很难

传统软件测试使用确定性的断言：

assert add(2, 3) == 5  # 始终通过或失败

LLM 的输出是非确定性的，需要：

语义等价性检查(而非字符串相等)
基于评分标准(rubric)的打分
无参照答案的质量评估
统计抽样(运行一次是不够的)

7.2 LLM-as-Judge

用一个有能力的 LLM 来评估另一个 LLM 的输出：

from openai import OpenAI

client = OpenAI()

JUDGE_PROMPT = """你正在评估一个 AI 助手的响应。
请按以下标准对响应打分(每项 1-5 分):
- 准确性: 信息是否正确?
- 有用性: 是否完整地回答了问题?
- 简洁性: 是否恰当地简明扼要?

问题: {question}
响应: {response}
参考答案: {reference}

以 JSON 格式输出: {{"accuracy": X, "helpfulness": X, "conciseness": X, "reasoning": "..."}}"""

def evaluate(question: str, response: str, reference: str) -> dict:
    import json
    result = client.chat.completions.create(
        model="gpt-4o",
        response_format={"type": "json_object"},
        messages=[{
            "role": "user",
            "content": JUDGE_PROMPT.format(
                question=question,
                response=response,
                reference=reference
            )
        }]
    )
    return json.loads(result.choices[0].message.content)

7.3 评估框架

DeepEval 提供了全面的 LLM 评估指标：

from deepeval import evaluate
from deepeval.metrics import (
    AnswerRelevancyMetric,
    FaithfulnessMetric,
    ContextualRecallMetric,
)
from deepeval.test_case import LLMTestCase

test_case = LLMTestCase(
    input="法国的首都是哪里?",
    actual_output="法国的首都是巴黎。",
    expected_output="巴黎",
    retrieval_context=["法国是西欧的一个国家。首都是巴黎。"]
)

metrics = [
    AnswerRelevancyMetric(threshold=0.8),
    FaithfulnessMetric(threshold=0.9),
    ContextualRecallMetric(threshold=0.8),
]

evaluate([test_case], metrics)

7.4 用 Promptfoo 做回归测试

Promptfoo 让你可以用 YAML 定义测试用例，并在不同模型版本之间运行：

# promptfooconfig.yaml
prompts:
  - '将以下文本总结为 2 句话: {{text}}'

providers:
  - openai:gpt-4o
  - openai:gpt-4o-mini

tests:
  - vars:
      text: '埃菲尔铁塔是为 1889 年的万国博览会而建造的...'
    assert:
      - type: llm-rubric
        value: '摘要中应提及 1889 年和万国博览会'
      - type: javascript
        value: "output.split('.').length <= 3" # 最多 3 句

8. 成本优化

8.1 Token 计数与预算编制

import tiktoken

def count_tokens(text: str, model: str = "gpt-4o") -> int:
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

def estimate_cost(
    input_tokens: int,
    output_tokens: int,
    model: str = "gpt-4o"
) -> float:
    # 每 100 万 token 的价格(截至 2026 年 3 月的大致数字)
    PRICING = {
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "claude-opus-4-5": {"input": 15.00, "output": 75.00},
        "claude-haiku-3-5": {"input": 0.80, "output": 4.00},
    }
    p = PRICING.get(model, {"input": 5.0, "output": 15.0})
    return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000

8.2 提示词缓存

Anthropic 和 OpenAI 都为重复出现的系统提示词或大型上下文提供提示词缓存：

# Anthropic 提示词缓存
import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": very_long_system_prompt,  # 例如 5 万 token 的政策文档
            "cache_control": {"type": "ephemeral"}  # 缓存这个前缀
        }
    ],
    messages=[{"role": "user", "content": user_question}]
)
# 首次调用: 全价。之后的调用: 缓存 token 约 90% 折扣。

8.3 模型路由

把任务路由到能胜任的最便宜的模型：

def route_to_model(task: str, complexity: str) -> str:
    """根据任务复杂度路由到合适的模型。"""
    if complexity == "simple":
        return "gpt-4o-mini"          # 简单分类、抽取
    elif complexity == "medium":
        return "gpt-4o"               # 摘要、问答
    else:
        return "claude-opus-4-5"      # 复杂推理、代码审查

# 示例: 在路由前先分类复杂度
def smart_complete(messages: list, task_description: str) -> str:
    from openai import OpenAI
    client = OpenAI()

    # 廉价的分类步骤
    complexity = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"把这个任务的复杂度评为 'simple'、'medium' 或 'complex' 之一: {task_description}"
        }],
        max_tokens=5
    ).choices[0].message.content.strip().lower()

    model = route_to_model(task_description, complexity)

    return client.chat.completions.create(
        model=model,
        messages=messages
    ).choices[0].message.content

9. 生产环境部署

9.1 FastAPI 后端

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import AsyncOpenAI
import asyncio

app = FastAPI(title="LLM API")
client = AsyncOpenAI()

class ChatRequest(BaseModel):
    messages: list[dict]
    model: str = "gpt-4o"
    temperature: float = 0.7
    max_tokens: int = 1000

class ChatResponse(BaseModel):
    content: str
    usage: dict

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        response = await client.chat.completions.create(
            model=request.model,
            messages=request.messages,
            temperature=request.temperature,
            max_tokens=request.max_tokens,
        )
        return ChatResponse(
            content=response.choices[0].message.content,
            usage={
                "input_tokens": response.usage.prompt_tokens,
                "output_tokens": response.usage.completion_tokens
            }
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

9.2 速率限制与重试

import asyncio
import random
from functools import wraps

def with_retry(max_attempts: int = 3, base_delay: float = 1.0):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    # 带抖动的指数退避
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                    await asyncio.sleep(delay)
        return wrapper
    return decorator

@with_retry(max_attempts=3)
async def robust_completion(messages: list) -> str:
    client = AsyncOpenAI()
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )
    return response.choices[0].message.content

9.3 用 Redis 做缓存

import hashlib
import json
import redis.asyncio as redis

redis_client = redis.from_url("redis://localhost:6379")
CACHE_TTL = 3600  # 1 小时

def cache_key(messages: list, model: str) -> str:
    payload = json.dumps({"messages": messages, "model": model}, sort_keys=True)
    return f"llm:{hashlib.md5(payload.encode()).hexdigest()}"

async def cached_completion(messages: list, model: str = "gpt-4o") -> str:
    key = cache_key(messages, model)

    # 检查缓存
    cached = await redis_client.get(key)
    if cached:
        return cached.decode()

    # 生成
    from openai import AsyncOpenAI
    client = AsyncOpenAI()
    response = await client.chat.completions.create(model=model, messages=messages)
    result = response.choices[0].message.content

    # 带 TTL 存储
    await redis_client.setex(key, CACHE_TTL, result)
    return result

10. 可观测性与监控

10.1 需要追踪的核心指标

指标	重要原因	告警阈值
延迟(p50, p95, p99)	用户体验	流式传输下 p95 > 5 秒
Token 使用量	成本	预算偏差 > 20%
错误率	可靠性	请求的 > 1%
缓存命中率	成本效率	< 30%(需要排查)
评估得分	质量	相较基线下降 > 5%

10.2 LangSmith 追踪

import os
from langchain_openai import ChatOpenAI
from langchain.callbacks.tracers import LangChainTracer

os.environ["LANGCHAIN_API_KEY"] = "ls__..."
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "my-llm-app"

# 所有 LangChain 调用都会被自动追踪
llm = ChatOpenAI(model="gpt-4o")
response = llm.invoke("什么是 RAG?")
# 完整的追踪记录(提示词、响应、延迟、token)可以在 LangSmith UI 中查看

10.3 自定义日志

import time
import logging
from dataclasses import dataclass, field, asdict

logger = logging.getLogger(__name__)

@dataclass
class LLMCallLog:
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: float
    success: bool
    error: str = ""
    metadata: dict = field(default_factory=dict)

async def traced_completion(messages: list, model: str = "gpt-4o", **metadata) -> str:
    from openai import AsyncOpenAI
    client = AsyncOpenAI()

    start = time.perf_counter()
    success = True
    error = ""
    input_tokens = output_tokens = 0

    try:
        response = await client.chat.completions.create(model=model, messages=messages)
        result = response.choices[0].message.content
        input_tokens = response.usage.prompt_tokens
        output_tokens = response.usage.completion_tokens
        return result
    except Exception as e:
        success = False
        error = str(e)
        raise
    finally:
        log = LLMCallLog(
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            latency_ms=(time.perf_counter() - start) * 1000,
            success=success,
            error=error,
            metadata=metadata
        )
        logger.info("llm_call", extra=asdict(log))

10.4 护栏与安全性

from guardrails import Guard
from guardrails.hub import ToxicLanguage, DetectPII

guard = Guard().use_many(
    ToxicLanguage(threshold=0.5, on_fail="exception"),
    DetectPII(pii_entities=["EMAIL_ADDRESS", "PHONE_NUMBER"], on_fail="fix"),
)

def safe_completion(user_input: str) -> str:
    from openai import OpenAI
    client = OpenAI()

    # 校验输入
    guard.validate(user_input)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": user_input}]
    )
    output = response.choices[0].message.content

    # 校验并修正输出
    validated = guard.validate(output)
    return validated.validated_output

总结

构建生产级的 LLM 应用需要在多个维度上做到精通：

领域	核心要点
提示词工程	把提示词当作代码来对待; 做版本管理、测试、迭代
RAG	混合检索 + 重排序能大幅提升检索质量
工具使用	并行工具调用能降低多步骤任务的延迟
流式传输	对交互式 UX 至关重要; 用 FastAPI 配合 SSE 实现
评估	LLM-as-judge + 自动化测试套件能捕捉回归
成本	缓存、路由、提示词缓存可以将成本削减 80% 以上
监控	从第一天起就追踪延迟、token、质量指标

这个领域演进得很快，但无论明年是哪种模型或框架占据主导，这些基本原则都会一直有用。从简单开始，度量一切，并基于真实的使用数据来迭代。

知识确认测验

Q1. RAG 是什么，为什么有用？

RAG 是 Retrieval-Augmented Generation(检索增强生成)的缩写。它之所以有用，是因为它能让 LLM 在推理时检索相关文本并注入到提示词中，从而回答那些它从未训练过的文档相关问题。这同时解决了知识截止日期问题和上下文窗口限制问题。

Q2. LLM 智能体中并行工具调用的主要优点是什么？

并行工具调用让模型能够在单一轮次中同时调用多个工具，而不是按顺序逐个调用。这降低了多步骤任务的总体延迟，前提是这些工具调用彼此独立。

Q3. 为什么 LLM-as-judge 评估比简单的字符串匹配更受青睐？

LLM 的输出可以用许多不同的表达方式实现语义上的等价，因此字符串匹配会产生假阴性。LLM 评判者可以使用评分标准来评估语义正确性、有用性和质量，比确定性比较提供了准确得多的质量信号。

Q4. 请说出两种能在不损害质量的前提下降低 LLM API 成本的技术。

提示词缓存：缓存重复出现的大型前缀(系统提示词、参考文档)，使其只在首次被收取全价。
模型路由：将简单任务(分类、抽取)交给便宜的小模型处理，只在复杂推理任务中才使用昂贵的大模型。

目录