LLM核心概念全景解析：从Token到Agent的工程实践指南

1次阅读

共计 2112 个字符，预计需要花费 6 分钟才能阅读完成。

Token – CPU 指令集
如同 CPU 处理的是机器指令，LLM 处理的是 Token 化的文本片段。一个英文单词约 1 - 2 个 Token，中文汉字通常 1 个 Token。理解这个粒度对计算资源分配至关重要。
Context Window – 内存总线带宽
像内存总线限制数据传输量，Context Window 限制模型单次处理的 Token 数量（如 GPT- 4 的 32k）。超出部分会被丢弃，需要像内存交换技术那样设计缓存策略。
MCP（Multi-Turn Conversation Protocol）– 会话状态机
类似 TCP 协议维护连接状态，MCP 管理多轮对话的上下文关联性，确保对话不 ” 断片 ”。

现象：长文档处理时关键信息被截断
根因：未做上下文重要性分级
影响：模型回答出现事实性错误

典型问题：重复指令占用 30%+Token
案例：请用简洁的语言... 同时保持详细...这类矛盾指令

症状：并行工具调用导致状态冲突
示例：同时调用 日历查询 和天气 API时返回结果错位

def compress_context(text: str, max_tokens: int) -> str:
    """
    基于 TF-IDF 的关键信息保留算法
    :param text: 原始文本
    :param max_tokens: 目标 Token 数
    :return: 压缩后的文本
    """
    from sklearn.feature_extraction.text import TfidfVectorizer

    sentences = text.split('.')
    vectorizer = TfidfVectorizer()
    try:
        tfidf = vectorizer.fit_transform(sentences)
        importance = tfidf.sum(axis=1).A1
        ranked = sorted(zip(sentences, importance), key=lambda x: -x[1])

        compressed = []
        current_tokens = 0
        for sent, _ in ranked:
            sent_tokens = len(sent.split())  # 简易 Token 估算
            if current_tokens + sent_tokens <= max_tokens:
                compressed.append(sent)
                current_tokens += sent_tokens
            else:
                break

        return '.'.join(compressed)
    except ValueError as e:
        print(f"压缩失败: {str(e)}")
        return text[:max_tokens*4]  # 降级方案

# 角色
你是一个资深技术顾问，擅长用类比解释复杂概念

# 格式要求
1. 首段给出技术类比
2. 次段说明实际差异
3. 最后提供实践建议

# 安全约束
- 禁止医疗建议
- 金融数据需标注来源
- 政治话题返回[权限不足]

stateDiagram
    [*] --> Idle
    Idle --> Processing: 收到用户输入
    Processing --> ToolCalling: 需要外部工具
    ToolCalling --> Processing: 获取工具结果
    Processing --> Responding: 生成最终回复
    Responding --> Idle

    ToolCalling --> Error: 工具超时
    Error --> Idle: 重置状态

误区：直接用空格分词统计
正解：使用 tiktoken 库精确计算

import tiktoken

def exact_token_count(text: str, model: str = "gpt-4") -> int:
    try:
        enc = tiktoken.encoding_for_model(model)
        return len(enc.encode(text))
    except KeyError:
        # 降级处理
        return len(text) // 4  # 经验估值

from concurrent.futures import ThreadPoolExecutor, TimeoutError

with ThreadPoolExecutor() as executor:
    future = executor.submit(call_weather_api, location)
    try:
        result = future.result(timeout=5.0)
    except TimeoutError:
        result = "API 响应超时"