02 - 大模型应用开发¶

# Transformer核心组件
import torch
import torch.nn as nn

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super().__init__()  # super()调用父类方法
        self.d_model = d_model
        self.num_heads = num_heads
        self.head_dim = d_model // num_heads

        self.q_linear = nn.Linear(d_model, d_model)
        self.k_linear = nn.Linear(d_model, d_model)
        self.v_linear = nn.Linear(d_model, d_model)
        self.out_linear = nn.Linear(d_model, d_model)

    def forward(self, query, key, value, mask=None):
        batch_size = query.size(0)

        # 线性变换并分头
        Q = self.q_linear(query).view(batch_size, -1, self.num_heads, self.head_dim).transpose(1, 2)
        K = self.k_linear(key).view(batch_size, -1, self.num_heads, self.head_dim).transpose(1, 2)
        V = self.v_linear(value).view(batch_size, -1, self.num_heads, self.head_dim).transpose(1, 2)

        # 注意力计算
        scores = torch.matmul(Q, K.transpose(-2, -1)) / torch.sqrt(torch.tensor(self.head_dim))
        if mask is not None:
            scores = scores.masked_fill(mask == 0, -1e9)

        attn = torch.softmax(scores, dim=-1)
        context = torch.matmul(attn, V)

        # 合并头并输出
        context = context.transpose(1, 2).contiguous().view(batch_size, -1, self.d_model)
        return self.out_linear(context)

# 1. 角色设定
system_prompt = """你是一位资深Python开发工程师，擅长代码审查和性能优化。
请对用户的代码进行审查，指出潜在问题并提供改进建议。"""

# 2. Few-Shot示例
few_shot_prompt = """
将自然语言转换为SQL查询：

示例1:
用户: 查询年龄大于25岁的用户
SQL: SELECT * FROM users WHERE age > 25;

示例2:
用户: 统计每个部门的平均工资
SQL: SELECT department, AVG(salary) FROM employees GROUP BY department;

现在请转换:
用户: {user_input}
SQL:
"""

# 3. Chain-of-Thought (CoT)
cot_prompt = """
问题: 一个农场有鸡和兔，头共35个，脚共94只。鸡和兔各有多少只？

让我们一步步思考：
1. 设鸡有x只，兔有y只
2. 根据头的数量：x + y = 35
3. 根据脚的数量：2x + 4y = 94
4. 从方程1得：x = 35 - y
5. 代入方程2：2(35-y) + 4y = 94
6. 70 - 2y + 4y = 94
7. 2y = 24
8. y = 12, x = 23

答案：鸡23只，兔12只
"""

# 1. ReAct模式 (Reasoning + Acting)
react_prompt = """
你可以使用以下工具：
- search(query): 搜索信息
- calculator(expression): 计算表达式
- code_interpreter(code): 执行Python代码

请按照以下格式回答：
思考: 我需要...
行动: tool_name(arguments)
观察: 工具返回的结果
...（重复思考-行动-观察）
最终答案: ...

问题: 2024年诺贝尔物理学奖得主是谁？他们的主要贡献是什么？
"""

# 2. Self-Consistency
self_consistency_prompt = """
请对以下问题给出5个不同的推理路径和答案，然后选择最一致的答案。

问题: {question}

路径1:
推理: ...
答案: ...

路径2:
推理: ...
答案: ...

...（其他路径）

最一致的答案是: ...
"""

# 3. Tree of Thoughts (ToT)
tot_prompt = """
问题: 如何设计一个高并发的电商系统？

请从多个角度思考，并构建思维树：

角度1: 架构层面
- 微服务拆分
- 服务治理
- ...

角度2: 数据库层面
- 分库分表
- 读写分离
- ...

角度3: 缓存层面
- 多级缓存
- 缓存一致性
- ...

综合以上角度，最佳方案是: ...
"""

# RAG完整流程（LCEL方式，替代已废弃的 VectorDBQA/LLMChain）
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

class RAGSystem:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings()
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.vector_store = None

    def load_documents(self, file_paths):
        """加载文档"""
        documents = []
        for path in file_paths:
            if path.endswith('.pdf'):
                loader = PyPDFLoader(path)
            else:
                loader = TextLoader(path)
            documents.extend(loader.load())
        return documents

    def split_documents(self, documents, chunk_size=1000, chunk_overlap=200):
        """文档分块"""
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len
        )
        return text_splitter.split_documents(documents)

    def create_vector_store(self, texts):
        """创建向量数据库"""
        self.vector_store = FAISS.from_documents(texts, self.embeddings)
        return self.vector_store

    def query(self, question, k=3):
        """RAG查询（LCEL方式）"""
        retriever = self.vector_store.as_retriever(search_kwargs={"k": k})

        system_prompt = """基于以下上下文回答问题。如果上下文中没有相关信息，请说"我不知道"。

{context}"""

        prompt = ChatPromptTemplate.from_messages([
            ("system", system_prompt),
            ("human", "{input}")
        ])

        question_answer_chain = create_stuff_documents_chain(self.llm, prompt)
        rag_chain = create_retrieval_chain(retriever, question_answer_chain)

        result = rag_chain.invoke({"input": question})
        return result["answer"], result["context"]

# 使用示例
rag = RAGSystem()
docs = rag.load_documents(['./docs/manual.pdf', './docs/faq.txt'])
texts = rag.split_documents(docs)
rag.create_vector_store(texts)

answer, sources = rag.query("如何重置密码？")
print(answer)

# 1. 混合检索（向量+关键词）
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

class HybridRAG:
    def __init__(self, documents):
        # BM25检索器
        self.bm25_retriever = BM25Retriever.from_documents(documents)
        self.bm25_retriever.k = 2

        # 向量检索器
        self.vector_retriever = FAISS.from_documents(
            documents, OpenAIEmbeddings()
        ).as_retriever(search_kwargs={"k": 2})

        # 混合检索器
        self.ensemble_retriever = EnsembleRetriever(
            retrievers=[self.bm25_retriever, self.vector_retriever],
            weights=[0.5, 0.5]
        )

# 2. 重排序（Reranking）
from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.document_compressors import LLMChainExtractor

class RerankRAG:
    def __init__(self, base_retriever, llm):
        compressor = LLMChainExtractor.from_llm(llm)  # 注意：生产环境推荐用CrossEncoder重排序
        self.compression_retriever = ContextualCompressionRetriever(
            base_compressor=compressor,
            base_retriever=base_retriever
        )

# 3. 多跳检索（Multi-hop）
class MultiHopRAG:
    def query(self, question, max_hops=3):
        current_question = question
        all_contexts = []

        for hop in range(max_hops):
            # 检索相关文档
            docs = self.retriever.get_relevant_documents(current_question)
            context = "\n".join([d.page_content for d in docs])
            all_contexts.append(context)

            # 生成子问题或判断是否需要继续
            sub_question = self.generate_sub_question(question, context)
            if sub_question is None:
                break
            current_question = sub_question

        # 综合所有上下文生成答案
        final_context = "\n\n".join(all_contexts)
        return self.generate_answer(question, final_context)

# LangChain Agent（LCEL方式）
from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI
from langchain_community.utilities import SerpAPIWrapper
from langchain_core.prompts import ChatPromptTemplate

# 定义工具
def calculator(expression: str) -> str:
    """安全的数学计算"""
    import numexpr  # 比 eval 更安全
    return str(numexpr.evaluate(expression))

tools = [
    Tool(
        name="Search",
        func=SerpAPIWrapper().run,
        description="用于搜索最新信息"
    ),
    Tool(
        name="Calculator",
        func=calculator,
        description="用于数学计算，输入数学表达式"
    ),
]

# 创建Agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的AI助手,可以使用工具回答问题。\n\n"
               "可用工具:\n{tools}\n工具名: {tool_names}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行Agent
response = agent_executor.invoke({
    "input": "2024年中国的GDP是多少？增长了多少百分比？"
})

# CrewAI多Agent协作
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

# 定义Agents
researcher = Agent(
    role='研究员',
    goal='收集和分析技术信息',
    backstory='你是一位资深技术研究员，擅长快速学习新技术',
    verbose=True,
    allow_delegation=False,
    llm=ChatOpenAI(model="gpt-4o", temperature=0.7)
)

writer = Agent(
    role='技术作家',
    goal='将技术内容转化为易读的文章',
    backstory='你是一位技术作家，擅长将复杂概念简单化',
    verbose=True,
    allow_delegation=False,
    llm=ChatOpenAI(model="gpt-4o", temperature=0.7)
)

# 定义Tasks
task1 = Task(
    description='研究2024年最新的前端框架趋势，列出3个最重要的趋势',
    agent=researcher,
    expected_output='一份包含3个前端趋势的详细报告'
)

task2 = Task(
    description='基于研究报告，撰写一篇面向开发者的技术博客文章',
    agent=writer,
    expected_output='一篇1000字左右的技术博客文章'
)

# 创建Crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[task1, task2],
    process=Process.sequential,
    verbose=True
)

# 运行
result = crew.kickoff()
print(result)

# 使用PEFT进行LoRA微调
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from trl import SFTTrainer

# 加载基础模型
model_name = "meta-llama/Llama-2-7b-hf"
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    load_in_4bit=True,  # 量化加载
    torch_dtype=torch.float16,
    device_map="auto"
)

# 准备模型
model = prepare_model_for_kbit_training(model)

# LoRA配置
lora_config = LoraConfig(
    r=16,  # LoRA秩
    lora_alpha=32,
    target_modules=["q_proj", "v_proj"],  # 目标模块
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM"
)

# 应用LoRA
model = get_peft_model(model, lora_config)

# 训练参数
training_args = TrainingArguments(
    output_dir="./results",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    optim="paged_adamw_32bit",
    save_steps=100,
    logging_steps=10,
    learning_rate=2e-4,
    weight_decay=0.001,
    fp16=True,
    bf16=False,
    max_grad_norm=0.3,
    max_steps=-1,
    warmup_ratio=0.03,
    group_by_length=True,
    lr_scheduler_type="constant",
)

# 训练
trainer = SFTTrainer(
    model=model,
    train_dataset=dataset,
    peft_config=lora_config,
    dataset_text_field="text",
    max_seq_length=512,
    tokenizer=tokenizer,
    args=training_args,
)

trainer.train()

# 使用vLLM进行高效推理
from vllm import LLM, SamplingParams

# 加载模型
llm = LLM(
    model="meta-llama/Llama-2-7b-hf",
    tensor_parallel_size=1,  # GPU数量
    gpu_memory_utilization=0.9,
    max_num_seqs=256
)

# 采样参数
sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.95,
    max_tokens=512
)

# 批量推理
prompts = [
    "The future of AI is",
    "In the next decade, technology will",
    "The most important skill for developers is"
]

outputs = llm.generate(prompts, sampling_params)

for output in outputs:
    prompt = output.prompt
    generated_text = output.outputs[0].text
    print(f"Prompt: {prompt!r}, Generated text: {generated_text!r}")

# GPTQ量化
from transformers import AutoModelForCausalLM, AutoTokenizer
from auto_gptq import AutoGPTQForCausalLM, BaseQuantizeConfig

# 量化配置
quantize_config = BaseQuantizeConfig(
    bits=4,  # 4-bit量化
    group_size=128,
    desc_act=False
)

# 量化模型
model = AutoGPTQForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b-hf",
    quantize_config=quantize_config
)

# 执行量化
model.quantize(examples)

# 保存量化模型
model.save_quantized("./llama-7b-4bit")

# 智能客服Agent（使用 create_react_agent + AgentExecutor）
from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, AIMessage

class CustomerServiceAgent:
    def __init__(self):
        self.rag_system = RAGSystem()  # RAG系统
        self.order_system = OrderAPI()  # 订单系统API
        self.chat_history = []  # 手动维护对话历史（替代已弃用的 ConversationBufferMemory）

        self.tools = [
            Tool(
                name="ProductFAQ",
                func=self.rag_system.query,
                description="查询产品FAQ和文档"
            ),
            Tool(
                name="OrderQuery",
                func=self.order_system.query_order,
                description="查询订单状态"
            ),
            Tool(
                name="HumanHandoff",
                func=self.human_handoff,
                description="转接到人工客服"
            )
        ]

        llm = ChatOpenAI(model="gpt-4o", temperature=0)
        prompt = ChatPromptTemplate.from_messages([
            ("system", "你是一个专业的智能客服Agent。请使用工具回答用户问题。\n\n"
                       "可用工具:\n{tools}\n工具名: {tool_names}"),
            ("placeholder", "{chat_history}"),
            ("human", "{input}"),
            ("placeholder", "{agent_scratchpad}"),
        ])
        agent = create_react_agent(llm, self.tools, prompt)
        self.agent_executor = AgentExecutor(agent=agent, tools=self.tools, verbose=True)

    def chat(self, user_input):
        response = self.agent_executor.invoke({
            "input": user_input,
            "chat_history": self.chat_history,
        })

        # 保存对话历史
        self.chat_history.append(HumanMessage(content=user_input))
        self.chat_history.append(AIMessage(content=response["output"]))

        return response["output"]