大语言模型1 基本用法
相关工具和库: langchain, huggingface, ollama, llama-index
模型: chatgpt, deepseek-chat, qwen, embedding bge-m3
调用大语言模型
注意 api key 不能放到 github 上,可以用环境变量传入。
macos 上本地模型有针对 apple m 芯片优化。
使用命令行
ollama 可以给予命令后使用
openai api 可以用curl调用
命令行调用 ollama
ollama run qwen2.5:3b "hello"
echo "hello" | ollama run qwen2.5:3b
命令使用 curl 也可以,这样方便切到 chatgpt
curl https://api.openai.com/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer 你的OPENAI_API_KEY" \
-d '{
"model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": "你好,介绍一下自己"}]
}'
使用 OpenAI API
调用 ollama
from openai import OpenAI
# 初始化客户端
client = OpenAI(
base_url="http://localhost:11434/v1", # 用的默认端口号
api_key="ollama" # API key 可以是任意值
)
# 调用模型
response = client.chat.completions.create(
model="qwen2.5:3b",
messages=[
{"role": "system", "content": "你是一个有帮助的助手"},
{"role": "user", "content": "请解释什么是大语言模型"}
]
)
print(response.choices[0].message.content)
调用 deepseek api
.env 文件
DEEPSEEK_API_KEY=...
#!pip3 install openai
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(
api_key=os.environ['DEEPSEEK_API_KEY'],
base_url="https://api.deepseek.com")
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "Hello"},
],
stream=False
)
print(response.choices[0].message.content)
简单扩展
多轮对话
多轮对话需要手动保持对话历史。
流式返回
调用工具
常用的工具,比如 联网搜索,计算器,维基百科。
- 计算器可以用 numexpr 或者 sympy (支持符号计算)
使用 OpenAI API
import openai
import json
import re
import os
from typing import Dict, Any
from dotenv import load_dotenv
load_dotenv()
# -------------------------- 配置部分 --------------------------
# 替换为你的 DeepSeek API Key
DEEPSEEK_API_KEY = os.environ['DEEPSEEK_API_KEY']
# DeepSeek API 基础地址
DEEPSEEK_BASE_URL = "https://api.deepseek.com/v1"
# 要使用的模型名称
MODEL_NAME = "deepseek-chat"
# -------------------------- 初始化客户端 --------------------------
client = openai.OpenAI(
api_key=DEEPSEEK_API_KEY,
base_url=DEEPSEEK_BASE_URL
)
# -------------------------- 工具定义 --------------------------
# 定义计算器工具
def calculator_tool(expression: str) -> str:
try:
if not re.match(r'^[\d\+\-\*\/\(\)\.\s]+$', expression):
return f"无效的表达式:仅支持数字和 +-*/() 运算符,输入内容:{expression}"
result = eval(expression, {"__builtins__": None}, {})
return f"计算结果:{expression} = {result}"
except Exception as e:
return f"计算失败:{str(e)},表达式:{expression}"
# 定义工具列表(可扩展更多工具)
tools = [
{
"type": "function",
"function": {
"name": "calculator_tool",
"description": "用于执行数学计算的工具,支持加减乘除和括号运算",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "需要计算的数学表达式,例如 '100+200*3' 或 '(50-20)/5'"
}
},
"required": ["expression"],
"additionalProperties": False
}
}
}
]
# 工具映射:将工具名称映射到实际函数
tool_functions = {
"calculator_tool": calculator_tool
}
# -------------------------- 核心对话函数 --------------------------
def chat_with_tools(user_message: str) -> str:
"""
与 DeepSeek 模型对话,并自动调用工具
:param user_message: 用户输入
:return: 最终回复
"""
# 1. 发送用户消息,请求模型判断是否需要调用工具
response = client.chat.completions.create(
model=MODEL_NAME,
messages=[{"role": "user", "content": user_message}],
tools=tools,
tool_choice="auto", # 让模型自动决定是否调用工具
temperature=0.1 # 降低随机性,保证工具调用的准确性
)
assistant_message = response.choices[0].message
final_response = ""
# 2. 检查模型是否要求调用工具
if assistant_message.tool_calls:
# 收集工具调用结果
tool_results = []
# 遍历所有工具调用请求
for tool_call in assistant_message.tool_calls:
tool_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)
# 执行工具调用
if tool_name in tool_functions:
tool_result = tool_functions[tool_name](**tool_args)
tool_results.append({
"tool_call_id": tool_call.id,
"role": "tool",
"name": tool_name,
"content": tool_result
})
else:
tool_results.append({
"tool_call_id": tool_call.id,
"role": "tool",
"name": tool_name,
"content": f"未知工具:{tool_name}"
})
# 3. 将工具调用结果返回给模型,生成最终回复
messages = [
{"role": "user", "content": user_message},
assistant_message, # 模型的工具调用请求
# 添加工具调用结果
*[
{
"role": "tool",
"tool_call_id": res["tool_call_id"],
"name": res["name"],
"content": res["content"]
}
for res in tool_results
]
]
# 重新调用模型,结合工具结果生成回复
final_response = client.chat.completions.create(
model=MODEL_NAME,
messages=messages,
temperature=0.7
).choices[0].message.content
else:
# 不需要调用工具,直接返回模型回复
final_response = assistant_message.content
return final_response
# -------------------------- 测试示例 --------------------------
if __name__ == "__main__":
# 测试用例1:需要调用计算器
print("测试1(需要计算):")
user_input1 = "请帮我计算 (100 - 25) * 8 + 150 的结果"
response1 = chat_with_tools(user_input1)
print(f"用户:{user_input1}")
print(f"模型:{response1}\n")
# 测试用例2:不需要调用工具
print("测试2(普通对话):")
user_input2 = "请介绍一下人工智能的发展历程"
response2 = chat_with_tools(user_input2)
print(f"用户:{user_input2}")
print(f"模型:{response2}\n")
# 测试用例3:复杂计算
print("测试3(复杂计算):")
user_input3 = "我有500元,每天花15元,每周额外花50元,请问能坚持多少天?(计算到剩余金额不足一天花费为止)"
response3 = chat_with_tools(user_input3)
print(f"用户:{user_input3}")
print(f"模型:{response3}")
LangChain
使用 langchain
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
# 加载环境变量
load_dotenv()
# 初始化模型
llm = ChatOpenAI(
model="qwen2.5:3b",
base_url="http://localhost:11434/v1",
api_key=""
)
# 定义提示词模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有帮助的助手"),
("user", "{input}")
])
# 构建链
chain = prompt | llm
# 测试
if __name__ == "__main__":
# 测试普通对话
response = chain.invoke({"input": "请介绍一下DeepSeek模型"})
print(f"结果: {response.content}")
调用工具
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
model="gpt-5",
temperature=0.1,
max_tokens=1000,
timeout=3
)
agent = create_agent(model, tools=tools)
result = agent.invoke( {"messages": [HumanMessage("Analyze the major themes in 'Pride and Prejudice'.")]} )
TODO:这个代码有问题 create_tool_calling_agent, create_react_agent
比如可以支持联网搜索。
from langchain_core.tools import tool # 导入tool装饰器
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub
import sympy
# --------------------------
# 1. 使用@tool装饰器定义工具(注解方式)
# --------------------------
@tool
def calculate_expression(expression: str) -> str:
"""
高级计算工具,支持解析字符串形式的数学表达式(如"128*45"、"(5+8)*2")
Args:
expression: 数学表达式字符串,支持加减乘除、括号、幂运算等
Returns:
计算结果的字符串描述
"""
try:
clean_expr = expression.replace("×", "*").replace("÷", "/").replace(" ", "")
result = sympy.sympify(clean_expr)
numeric_result = float(result.evalf())
if numeric_result.is_integer():
numeric_result = int(numeric_result)
return f"表达式 '{expression}' 的计算结果是:{numeric_result}"
except ZeroDivisionError:
return f"错误:表达式 '{expression}' 中包含除以0的运算"
except Exception as e:
return f"计算出错:无法解析表达式 '{expression}',错误信息:{str(e)}"
@tool
def get_weather(city: str) -> str:
"""
模拟获取天气的工具(实际场景可替换为真实API调用)
Args:
city: 城市名称
Returns:
该城市的模拟天气信息
"""
weather_data = {
"北京": "晴,温度 15-25℃,微风",
"上海": "多云,温度 18-28℃,东风3级",
"广州": "雷阵雨,温度 22-30℃,南风2级",
"深圳": "阴,温度 23-29℃,北风1级"
}
return weather_data.get(city.strip(), f"暂无{city}的天气数据")
# --------------------------
# 2. 直接使用装饰后的函数作为tools列表
# --------------------------
# 装饰后的函数本质上是Tool对象,可直接放入列表
tools = [calculate_expression, get_weather]
# --------------------------
# 3. 后续逻辑与之前一致(省略重复部分)
# --------------------------
llm = ChatOpenAI(
model="qwen2.5:3b",
temperature=0.1,
openai_api_base="http://localhost:11434/v1",
openai_api_key="ollama",
max_tokens=2048
)
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(llm=llm, tools=tools, prompt=prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
handle_parsing_errors=True,
max_iterations=5
)
# 测试调用
if __name__ == "__main__":
print("===== 测试注解式Tool:天气查询 =====")
result1 = agent_executor.invoke({"input": "请问北京今天的天气怎么样?"})
print("最终回答:", result1["output"])
print("\n===== 测试注解式Tool:复杂计算 =====")
result2 = agent_executor.invoke({"input": "计算(80+20)*3-150/3的结果是多少?"})
print("最终回答:", result2["output"])
微调 Qwen
能用提示词的就不要去微调。很多时候差异并不明显。通常 Prompt > RAG > 高效微调(如LoRA)> 全参数微调。
下面是在 Colab 上微调,不同环境可以需要改动代码。
构建微调数据集,问答对。
LoRA 低秩
量化需要 cuda
选一个较小的模型演示 “Qwen/Qwen2.5-0.5B-Instruct”
huggingface 有封装好的么?
TODO:这个代码有问题
依赖
pip3 install -q transformers peft accelerate datasets bitsandbytes trl sentencepiece protobuf
pip3 install -q torch torchvision torchaudio
代码
# ======================== 1. 环境准备 ========================
!pip install -U transformers accelerate peft trl datasets bitsandbytes torch sentencepiece
import torch
import warnings
warnings.filterwarnings("ignore")
# 检查 GPU
!nvidia-smi
print(f"GPU 可用: {torch.cuda.is_available()}")
print(f"GPU 名称: {torch.cuda.get_device_name(0) if torch.cuda.is_available() else '无'}")
# ======================== 2. 加载模型与Tokenizer (4bit量化) ========================
from transformers import (
AutoModelForCausalLM,
AutoTokenizer,
BitsAndBytesConfig,
TrainingArguments,
Trainer,
DataCollatorForLanguageModeling
)
from peft import LoraConfig, get_peft_model, PeftModel
from datasets import Dataset
import numpy as np
# 4bit量化配置(核心:省显存,T4必开)
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_use_double_quant=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_compute_dtype=torch.bfloat16
)
# 加载模型和Tokenizer
model_name = "Qwen/Qwen2.5-0.5B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(
model_name,
trust_remote_code=True,
padding_side="right" # 推理时避免警告
)
# Qwen系列需要手动设置pad_token
tokenizer.pad_token = tokenizer.eos_token
tokenizer.pad_token_id = tokenizer.eos_token_id
tokenizer.padding = "max_length" # 强制padding到固定长度
# 加载量化后的模型
model = AutoModelForCausalLM.from_pretrained(
model_name,
quantization_config=bnb_config,
device_map="auto",
trust_remote_code=True,
torch_dtype=torch.bfloat16
)
# 训练时关闭缓存和并行配置
model.config.use_cache = False
model.config.pretraining_tp = 1
# ======================== 3. 准备微调数据 (修复padding/truncation) ========================
# 示例数据:Python编程助手指令集(可替换为任意符合格式的数据集)
train_data = [
{
"system": "你是一个专业的Python编程助手,简洁、准确地回答代码问题,只输出代码和必要注释。",
"instruction": "写一个Python函数,计算列表中所有偶数的和。",
"output": "def sum_even_numbers(lst):\n \"\"\"计算列表中偶数的和\"\"\"\n return sum(num for num in lst if num % 2 == 0)"
},
{
"system": "你是一个专业的Python编程助手,简洁、准确地回答代码问题,只输出代码和必要注释。",
"instruction": "用Python读取CSV文件并打印前5行。",
"output": "import pandas as pd\n\ndef read_csv_sample(file_path):\n \"\"\"读取CSV文件并打印前5行\"\"\"\n df = pd.read_csv(file_path)\n print(df.head())\n\n# 调用示例\n# read_csv_sample('data.csv')"
},
{
"system": "你是一个专业的Python编程助手,简洁、准确地回答代码问题,只输出代码和必要注释。",
"instruction": "写一个Python函数,判断一个数是否为质数。",
"output": "def is_prime(n):\n \"\"\"判断一个数是否为质数\"\"\"\n if n <= 1:\n return False\n for i in range(2, int(n**0.5) + 1):\n if n % i == 0:\n return False\n return True"
},
{
"system": "你是一个专业的Python编程助手,简洁、准确地回答代码问题,只输出代码和必要注释。",
"instruction": "用Python实现冒泡排序。",
"output": "def bubble_sort(lst):\n \"\"\"冒泡排序实现\"\"\"\n arr = lst.copy()\n n = len(arr)\n for i in range(n):\n for j in range(0, n-i-1):\n if arr[j] > arr[j+1]:\n arr[j], arr[j+1] = arr[j+1], arr[j]\n return arr"
},
{
"system": "你是一个专业的Python编程助手,简洁、准确地回答代码问题,只输出代码和必要注释。",
"instruction": "Python如何遍历字典的键值对?",
"output": "# 方法1:items()(推荐)\nmy_dict = {'a': 1, 'b': 2, 'c': 3}\nfor key, value in my_dict.items():\n print(f\"键:{key},值:{value}\")\n\n# 方法2:遍历键再取值\nfor key in my_dict:\n print(f\"键:{key},值:{my_dict[key]}\")"
}
]
# 第一步:格式化数据为Qwen官方对话模板
def format_dataset(example):
"""将单条数据转为Qwen的chat模板格式"""
messages = [
{"role": "system", "content": example["system"]},
{"role": "user", "content": example["instruction"]},
{"role": "assistant", "content": example["output"]}
]
example["text"] = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=False,
padding=False
)
return example
# 第二步:手动Tokenize数据(🔥 核心修复:统一长度)
def tokenize_function(examples):
"""手动编码文本,强制统一长度"""
# 🔥 关键修复:开启padding和truncation,统一到512长度
tokenized = tokenizer(
examples["text"],
truncation=True, # 截断超长文本
max_length=512, # 固定最大长度
padding="max_length", # 不足补pad_token
return_overflowing_tokens=False,
return_tensors=None # 返回list而非tensor,避免批次问题
)
# 语言模型训练需要labels(与input_ids一致)
tokenized["labels"] = tokenized["input_ids"].copy()
# 🔥 重要:将labels中pad_token的位置设为-100(训练时忽略)
tokenized["labels"] = [
[-100 if token == tokenizer.pad_token_id else token for token in label]
for label in tokenized["labels"]
]
return tokenized
# 构建Dataset并处理
dataset = Dataset.from_list(train_data)
dataset = dataset.map(format_dataset) # 格式化文本
dataset = dataset.map(
tokenize_function,
batched=True, # 批量处理更快
remove_columns=dataset.column_names, # 移除原始列
desc="Tokenizing dataset"
)
# 转换为数组(避免list嵌套问题)
dataset.set_format(type="torch", columns=["input_ids", "attention_mask", "labels"])
# 查看处理后的数据样例
print("=== 处理后的数据样例 ===")
print(f"input_ids形状: {dataset[0]['input_ids'].shape}")
print(f"labels形状: {dataset[0]['labels'].shape}")
print(f"padding位置label值: {dataset[0]['labels'][-10:]}") # 验证pad位置是否为-100
# ======================== 4. 配置LoRA (低秩适配) ========================
lora_config = LoraConfig(
r=64, # 秩:0.5B模型推荐64,小数据集可降为32
lora_alpha=16, # 缩放因子,通常为r的1/4
target_modules=["q_proj", "v_proj"], # Qwen核心可训练层
lora_dropout=0.05, # dropout防止过拟合
bias="none", # 不训练bias参数
task_type="CAUSAL_LM" # 因果语言模型任务
)
# 包装模型为LoRA模型
model = get_peft_model(model, lora_config)
# 打印可训练参数占比(0.5B模型约0.1%,极省显存)
print("\n=== 可训练参数 ===")
model.print_trainable_parameters()
# 🔥 修复数据整理器:禁用自动padding(已手动处理)
data_collator = DataCollatorForLanguageModeling(
tokenizer=tokenizer,
mlm=False, # 因果语言模型用False
pad_to_multiple_of=None, # 禁用自动padding
)
# ======================== 5. 配置训练参数并启动训练 ========================
training_args = TrainingArguments(
output_dir="./qwen2.5-0.5b-finetune", # 训练结果保存路径
per_device_train_batch_size=2, # 单卡batch size(T4最大4)
gradient_accumulation_steps=4, # 梯度累积,等效batch=8
learning_rate=2e-4, # 学习率(LoRA常用2e-4)
num_train_epochs=3, # 训练轮数(小数据3-5足够)
logging_steps=1, # 每1步打印一次日志
save_strategy="epoch", # 每轮保存一次模型
fp16=True, # 混合精度训练(T4支持)
optim="paged_adamw_8bit", # 8bit优化器(省显存)
report_to="none", # 关闭wandb日志
gradient_checkpointing=True, # 梯度检查点(进一步省显存)
disable_tqdm=False, # 显示进度条
# 🔥 新增:禁用自动发现长度
auto_find_batch_size=False,
dataloader_pin_memory=False,
)
# 初始化Trainer
trainer = Trainer(
model=model,
args=training_args,
train_dataset=dataset,
data_collator=data_collator, # 处理batch的padding
)
# 开始训练
print("\n=== 开始训练 ===")
trainer.train()
# 保存最终LoRA模型(仅几十MB)
trainer.model.save_pretrained("./qwen2.5-0.5b-lora-final")
tokenizer.save_pretrained("./qwen2.5-0.5b-lora-final")
print("\n=== 训练完成,模型已保存 ===")
# ======================== 6. 推理测试(加载微调后的模型) ========================
def generate_response(prompt):
"""
生成回答函数
:param prompt: 用户提问
:param system_prompt: 系统提示词
:return: 模型回答
"""
# 构造对话模板
messages = [
{"role": "system", "content": "你是一个专业的Python编程助手,简洁、准确地回答代码问题。"},
{"role": "user", "content": prompt}
]
# 编码输入
inputs = tokenizer.apply_chat_template(
messages,
tokenize=True,
add_generation_prompt=True,
return_tensors="pt"
).to("cuda")
# 生成回答
outputs = model.generate(
**inputs,
max_new_tokens=256, # 最大生成长度
temperature=0.7, # 随机性(0-1,越小越确定)
top_p=0.9, # 核采样
do_sample=True, # 采样生成
num_return_sequences=1,
pad_token_id=tokenizer.pad_token_id,
eos_token_id=tokenizer.eos_token_id
)
# 解码并返回结果
response = tokenizer.decode(outputs[0], skip_special_tokens=True)
# 提取助手回答部分
assistant_response = response.split("assistant\n")[-1].strip()
return assistant_response
# 测试案例
print("\n=== 推理测试 ===")
test_prompts = [
"写一个Python函数,计算列表中所有奇数的和。",
"如何用Python实现快速排序?"
]
for prompt in test_prompts:
print(f"\n用户提问:{prompt}")
response = generate_response(prompt)
print(f"模型回答:\n{response}")
提示词工程
提示词工程 https://www.promptingguide.ai/
平时能用提示词的就不要去微调。Prompt → RAG → 高效微调(如LoRA)→ 全参数微调
生成 SQL
使用工具
提示词工程
- 思维链 CoT 可以要求按步骤写出过程,也可以给一个思维链的示例,这样可以按照某种特定的方式引导。
- 示例 Example 给几个示例。
拆解任务
RAG
优化提示词的工具,通过在任务上评估。
Agent
记忆,通过总结来实现。记忆,压缩状态。
ReAct
协议
MCP Skill.md
调用工具 -, 调用自定义工具(天气、计算器、数据库、API…) -, 联网搜索实时信息 -, 自动判断什么时候用工具 -, 多轮工具调用
- 工具: - 文件读写 - 联网搜索 - Excel/CSV/JSON 解析与写入, - 天气 - 计算器 - 数据库 - 执行 Python 代码 -, 文件读写(txt/md), Excel / CSV / JSON 全解析, 联网搜索, 文件夹遍历, 运行 Python 代码(安全沙盒), 读取图片/OCR, 系统信息查询, 命令行多轮对话- 运行Python代码 - 数据库查询 - 文件夹遍历、 - 图片读取
- 可以实现以下任务: - 读取 Excel: 帮我读一下 data.xlsx 里面的内容 - 读取 CSV: 帮我分析 data.csv - 生成表格: 帮我创建一个学生名单.csv,包含姓名、年龄、班级 - 生成 JSON: 把用户信息保存到 user.json - 搜索 + 保存: 搜索2026手机销量,保存到 report.csv - 读取后总结: 帮我读 data.json 并总结 - 帮我列出当前文件夹所有文件 - 读一下 test.xlsx 并分析 - 把 2026 年手机销量数据存成 data.csv - 识别这张图片 img.png 的文字 - 写一段 Python 代码计算 1 到 100 的和 - 查我的电脑 CPU 占用 - 搜索 2026 年 AI 趋势并保存为 report.md - 读取 info.json 并帮我总结
加载 Skill.md
def load_skill_md(skill_path="Skill.md"):
"""加载并解析 Skill.md 技能手册"""
try:
with open(skill_path, "r", encoding="utf-8") as f:
content = f.read()
# 提取元数据与技能描述
meta_match = re.search(r'---\n(.*?)\n---', content, re.DOTALL)
meta = {}
if meta_match:
meta_lines = meta_match.group(1).splitlines()
for line in meta_lines:
if ":" in line:
k, v = line.split(":", 1)
meta[k.strip()] = v.strip()
skill_body = content.replace(meta_match.group(0), "") if meta_match else content
return f"""
【已加载技能:{meta.get('name', '默认技能')}】
描述:{meta.get('description', '无')}
---
{skill_body}
"""
except:
return "未加载 Skill.md 或文件不存在"
相关论文
ReAct(Reason + Act,2022,ICLR) 通过提示词实现,可以参考 langchain 中的实现。
DeepSeek
- DeepMath
- DeepSeek R1 使用合成数据
- DeepProver 使用