构建AI智能体：智能协作的艺术：大模型上下文与Token优化指南

# 在API调用中明确限制输出长度
api_parameters = {
    "max_tokens": 500,        # 限制最大输出长度
    "temperature": 0.7,       # 控制随机性，避免冗长
    "stop_sequences": ["\n\n", "。"]  # 设置停止序列
}

# 不同场景的推荐设置
output_optimization = {
    "代码生成": {"max_tokens": 1000, "temperature": 0.2},
    "创意写作": {"max_tokens": 800, "temperature": 0.8},
    "数据分析": {"max_tokens": 400, "temperature": 0.3},
    "简单问答": {"max_tokens": 200, "temperature": 0.5}
}

vague_request = "给我一些提高代码质量的建议"
# 模型可能生成冗长的散文式回答

structured_request = """
以要点形式提供5条代码质量改进建议：
1. 
2. 
3. 
4. 
5. 
每条不超过20字。
"""
# 强制简洁、结构化的输出

def monitor_token_usage(api_client):
    """监控Token使用情况"""
    usage_stats = {
        "total_requests": 0,
        "total_input_tokens": 0,
        "total_output_tokens": 0,
        "average_tokens_per_request": 0
    }
    
    def wrapped_call(prompt, **kwargs):
        # 估算输入Tokens
        input_tokens = estimate_tokens(prompt)
        
        # 调用API
        response = api_client.generate(prompt, **kwargs)
        
        # 获取输出Tokens（如果API返回）
        output_tokens = response.usage.completion_tokens if hasattr(response, 'usage') else estimate_tokens(response.text)
        
        # 更新统计
        usage_stats["total_requests"] += 1
        usage_stats["total_input_tokens"] += input_tokens
        usage_stats["total_output_tokens"] += output_tokens
        usage_stats["average_tokens_per_request"] = (
            usage_stats["total_input_tokens"] + usage_stats["total_output_tokens"]
        ) / usage_stats["total_requests"]
        
        print(f"本次调用: 输入{input_tokens} + 输出{output_tokens} = 总计{input_tokens + output_tokens} tokens")
        print(f"累计统计: {usage_stats}")
        
        return response
    
    return wrapped_call

# 使用装饰器监控
monitored_api = monitor_token_usage(api_client)

from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
from modelscope import snapshot_download

# 加载模型和分词器
cache_dir = "D:\\modelscope\\hub"
model_name = "qwen/Qwen1.5-1.8B-Chat"
local_model_path = snapshot_download(model_name, cache_dir=cache_dir)
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.float16,
    device_map="auto",
    trust_remote_code=True
)

# 设置长上下文配置
model.generation_config.max_new_tokens = 512  # 生成的最大新token数
model.generation_config.do_sample = True

# 构建一个长文档（约2000字，模拟长上下文）
long_document = """
第一章：人工智能发展简史

人工智能的概念最早可以追溯到1950年代。图灵在他的经典论文《计算机器与智能》中提出了著名的"图灵测试"，为AI研究奠定了基础。1956年，在达特茅斯会议上，约翰·麦卡锡首次提出了"人工智能"这一术语。

早期AI研究主要集中在符号推理和专家系统上。1970-1980年代，专家系统在医疗诊断、化学分析等领域取得了成功应用。然而，由于计算能力的限制和知识的获取瓶颈，AI经历了两次"AI寒冬"。

第二章：深度学习革命

2012年，AlexNet在ImageNet竞赛中的突破性表现标志着深度学习时代的到来。基于神经网络的方法在计算机视觉、自然语言处理等领域取得了显著进展。

Transformer架构的提出是NLP领域的里程碑。2017年，Vaswani等人发表的《Attention Is All You Need》引入了自注意力机制，为后来的一系列大语言模型奠定了基础。

第三章：大语言模型时代

GPT系列模型展示了缩放定律的有效性。从GPT-1到GPT-3，模型参数从1.17亿增加到1750亿，性能也实现了质的飞跃。

2023年以来，开源模型如LLaMA、Qwen等迅速发展，使得大模型技术更加普及。这些模型在代码生成、创意写作、逻辑推理等方面展现出强大能力。

第四章：技术挑战与未来方向

当前大模型面临的主要挑战包括：幻觉问题、推理能力有限、长上下文处理、多模态理解等。研究人员正在从多个角度寻求突破。

未来发展方向可能包括：更强的推理能力、世界模型构建、具身智能、以及更高效的自监督学习方法。量子计算也可能为AI带来新的突破。

重要概念定义：
-  transformer: 基于自注意力机制的神经网络架构
-  位置编码：帮助模型理解序列中元素的位置信息
-  上下文窗口：模型一次性能处理的文本长度
-  缩放定律：模型性能随参数和数据规模增长的规律
"""

def test_long_context_qa(document, question):
    # 构建对话格式
    messages = [
        {
            "role": "system",
            "content": "你是一个专业的技术助手，需要基于提供的长文档内容准确回答问题。"
        },
        {
            "role": "user", 
            "content": f"请仔细阅读以下文档，然后回答问题：\n\n{document}\n\n问题：{question}"
        }
    ]
    
    # 编码输入
    text = tokenizer.apply_chat_template(
        messages, 
        tokenize=False, 
        add_generation_prompt=True
    )
    
    # 计算输入长度
    input_ids = tokenizer.encode(text)
    print(f"输入文本长度: {len(input_ids)} tokens")
    
    # 生成回答
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=300,
            do_sample=True,
            temperature=0.7,
            top_p=0.9,
            pad_token_id=tokenizer.eos_token_id
        )
    
    response = tokenizer.decode(outputs[0][len(inputs.input_ids[0]):], skip_special_tokens=True)
    return response

# 测试问题1：需要整合多个章节信息
question1 = "请总结AI发展历程中的三个主要阶段及其特点"
answer1 = test_long_context_qa(long_document, question1)
print(f"问题1: {question1}")
print(f"回答1: {answer1}\n")

# 测试问题2：需要理解文档末尾的定义
question2 = "根据文档定义，请解释什么是位置编码和上下文窗口"
answer2 = test_long_context_qa(long_document, question2)
print(f"问题2: {question2}")
print(f"回答2: {answer2}\n")

# 测试问题3：需要长距离依赖理解
question3 = "Transformer架构是在哪个阶段提出的？它对后续发展产生了什么影响？"
answer3 = test_long_context_qa(long_document, question3)
print(f"问题3: {question3}")
print(f"回答3: {answer3}")

# 检查模型的RoPE配置
print(f"模型架构: {model.config.model_type}")
print(f"最大位置嵌入: {getattr(model.config, 'max_position_embeddings', 'Not specified')}")
print(f"RoPE theta: {getattr(model.config, 'rope_theta', 10000.0)}")

def analyze_attention_patterns(text):
    inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=2048)
    
    with torch.no_grad():
        outputs = model(**inputs, output_attentions=True)
    
    # 分析注意力权重
    attentions = outputs.attentions
    print(f"注意力层数: {len(attentions)}")
    print(f"每层注意力形状: {attentions[0].shape}")
    
    # 最后一层的平均注意力距离
    last_layer_attention = attentions[-1][0]  # [head, seq_len, seq_len]
    return last_layer_attention

# 1. 分块处理策略（对于超长文本）
def process_long_document_in_chunks(document, chunk_size=1500, overlap=200):
    chunks = []
    tokens = tokenizer.encode(document)
    
    for i in range(0, len(tokens), chunk_size - overlap):
        chunk_tokens = tokens[i:i + chunk_size]
        chunk_text = tokenizer.decode(chunk_tokens)
        chunks.append(chunk_text)
    
    return chunks

# 2. 关键信息提取
def extract_key_information(chunks, question):
    # 对每个分块进行相关性评分
    relevant_chunks = []
    for chunk in chunks:
        # 简单的关键词匹配评分
        score = sum(1 for word in question.split() if word.lower() in chunk.lower())
        if score > 0:
            relevant_chunks.append((score, chunk))
    
    # 按相关性排序并返回最相关的几个分块
    relevant_chunks.sort(reverse=True)
    return [chunk for _, chunk in relevant_chunks[:3]]

# 3. 流式处理（减少内存压力）
def stream_generate_response(prompt):
    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
    
    for output in model.generate(
        **inputs,
        max_new_tokens=300,
        do_sample=True,
        temperature=0.7,
        streamer=None,  # 可以配置流式输出
        early_stopping=True
    ):