大模型微调与 Agent 实战：从 LoRA 到多工具调用

{
  "instruction": "请根据以下财报内容，提取营收、净利润、同比增长率。",
  "input": "2024年Q3，公司营收120.5亿元，同比增长18.7%，净利润23.4亿元...",
  "output": "营收120.5亿，同比+18.7%；净利润23.4亿。"
}

import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from datasets import load_dataset
from trl import SFTTrainer

# 1. 量化配置（QLoRA）
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,
)

# 2. 加载基座模型（以 Qwen2-7B 为例）
model_name = "Qwen/Qwen2-7B"
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    quantization_config=bnb_config,
    device_map="auto",
    trust_remote_code=True,
)
model = prepare_model_for_kbit_training(model)

# 3. LoRA 配置
lora_config = LoraConfig(
    r=16,                    # 秩
    lora_alpha=32,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora_config)

# 4. 加载数据
dataset = load_dataset("json", data_files="train.jsonl", split="train")

# 5. 训练（使用 SFTTrainer）
trainer = SFTTrainer(
    model=model,
    tokenizer=tokenizer,
    train_dataset=dataset,
    max_seq_length=2048,
    dataset_text_field="instruction",   # 需配合格式化函数
    args=TrainingArguments(
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,
        num_train_epochs=3,
        learning_rate=2e-4,
        fp16=True,
        logging_steps=10,
        save_steps=100,
        output_dir="./qwen2-7b-lora",
    ),
)
trainer.train()

# 保存 LoRA 权重
model.save_pretrained("./qwen2-7b-lora-final")
tokenizer.save_pretrained("./qwen2-7b-lora-final")

from peft import PeftModel

base_model = AutoModelForCausalLM.from_pretrained(model_name, device_map="auto")
lora_model = PeftModel.from_pretrained(base_model, "./qwen2-7b-lora-final")

prompt = "请提取以下财报关键指标：2024年Q3营收120.5亿，净利润23.4亿，同比增长18.7%。"
inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
output = lora_model.generate(**inputs, max_new_tokens=128)
print(tokenizer.decode(output[0], skip_special_tokens=True))

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, AwqConfig

app = FastAPI()

# 加载 AWQ 量化模型（提前转换）
model_path = "./qwen2-7b-awq"
tokenizer = AutoTokenizer.from_pretrained(model_path)
model = AutoModelForCausalLM.from_pretrained(
    model_path,
    device_map="auto",
    torch_dtype=torch.float16,
)

class Query(BaseModel):
    prompt: str
    max_tokens: int = 256

@app.post("/generate")
def generate(query: Query):
    inputs = tokenizer(query.prompt, return_tensors="pt").to("cuda")
    outputs = model.generate(
        **inputs,
        max_new_tokens=query.max_tokens,
        temperature=0.1,
        do_sample=False,
    )
    result = tokenizer.decode(outputs[0], skip_special_tokens=True)
    # 去除输入重复
    result = result[len(query.prompt):].strip()
    return {"result": result}

# 运行：uvicorn server:app --host 0.0.0.0 --port 8000

tools = {
    "search": {
        "desc": "搜索网络信息，输入关键词",
        "func": lambda q: f"模拟搜索结果：关于'{q}'的最新资料..."
    },
    "calc": {
        "desc": "计算数学表达式，输入表达式",
        "func": lambda exp: str(eval(exp))
    },
    "db_query": {
        "desc": "查询内部财务数据库，输入SQL条件",
        "func": lambda sql: f"模拟查询结果：{sql} 返回3条记录..."
    }
}

def agent_loop(user_question, max_steps=5):
    prompt = f"""
你是一个智能助手，可以调用以下工具：
{chr(10).join([f"- {name}: {info['desc']}" for name, info in tools.items()])}

请按以下格式输出：
思考：我需要做什么？
行动：工具名(参数)
观察：工具返回结果
...（可重复）
最终回答：最终结论

用户问题：{user_question}
"""
    for _ in range(max_steps):
        response = call_llm(prompt)  # 调用上一节的 API
        if "最终回答" in response:
            return response.split("最终回答")[-1].strip()
        # 解析行动
        if "行动：" in response:
            action_line = response.split("行动：")[1].split("\n")[0]
            tool_name = action_line.split("(")[0]
            arg = action_line.split("(")[1].rstrip(")")
            if tool_name in tools:
                obs = tools[tool_name]["func"](arg)
                prompt += f"\n观察：{obs}\n"
            else:
                prompt += "\n观察：工具不存在，请重试\n"
        else:
            prompt += "\n请按照格式输出思考→行动→观察。"
    return "超出步骤上限，请简化问题。"

# 示例
print(agent_loop("查询公司上季度营收，然后计算环比增长"))