本文档系统阐述大型语言模型（LLM）在应用开发层面的核心技术概念，涵盖从 Token 计费到上下文工程的完整应用链路，是 01_LLM核心.md（内部机制）的应用层补充，适合作为 LLM 应用开发的核心知识参考资料。
1. Tokenization 应用层理解
1.1 定义
在应用开发层面，Tokenization 不仅是模型内部的分词过程，更是 计费单位、预算估算和上下文管理的基础。应用开发者不需要实现分词算法，但必须理解 Token 如何影响成本、延迟和功能边界。
对应 Demo: demos/01_LLM应用基础/01_Tokenization与分词.py
1.2 Token 计费模型
LLM API 的计费以 Token 为单位，而非字符或字节。理解 Token 与文本的换算关系直接决定成本估算的准确性。
计费公式:
总成本 = (输入 Token 数 × 输入单价) + (输出 Token 数 × 输出单价)
示例 (GPT-4 定价):
输入: $0.03 / 1K tokens
输出: $0.06 / 1K tokens
用户输入 500 tokens, 模型输出 200 tokens:
成本 = 500 × 0.03/1000 + 200 × 0.06/1000
= 0.015 + 0.012
= $0.0271.3 不同文本类型的 Token 换算
不同语言和格式的文本，Token 效率差异巨大：
关键规律:
英文: 1 token ≈ 4 字符 (效率最高)
中文: 1 字 ≈ 1-2 token (成本约为英文 4-8 倍)
Emoji: 1 个 ≈ 1-3 token (尽量避免大量使用)
代码: 1 token ≈ 2-3 字符 (符号多, 效率较低)1.4 Token 预算估算
def estimate_token_budget(text: str, model: str = "gpt-3.5-turbo") -> dict:
"""估算文本的 Token 数量和成本"""
enc = tiktoken.get_encoding("cl100k_base")
tokens = enc.encode(text)
token_count = len(tokens)
# 不同模型的上下文窗口
context_windows = {
"gpt-3.5-turbo": 4096,
"gpt-4": 8192,
"gpt-4-turbo": 128000,
"claude-3": 200000,
}
max_tokens = context_windows.get(model, 4096)
usage_rate = token_count / max_tokens
return {
"token_count": token_count,
"max_tokens": max_tokens,
"usage_rate": f"{usage_rate:.1%}",
"remaining": max_tokens - token_count,
}1.5 应用层最佳实践
成本预估：在发送请求前预估 Token 数，避免意外高额账单
上下文规划：根据模型窗口大小规划对话历史长度
多语言优化：中文场景下 Token 成本约为英文 4-8 倍，翻译策略需权衡
输入预处理：去除冗余空白、HTML 标签等无意义内容以节省 Token
1.6 与其他概念的关联
-> 上下文窗口管理：Token 数量决定是否超出窗口限制
-> LLM API 调用：Token 是 API 计费和限流的基本单位
-> Context Engineering：Token 预算分配是上下文工程的核心约束
2. Embedding 与语义搜索
2.1 定义
Embedding（嵌入）在应用层用于将文本转换为高维向量，使语义相近的文本在向量空间中距离更近。语义搜索（Semantic Search）基于向量相似度匹配查询与文档，超越传统关键词匹配的限制，能理解"意思"而非"字面"。
对应 Demo: demos/01_LLM应用基础/02_Embedding与向量.py
2.2 核心原理
2.2.1 文本向量化
文本 -> Embedding 模型 -> 高维向量 (如 1536 维)
"机器学习" -> [0.12, -0.34, 0.56, ..., 0.78] (1536维)
"深度学习" -> [0.11, -0.32, 0.55, ..., 0.77] (向量相近)
"天气预报" -> [0.89, 0.23, -0.67, ..., 0.01] (向量差异大)2.2.2 相似度计算
余弦相似度（最常用）：
值域为 [-1, 1]，越接近 1 表示语义越相似。
def cosine_similarity(vec_a: list[float], vec_b: list[float]) -> float:
"""计算两个向量的余弦相似度"""
dot = sum(a * b for a, b in zip(vec_a, vec_b))
norm_a = math.sqrt(sum(a * a for a in vec_a))
norm_b = math.sqrt(sum(b * b for b in vec_b))
if norm_a == 0 or norm_b == 0:
return 0.0
return dot / (norm_a * norm_b)2.2.3 其他相似度度量
2.3 语义搜索流程
┌─────────────────────────────────────────────────────────────┐
│ 语义搜索完整流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 离线建库: │
│ 文档库 -> 分块 -> Embedding -> 向量存储 │
│ ↓ ↓ ↓ │
│ ["文档1块1"] [0.1,...] 向量数据库 │
│ ["文档1块2"] [0.2,...] (Qdrant/Chroma) │
│ ["文档2块1"] [0.3,...] │
│ │
│ 在线查询: │
│ 用户问题 -> Embedding -> 向量检索 -> Top-K -> 排序 -> 结果 │
│ "如何训练模型" [0.15,...] 比较相似度 [块1,块3,块7] │
│ │
└─────────────────────────────────────────────────────────────┘2.4 语义搜索 vs 关键词搜索
2.5 实际应用场景
场景1: 语义文档检索
用户: "怎么提高模型准确率"
关键词搜索: 找不到（文档中没有"提高"和"准确率"）
语义搜索: 找到"优化模型性能的方法"（语义匹配）
场景2: 智能客服
用户: "退货流程是什么"
语义搜索: 匹配到"退换货政策"相关文档
场景3: 代码搜索
用户: "排序算法"
语义搜索: 匹配到 sort(), quicksort, merge_sort 相关代码2.6 与其他概念的关联
<- Tokenization：Embedding 以 Token 为输入
-> 上下文窗口管理：检索结果需要注入上下文窗口
-> RAG：语义搜索是 RAG 系统的核心检索组件
3. 上下文窗口管理
3.1 定义
上下文窗口（Context Window）是模型一次推理能处理的最大 Token 数。应用开发者需要在有限窗口内合理分配系统提示、对话历史、检索内容和响应空间。上下文窗口管理是在有限 Token 预算下最大化信息价值的关键工程。
对应 Demo: demos/01_LLM应用基础/03_上下文窗口管理.py
3.2 Token 预算分配
上下文窗口 (如 4096 tokens)
┌──────────────────────────────────────────────┐
│ System Prompt │ 系统提示词 (固定) │ ~500 tokens
├───────────────────┼──────────────────────────┤
│ Conversation │ 对话历史 (可变) │ ~2500 tokens
│ History │ (多轮对话消息) │
├───────────────────┼──────────────────────────┤
│ RAG Context │ 检索增强内容 (可变) │ ~500 tokens
├───────────────────┼──────────────────────────┤
│ Reserved for │ 模型响应预留 (必须保留) │ ~596 tokens
│ Response │ │
└──────────────────────────────────────────────┘3.3 消息截断策略
当对话历史超出预算时，需要截断旧消息：
class ContextWindow:
"""管理对话上下文的 Token 预算"""
def __init__(self, max_tokens: int = 4096, reserved_for_response: int = 500):
"""初始化窗口大小和响应预留空间"""
self.max_tokens = max_tokens
self.reserved_for_response = reserved_for_response
self.messages: list[dict[str, str]] = []
def get_budget(self) -> int:
"""获取可用于历史消息的 Token 预算"""
return self.max_tokens - self.reserved_for_response
def add_message(self, role: str, content: str) -> None:
"""添加消息，超预算则触发截断"""
self.messages.append({"role": role, "content": content})
while self.get_current_usage() > self.get_budget() and len(self.messages) > 1:
self._truncate_oldest()
def _truncate_oldest(self) -> None:
"""截断最旧的非系统消息（保护 System Prompt）"""
for i, msg in enumerate(self.messages):
if msg["role"] != "system":
self.messages.pop(i)
break3.4 截断策略对比
3.5 摘要压缩
def summarize_messages(messages: list[dict], llm_handler) -> str:
"""将旧消息列表压缩为摘要"""
conversation = "\n".join(f"{m['role']}: {m['content'][:100]}" for m in messages)
prompt = f"请将以下对话历史压缩为简洁摘要，保留关键信息:\n{conversation}"
return llm_handler(prompt)截断 + 摘要组合策略:
原始历史 (8条消息, 3000 tokens):
[system] [user1] [ai1] [user2] [ai2] [user3] [ai3] [user4]
Step 1: 保护 system, 截断 user1~ai2
Step 2: 将 user1~ai2 摘要为 summary (200 tokens)
Step 3: 保留 [system] [summary] [user3] [ai3] [user4]
结果: 3000 tokens -> 1200 tokens (节省 60%)3.6 不同模型的上下文窗口
3.7 与其他概念的关联
<- Tokenization：Token 计数是窗口管理的基础
-> Context Engineering：窗口管理是上下文工程的具体实现
-> Function Calling：工具调用的结果也占用窗口预算
4. 采样参数实战
4.1 定义
采样参数（Temperature / Top-k / Top-p）控制 LLM 生成过程中的随机性和多样性。在应用开发中，不同任务需要不同的采样配置，正确组合使用这些参数是保证输出质量和适用性的关键。
对应 Demo: demos/01_LLM应用基础/04_温度与采样策略.py
4.2 三个核心参数
4.2.1 Temperature（温度）
Temperature 调整概率分布的"锐度":
原始 Logits: [3.0, 2.0, 1.0, 0.5]
T = 0.0: [1.00, 0.00, 0.00, 0.00] <- 完全确定 (贪婪)
T = 0.7: [0.70, 0.22, 0.05, 0.03] <- 偏向最优，适度随机
T = 1.0: [0.64, 0.24, 0.09, 0.03] <- 原始分布
T = 1.5: [0.42, 0.31, 0.19, 0.08] <- 更随机，更有创意4.2.2 Top-k（固定截断）
Top-k = 5: 只从概率最高的 5 个 Token 中采样
原始分布: A(0.30) B(0.25) C(0.15) D(0.10) E(0.08) F(0.05) G(0.03) ...
Top-5: A(0.34) B(0.28) C(0.17) D(0.11) E(0.09) <- 重新归一化
其余: 置 0，不参与采样4.2.3 Top-p（动态截断 / 核采样）
Top-p = 0.9: 保留累积概率达到 0.9 的最小集合
排序: A(0.45) B(0.25) C(0.12) D(0.08) E(0.05) ...
累积: 0.45 0.70 0.82 0.90 <- 达到 0.9, 截断
核集合: {A, B, C, D} -> 重新归一化后采样4.3 参数组合处理流程
完整采样流水线:
模型输出 Logits
|
v
+-------------+
| Temperature | 调整分布锐度 (logits / T -> softmax)
+------+------+
|
v
+-------------+
| Top-k | 截断为概率最高的 k 个候选
+------+------+
|
v
+-------------+
| Top-p | 从累积概率达 p 的核集合中采样
+------+------+
|
v
最终 Tokendef apply_temperature(logits: list[float], temperature: float) -> list[float]:
"""应用温度缩放：高温更随机，低温更确定"""
scaled = [l / max(temperature, 0.01) for l in logits]
return softmax(scaled)
def top_k_sampling(logits: list[float], k: int, temperature: float = 1.0) -> list[float]:
"""Top-k 采样：只保留概率最高的 k 个 token"""
probs = apply_temperature(logits, temperature)
indexed = sorted(enumerate(probs), key=lambda x: x[1], reverse=True)
top_indices = set(idx for idx, _ in indexed[:k])
filtered = [p if i in top_indices else 0.0 for i, p in enumerate(probs)]
total = sum(filtered)
return [p / total for p in filtered] if total > 0 else filtered
def top_p_sampling(logits: list[float], p: float, temperature: float = 1.0) -> list[float]:
"""Top-p 核采样：保留累计概率达到 p 的最小 token 集合"""
probs = apply_temperature(logits, temperature)
indexed = sorted(enumerate(probs), key=lambda x: x[1], reverse=True)
cumulative = 0.0
keep_indices = set()
for idx, prob in indexed:
cumulative += prob
keep_indices.add(idx)
if cumulative >= p:
break
filtered = [p if i in keep_indices else 0.0 for i, p in enumerate(probs)]
total = sum(filtered)
return [p / total for p in filtered] if total > 0 else filtered4.4 场景化参数推荐
4.5 实际应用中的注意事项
注意事项:
1. Temperature = 0 不等于完全确定性
- 不同 API 实现可能略有差异
- 浮点精度可能导致微小变化
2. Top-k 和 Top-p 可以同时使用
- 处理顺序: Temperature -> Top-k -> Top-p
- 两者叠加会进一步缩小候选集
3. 参数选择的影响
- 过低温度: 输出重复、呆板
- 过高温度: 输出不连贯、幻觉
- 建议从推荐值开始，根据效果微调
4. 流式输出中的采样
- 采样参数在每一步 Token 生成都生效
- 每步独立采样，不影响后续步的分布4.6 与其他概念的关联
<- LLM核心.md 第7-10节：采样参数的底层原理
-> LLM API 调用：采样参数通过 API 请求传递
-> 结构化输出：低温度 + Top-p 确保 JSON 输出可靠性
5. Function Calling（工具函数调用）
5.1 定义
Function Calling 是 LLM 与外部世界交互的核心机制。LLM 根据用户意图决定调用哪个工具函数、生成调用参数，系统执行函数后将结果返回给 LLM，LLM 再基于结果生成最终回复。这使得 LLM 能访问实时数据、执行计算、操作外部系统。
对应 Demo: demos/01_LLM应用基础/05_Function_Calling.py
5.2 完整调用流程
+------------------------------------------------------------------+
| Function Calling 完整流程 |
+------------------------------------------------------------------+
| |
| 用户: "北京今天天气怎么样?" |
| | |
| v |
| +-------------+ |
| | 1. 构建请求 | system + user message + tool_definitions |
| +------+------+ |
| | |
| v |
| +------------------+ |
| | 2. LLM 推理 | 模型分析意图，决定调用 search_web 工具 |
| | 返回 tool_call| {name: "search_web", args: {query:"..."}} |
| +------+-----------+ |
| | |
| v |
| +------------------+ |
| | 3. 参数校验 | 校验 query 参数类型和必填性 |
| +------+-----------+ |
| | |
| v |
| +------------------+ |
| | 4. 执行函数 | search_web("北京天气") -> "北京晴, 25 C" |
| +------+-----------+ |
| | |
| v |
| +------------------+ |
| | 5. 结果返回 | 将函数结果作为 tool message 送回 LLM |
| +------+-----------+ |
| | |
| v |
| +------------------+ |
| | 6. LLM 生成回复 | "北京今天晴天，气温25度，适合户外活动。" |
| +------------------+ |
| |
+------------------------------------------------------------------+5.3 工具定义格式（OpenAI 标准）
TOOL_DEFINITIONS = [
{
"type": "function",
"function": {
"name": "search_web",
"description": "搜索网络获取实时信息",
"parameters": {
"type": "object",
"properties": {"query": {"type": "string", "description": "搜索关键词"}},
"required": ["query"],
},
},
}
]5.4 工具定义的关键要素
5.5 参数校验与错误处理
def execute_tool(tool_name: str, arguments: dict, tools: dict) -> str:
"""执行工具调用，包含参数校验和错误处理"""
if tool_name not in tools:
return f"错误: 未知工具 '{tool_name}'"
tool = tools[tool_name]
handler = tool["handler"]
params = tool.get("parameters", {})
# 参数校验
for param_name, param_def in params.items():
if param_def.get("required") and param_name not in arguments:
return f"错误: 缺少必填参数 '{param_name}'"
# 类型检查
for param_name, value in arguments.items():
expected_type = params.get(param_name, {}).get("type", "string")
if not check_type(value, expected_type):
return f"错误: 参数 '{param_name}' 类型错误"
# 执行
try:
result = handler(**arguments)
return str(result)
except Exception as e:
return f"执行错误: {e}"5.6 多工具选择
用户: "帮我查一下天气，然后算一下 25 * 37 等于多少"
LLM 决策:
Step 1: 调用 search_web(query="今天天气")
Step 2: 调用 calculate(expression="25 * 37")
Step 3: 综合两个结果生成回复
多工具调用模式:
- 串行: 工具 B 依赖工具 A 的结果
- 并行: 工具 A 和 B 独立执行
- 条件: 根据前一个工具结果决定是否调用下一个5.7 与其他概念的关联
-> Agent 核心能力：Function Calling 是 Agent 调用工具的基础
-> 结构化输出：工具参数本身就是一种结构化输出
<- 上下文窗口管理：工具结果占用上下文窗口预算
6. 结构化输出
6.1 定义
结构化输出确保 LLM 返回可解析的格式化数据（如 JSON），而非自由文本。通过 JSON Mode、Schema 约束和输出验证，将不可靠的自然语言转换为程序可信赖的结构化对象，是 LLM 与传统软件系统集成的关键桥梁。
对应 Demo: demos/01_LLM应用基础/06_结构化输出.py
6.2 为什么需要结构化输出
无结构化输出 (不可靠):
用户: "提取这句话的信息: 张三，25岁，北京"
LLM: "这个人叫张三，今年25岁，住在北京市。"
-> 程序无法直接解析，需正则提取，容易出错
有结构化输出 (可靠):
用户: "提取这句话的信息: 张三，25岁，北京" (要求 JSON)
LLM: {"name": "张三", "age": 25, "city": "北京"}
-> 程序直接 json.loads() 解析，类型安全6.3 实现方式
6.3.1 JSON Mode
# 在 API 请求中指定 response_format
response = client.chat.completions.create(
model="gpt-4",
messages=[...],
response_format={"type": "json_object"}, # 强制 JSON 输出
)6.3.2 Schema 验证
@dataclass
class FieldSchema:
"""字段模式定义，模拟 Pydantic 的字段约束"""
name: str
field_type: str # str, int, float, bool, list, dict
required: bool = True
default: Any = None
description: str = ""
enum: list[str] | None = None
@dataclass
class ModelSchema:
"""模型模式定义"""
name: str
fields: list[FieldSchema] = field(default_factory=list)
def validate(self, data: dict) -> tuple[bool, str, dict]:
"""验证数据是否符合 schema"""
result = {}
for f in self.fields:
if f.name not in data:
if f.required:
return False, f"缺少必填字段: {f.name}", {}
result[f.name] = f.default
continue
val = data[f.name]
type_map = {"str": str, "int": int, "float": (int, float), "bool": bool, "list": list, "dict": dict}
expected = type_map.get(f.field_type)
if expected and not isinstance(val, expected):
return False, f"类型错误: 期望{f.field_type}", {}
if f.enum and val not in f.enum:
return False, f"值 {val} 不在允许范围 {f.enum} 内", {}
result[f.name] = val
return True, "", result6.4 输出解析与修复
def parse_llm_output(raw_output: str, schema: ModelSchema) -> dict:
"""解析 LLM 输出并验证，支持自动修复"""
# 1. 尝试直接解析 JSON
try:
data = json.loads(raw_output)
except json.JSONDecodeError:
# 2. 尝试从文本中提取 JSON
json_match = re.search(r"\{[^{}]*\}", raw_output, re.DOTALL)
if json_match:
data = json.loads(json_match.group())
else:
raise ValueError("无法解析为 JSON")
# 3. Schema 验证
success, error, cleaned = schema.validate(data)
if not success:
raise ValueError(f"Schema 验证失败: {error}")
return cleaned6.5 结构化输出的层次
可靠性递增:
Level 1: 自然语言提示 ("请返回JSON格式")
-> 模型可能不遵循，格式不稳定
Level 2: JSON Mode (response_format)
-> 保证输出是合法 JSON，但不保证字段结构
Level 3: Schema 约束 (Function Calling / Structured Output)
-> 模型按指定 Schema 输出，字段名和类型有保障
Level 4: Schema 验证 + 自动重试
-> 输出后验证，失败则重试，最高可靠性6.6 实际应用场景
6.7 与其他概念的关联
<- Function Calling：工具参数依赖结构化输出
-> Few-Shot：示例可引导结构化输出格式
-> Prompt Chaining：链中每步的结构化输出是下一步的输入
7. System Prompt 设计
7.1 定义
System Prompt 是 LLM 的"操作系统"，定义角色身份、能力边界、行为约束和输出格式。优秀的系统提示词让模型行为可预测、可控制；差的提示词导致幻觉和越界行为。System Prompt 拥有最高指令优先级，能约束后续所有用户输入。
对应 Demo: demos/01_LLM应用基础/07_System_Prompt设计.py
7.2 指令层级体系
指令优先级 (从高到低):
+-------------------------------------------+
| System Prompt (最高优先级) | 定义身份、边界、规则
| "你是一个专业的客服助手，只能回答..." |
+-------------------------------------------+
| Task Prompt (中优先级) | 定义当前任务要求
| "请总结以下文档的核心要点..." |
+-------------------------------------------+
| User Input (低优先级) | 用户实际输入
| "帮我写一首诗" |
+-------------------------------------------+
规则: 高层级指令可以覆盖低层级指令
System Prompt 约束所有后续交互7.3 系统提示词的结构化构建
class SystemPromptBuilder:
"""结构化系统提示词构建器"""
def __init__(self):
self.role: str = ""
self.capabilities: list[str] = []
self.constraints: list[str] = []
self.output_format: str = ""
self.examples: list[dict[str, str]] = []
def set_role(self, persona: str, expertise: str) -> SystemPromptBuilder:
"""设定角色身份和专业领域"""
self.role = f"你是一个{persona}，擅长{expertise}。"
return self
def add_constraint(self, constraint: str) -> SystemPromptBuilder:
"""添加行为约束"""
self.constraints.append(constraint)
return self
def build(self) -> str:
"""组装完整的系统提示词"""
sections = [self.role]
if self.capabilities:
caps = "\n".join(f" - {c}" for c in self.capabilities)
sections.append(f"【你的能力】\n{caps}")
if self.constraints:
cons = "\n".join(f" - {c}" for c in self.constraints)
sections.append(f"【行为约束】\n{cons}")
if self.output_format:
sections.append(f"【输出格式】\n{self.output_format}")
return "\n\n".join(sections)7.4 系统提示词的五大组件
+-----------------------------------------------------------+
| 优秀 System Prompt 的结构 |
+-----------------------------------------------------------+
| |
| 1. 角色定义 (Role) |
| "你是一个资深 Python 开发工程师" |
| -> 设定专业领域和知识边界 |
| |
| 2. 能力描述 (Capabilities) |
| "你可以：编写代码、审查代码、解释技术概念" |
| -> 明确能做什么 |
| |
| 3. 行为约束 (Constraints) |
| "你不能：执行代码、访问网络、讨论非技术话题" |
| -> 明确不能做什么（安全边界） |
| |
| 4. 输出格式 (Output Format) |
| "回答时请使用 Markdown 格式，代码块标注语言" |
| -> 规范输出结构 |
| |
| 5. 示例引导 (Examples) |
| "示例：用户问X，你回答Y" |
| -> 通过示例锚定行为模式 |
| |
+-----------------------------------------------------------+7.5 常见设计模式
7.6 Token 预算考量
System Prompt 的 Token 成本:
每次请求都会发送 System Prompt
-> 如果 System Prompt 1000 tokens，1000 次请求 = 1M tokens
优化策略:
1. 精简描述，去除冗余
2. 使用列表而非段落（更紧凑）
3. 将固定规则缓存（Prompt Caching）
4. 分层设计：核心规则放 System，场景规则放 Task7.7 与其他概念的关联
-> Few-Shot 与提示模板：示例是 System Prompt 的组件之一
-> 上下文窗口管理：System Prompt 占用固定 Token 预算
-> Context Engineering：System Prompt 是上下文中最高优先级的内容
8. Few-Shot 与提示模板
8.1 定义
Few-Shot Learning 通过在提示词中提供少量输入-输出示例来引导模型行为，使其无需微调即可适应新任务。提示模板（Prompt Template）将变量与固定文本组合，实现可复用、可管理的提示词工程。两者结合构成了应用层提示工程的核心工具。
对应 Demo: demos/01_LLM应用基础/08_Few_Shot与提示模板.py
8.2 Few-Shot 原理
Zero-Shot (零样本):
"判断情感：'这个产品太棒了'"
-> 模型可能不确定输出格式
Few-Shot (少样本):
"判断情感：
输入: '太难用了' -> 输出: 负面
输入: '非常满意' -> 输出: 正面
输入: '一般般' -> 输出: 中性
输入: '这个产品太棒了' -> 输出: ?"
-> 模型学会格式和分类标准，输出: 正面8.3 示例数据结构
@dataclass
class FewShotExample:
"""单个 Few-Shot 示例: 输入 + 输出 + 可选解释"""
input: str
output: str
explanation: str = ""
def render(self) -> str:
"""渲染为提示词文本"""
text = f"输入: {self.input}\n输出: {self.output}"
if self.explanation:
text += f"\n(理由: {self.explanation})"
return text8.4 示例选择策略
class ExampleSelector:
"""示例选择器：根据输入动态选择最相关的示例"""
def __init__(self, examples: list[FewShotExample], max_examples: int = 3):
self.examples = examples
self.max_examples = max_examples
def select(self, query: str) -> list[FewShotExample]:
"""选择与查询最相似的示例"""
scored = [(self._similarity(query, ex.input), ex) for ex in self.examples]
scored.sort(key=lambda x: x[0], reverse=True)
return [ex for _, ex in scored[: self.max_examples]]8.5 提示模板
class PromptTemplate:
"""带变量替换的提示词模板"""
def __init__(self, name: str, template: str, input_variables: list[str]):
self.name = name
self.template = template
self.input_variables = input_variables
def render(self, **kwargs) -> str:
"""用实际值填充模板变量"""
missing = set(self.input_variables) - set(kwargs.keys())
if missing:
raise ValueError(f"缺少变量: {missing}")
return self.template.format(**kwargs)
def token_count(self, **kwargs) -> int:
"""渲染后统计 Token 数"""
enc = tiktoken.get_encoding("cl100k_base")
return len(enc.encode(self.render(**kwargs)))8.6 模板管理最佳实践
模板库管理:
templates/
+-- classification/
| +-- sentiment.txt # 情感分类模板
| +-- intent.txt # 意图分类模板
+-- extraction/
| +-- entity.txt # 实体抽取模板
| +-- relation.txt # 关系抽取模板
+-- generation/
+-- summary.txt # 摘要生成模板
+-- code_review.txt # 代码审查模板
版本管理:
- 每个模板标注版本号和变更记录
- A/B 测试不同模板的效果
- 模板变量需有默认值和类型约束8.7 Few-Shot vs Fine-Tuning
8.8 与其他概念的关联
<- System Prompt 设计：Few-Shot 示例可作为 System Prompt 组件
-> Prompt Chaining：每步链使用独立模板
-> 结构化输出：示例可引导结构化输出格式
9. Prompt Chaining（链式提示）
9.1 定义
Prompt Chaining 将复杂任务拆解为多个步骤，前一步的输出作为后一步的输入，形成链式处理流程。支持顺序链、并行分支和条件路由，实现可组合、可调试的 LLM 工作流。这是从单次 LLM 调用走向复杂 Agent 工作流的过渡技术。
对应 Demo: demos/01_LLM应用基础/09_Prompt_Chaining.py
9.2 链式架构
+----------------------------------------------------------+
| Prompt Chaining 架构 |
+----------------------------------------------------------+
| |
| 顺序链 (Sequential Chain): |
| |
| Step 1 Step 2 Step 3 Step 4 |
| +-----+ +-----+ +-----+ +-----+ |
| |理解 |--->|检索 |--->|分析 |--->|生成 | |
| |意图 | |信息 | |数据 | |回复 | |
| +-----+ +-----+ +-----+ +-----+ |
| |
| 并行分支 (Parallel Branches): |
| |
| +-----+ |
| +--->|分支A |---+ |
| +-----+| +-----+ |+-----+ |
| |输入 | ||合并 | |
| +-----+| +-----+ |+-----+ |
| +--->|分支B |---+ |
| +-----+ |
| |
| 条件路由 (Conditional Routing): |
| |
| +-----+ +---------+ |
| |输入 |--->|分类判断 |---> 条件A -> Step A |
| +-----+ +----+----+---> 条件B -> Step B |
| | +-> 默认 -> Step C |
| v |
+----------------------------------------------------------+9.3 链步骤定义
@dataclass
class ChainStep:
"""链中的一个步骤"""
name: str
prompt_template: str
input_keys: list[str]
output_key: str
handler: Callable[[str], str] | None = None
retry_count: int = 0
fallback: str | None = None
def execute(self, context: dict[str, Any]) -> str:
"""执行步骤：渲染提示词 -> 调用处理器 -> 返回结果"""
prompt = self.render_prompt(context)
for attempt in range(self.retry_count + 1):
try:
if self.handler:
return self.handler(prompt)
return f"[模拟输出] {self.name} 已处理"
except Exception:
if attempt < self.retry_count:
continue
if self.fallback is not None:
return self.fallback
raise9.4 提示链管理
class PromptChain:
"""提示链：管理步骤序列和中间结果传递"""
def __init__(self, name: str = "chain"):
self.name = name
self.steps: list[ChainStep] = []
self.context: dict[str, Any] = {}
def add_step(self, step: ChainStep) -> PromptChain:
"""添加步骤到链尾"""
self.steps.append(step)
return self
def run(self, initial_context: dict[str, Any]) -> dict[str, Any]:
"""执行整条链"""
self.context.update(initial_context)
for step in self.steps:
output = step.execute(self.context)
self.context[step.output_key] = output
return self.context9.5 链式 vs 单次调用
9.6 实际应用示例
场景: 智能文档分析
Chain:
Step 1 [提取关键信息] -> "文档主题、关键实体、核心论点"
|
v
Step 2 [情感分析] -> "整体情感倾向、情感强度"
|
v
Step 3 [风险评估] -> "风险等级、风险因素、建议措施"
|
v
Step 4 [生成报告] -> "结构化分析报告"
优势:
- 每步可独立测试和优化
- 中间结果可用于其他链
- 某步失败可重试而不影响其他步9.7 错误处理与容错
# 链式调用的三种容错策略
# 1. 重试机制
step.retry_count = 3 # 最多重试 3 次
# 2. 兜底输出
step.fallback = "无法处理此步骤，请人工检查"
# 3. 条件跳过
def should_skip(context):
return context.get("skip_analysis", False)9.8 与其他概念的关联
-> Agent 核心能力：Prompt Chaining 是 Agent 任务分解的简化版
<- Few-Shot 与提示模板：每步链使用独立模板
-> LangGraph：LangGraph 是 Prompt Chaining 的图化升级
10. LLM API 调用
10.1 定义
LLM API 调用是应用层与大模型交互的核心接口，涵盖请求构建、流式响应、错误重试、Token 计费和多模型管理。正确封装 API 调用层是构建可靠 LLM 应用的基础设施。
对应 Demo: demos/01_LLM应用基础/10_LLM_API调用.py
10.2 API 调用完整流程
+----------------------------------------------------------+
| LLM API 调用完整流程 |
+----------------------------------------------------------+
| |
| 1. 构建请求 |
| - 选择模型 (model) |
| - 组装消息 (messages: system + user + assistant) |
| - 设置参数 (temperature, max_tokens, tools) |
| - 选择模式 (stream: true/false) |
| |
| 2. 发送请求 |
| - HTTP POST -> /v1/chat/completions |
| - 认证 (Authorization: Bearer {api_key}) |
| - 超时设置 |
| |
| 3. 处理响应 |
| - 非流式: 等待完整响应 -> 解析 JSON |
| - 流式: SSE 逐 Token 接收 -> 拼接文本 |
| |
| 4. 错误处理 |
| - 429 Rate Limit -> 指数退避重试 |
| - 500 Server Error -> 重试 |
| - 400 Bad Request -> 不重试，返回错误 |
| |
| 5. 成本追踪 |
| - 记录 input_tokens, output_tokens |
| - 计算费用 |
| - 累计统计 |
| |
+----------------------------------------------------------+10.3 流式输出
def stream_chat(messages: list[dict], model: str = "gpt-4") -> Generator:
"""流式输出：逐 Token 返回，前端实时展示"""
stream = client.chat.completions.create(
model=model,
messages=messages,
stream=True, # 开启流式
)
for chunk in stream:
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content流式 vs 非流式:
非流式:
用户请求 -> [等待 5 秒] -> 完整回复一次性返回
体验: 等待时间长，无进度反馈
流式 (SSE):
用户请求 -> "你" -> "好" -> "！" -> "今天" -> "天气" -> ...
体验: 实时打字效果，首 Token 延迟低
关键指标:
- TTFT (Time To First Token): 首 Token 延迟
- TPOT (Time Per Output Token): 每 Token 生成时间10.4 错误重试机制
def retry_with_backoff(func, max_retries=3, base_delay=1.0):
"""指数退避重试"""
for attempt in range(max_retries):
try:
return func()
except RateLimitError:
delay = base_delay * (2**attempt) # 1, 2, 4 秒
time.sleep(delay)
except ServerError:
if attempt < max_retries - 1:
time.sleep(base_delay)
continue
raise错误处理决策树:
收到错误
+-- 429 (Rate Limit)
| +-- 指数退避重试 (1s -> 2s -> 4s)
+-- 500 (Server Error)
| +-- 重试 2-3 次
+-- 400 (Bad Request)
| +-- 不重试，修正请求
+-- 401 (Auth Error)
| +-- 不重试，检查 API Key
+-- Timeout
+-- 重试，增加超时时间10.5 成本追踪
@dataclass
class TokenUsage:
"""Token 使用量和成本追踪"""
input_tokens: int = 0
output_tokens: int = 0
model: str = "gpt-3.5-turbo"
total_calls: int = 0
@property
def cost(self) -> float:
"""计算总成本（美元）"""
pricing = MODEL_PRICING.get(self.model, {"input": 0, "output": 0})
return self.input_tokens * pricing["input"] + self.output_tokens * pricing["output"]
def report(self) -> str:
"""生成成本报告"""
return (
f"调用次数: {self.total_calls}\n"
f"输入Token: {self.input_tokens}\n"
f"输出Token: {self.output_tokens}\n"
f"总Token: {self.total_tokens}\n"
f"总成本: ${self.cost:.4f}"
)10.6 模型定价对比
10.7 多模型管理
class ModelRouter:
"""根据任务复杂度路由到不同模型"""
def select_model(self, task_type: str, token_estimate: int) -> str:
"""选择最合适的模型"""
if task_type == "complex_reasoning":
return "gpt-4"
elif token_estimate > 4096:
return "claude-3" # 长上下文
else:
return "gpt-3.5-turbo" # 经济选择10.8 与其他概念的关联
<- Tokenization：API 以 Token 为计费单位
-> Context Engineering：API 调用前需做上下文预算
-> Agent 核心能力：Agent 的每轮推理都是一次 API 调用
11. Context Engineering（上下文工程）
11.1 定义
Context Engineering 是管理 LLM 上下文窗口的工程实践，通过优先级排序、Token 预算分配和信息压缩，在有限窗口内最大化信息价值。它不是简单的"把所有信息塞给模型"，而是系统性地决定"什么信息进入上下文、以什么顺序、占多少预算"。
对应 Demo: demos/01_LLM应用基础/11_Context_Engineering.py
11.2 上下文优先级体系
上下文信息优先级 (从高到低):
+--------------------------------------------------+
| Priority: CRITICAL (4) |
| - System Prompt (角色定义、安全约束) |
| - 当前用户指令 |
| -> 必须保留，不可压缩 |
+--------------------------------------------------+
| Priority: HIGH (3) |
| - RAG 检索结果（与问题直接相关） |
| - 关键工具调用结果 |
| -> 优先保留，必要时摘要 |
+--------------------------------------------------+
| Priority: NORMAL (2) |
| - 近期对话历史 |
| - 任务上下文 |
| -> 按预算截断或摘要 |
+--------------------------------------------------+
| Priority: LOW (1) |
| - 远期对话历史 |
| - 补充信息 |
| -> 首先被截断或丢弃 |
+--------------------------------------------------+class Priority(IntEnum):
"""上下文优先级：数值越大优先级越高"""
LOW = 1
NORMAL = 2
HIGH = 3
CRITICAL = 4
@dataclass
class ContextItem:
"""单条上下文：内容 + 优先级 + Token数 + 来源"""
content: str
priority: Priority
source: str # system / task / rag / memory / history
token_count: int = 0
compressed: bool = False11.3 Token 预算分配
@dataclass
class TokenBudget:
"""Token 预算分配: system(20%) + task(10%) + RAG(40%) + history(30%)"""
total: int = 4096
@property
def system(self) -> int:
return int(self.total * 0.20) # 819 tokens
@property
def task(self) -> int:
return int(self.total * 0.10) # 409 tokens
@property
def rag(self) -> int:
return int(self.total * 0.40) # 1638 tokens
@property
def history(self) -> int:
return int(self.total * 0.30) # 1228 tokens预算分配可视化 (4096 tokens):
系统 20% ######## 819
任务 10% #### 409
RAG 40% ################ 1638
历史 30% ############ 1228
响应预留 (从总预算中扣除)
分配原则:
- System: 固定占比，不可压缩
- Task: 当前任务指令，简短但关键
- RAG: 检索结果是最有价值的信息源
- History: 近期对话，可压缩11.4 上下文组装流程
+----------------------------------------------------------+
| 上下文工程组装流程 |
+----------------------------------------------------------+
| |
| 输入源: |
| +----------+ +----------+ +----------+ +----------+ |
| |System | |Task | |RAG | |History | |
| |Prompt | |Instruction| |Results | |(对话历史) | |
| +----+-----+ +----+-----+ +----+-----+ +----+-----+ |
| | | | | |
| v v v v |
| +-----------------------------------------------+ |
| | 优先级排序 + Token 预算控制 | |
| | | |
| | 1. 按优先级分组 (CRITICAL>HIGH>NORMAL>LOW) | |
| | 2. 从高到低分配 Token 预算 | |
| | 3. 超预算的低优先级内容: 截断/摘要/丢弃 | |
| | 4. 组装为最终 messages 列表 | |
| +----------------------+------------------------+ |
| | |
| v |
| +-----------------------------------------------+ |
| | 最终上下文 (<= max_tokens - reserved) | |
| | | |
| | [system] 你是一个助手... (CRITICAL) | |
| | [system] 安全规则: 不讨论... (CRITICAL) | |
| | [user] 请根据以下信息回答... (HIGH) | |
| | [tool] RAG检索: 根据文档... (HIGH) | |
| | [user] 前面提到的方案是什么? (NORMAL) | |
| | [assistant] 方案是... (NORMAL) | |
| | [user] 最新问题 (CRITICAL) | |
| +-----------------------------------------------+ |
| |
+----------------------------------------------------------+11.5 信息压缩策略
11.6 Context Engineering vs Prompt Engineering
Prompt Engineering (提示工程):
关注: "如何写好一条提示词"
范围: 单次交互
目标: 优化单次输出质量
方法: 角色设定、Few-Shot、格式约束
Context Engineering (上下文工程):
关注: "如何管理进入上下文的所有信息"
范围: 多轮交互、多信息源
目标: 在有限窗口内最大化信息价值
方法: 优先级管理、预算分配、压缩策略
关系: Context Engineering 包含 Prompt Engineering
Prompt Engineering 是 Context Engineering 的子集11.7 实际应用模式
模式1: RAG 场景的上下文管理
-> 检索结果按相关性排序，Top-K 注入上下文
-> 低相关性结果被截断
模式2: 长对话场景的上下文管理
-> 近 3 轮对话完整保留
-> 4-10 轮对话摘要压缩
-> 10 轮以上对话丢弃
模式3: Agent 场景的上下文管理
-> System Prompt + 工具定义 (固定)
-> 当前任务目标 (CRITICAL)
-> 工具调用历史 (可压缩)
-> 观察结果 (可截断)11.8 与其他概念的关联
<- 上下文窗口管理：窗口管理是上下文工程的具体实现
<- Tokenization：Token 计数是预算分配的基础
-> Agent 核心能力：Agent 的短期记忆管理即上下文工程
-> RAG：RAG 检索结果是上下文的重要来源
概念关系总览
+--------------------------------------------------+
| LLM 应用开发技术体系 |
+--------------------------------------------------+
基础层 应用层
| |
v v
+-----------+ +---------------+
|Tokenization|--- 计费/预算 ---> 上下文窗口管理 ---------> |Context |
|(应用层理解) | (Token预算/截断/摘要) |Engineering |
+-----+-----+ |(上下文工程) |
| +-------+-------+
v |
+-----------+ +--------------+ |
| Embedding |-------->| 语义搜索 | |
|(语义向量化)| |(向量相似度匹配)| |
+-----------+ +--------------+ |
|
+-----------+ +--------------+ +--------------+ |
|采样参数 |-------->| LLM API 调用 |-------->| Prompt |<-+
|(Temp/Top-k)| |(流式/重试/计费)| | Chaining |
+-----------+ +--------------+ |(链式提示) |
+-------+-------+
+-----------+ +--------------+ |
|System |-------->| Few-Shot 与 |-----------------+
|Prompt 设计 | | 提示模板 |
|(角色/约束) | |(示例/模板管理) |
+-----------+ +--------------+
+-----------+
|Function |--- 工具调用流程 ---> Agent 核心能力
|Calling |--- 参数结构化 -----> 结构化输出
+-----------+概念间的依赖关系
Tokenization -> 上下文窗口管理 -> Context Engineering
^
Embedding -> 语义搜索 ----------------+
^
System Prompt -> Few-Shot -> Prompt Chaining
| |
结构化输出 <---- Function Calling
|
LLM API 调用 (串联所有)
评论区