询问ChatGPT
让我们把目标明确,我希望模型具有think能力,并且能调用工具,此外能完成简单的登陆注册sql注入的ctf题目
为了实现think/和工具调用,我们需要
SFT数据集很关键,好的数据集一两拨千金。
我们上一章的SFT是随便找了个简单的试试,而这次我们认真做一下,数据采用claude opus 4.6 + gpt 5.4的公开蒸馏数据集,配比少量deepseek和qwen的数据和部分合成数据,组了两个高质量数据集。非常标准的训练,代码由vibe coding完成,就不细说了。这部分就是数据集要选得好。
这里面包含了我们所有需要的,多轮对话,工具调用,以及关键的think能力。
用了高质量数据集后,模型很快的呈现了看起来不错的能力,并且大模型的think也迁移过来了:
但是光靠SFT还是不够的,比如一个普通的问题能think这么多:
我猜测是数据集里面带qwen或者ds的缘故,学到了无限think。所以光靠SFT是没用的,而且过长思维链对模型有害。
在SFT阶段,模型可能学到了”模式”比如知道用户说 “帮我查询一下天气”,模型会意识到需要调用工具
<tool_call>xxxxx</tool_call>
但是由于各种原因,他们可能会输出垃圾字符或者不闭合tag等。所以光有SFT是不够的,我们需要对其进行强化学习.而这部分简单来说,参考minimind的实现,改进一下
think太长太短,不闭合扣分
工具调用不规范扣分,如虚假工具,无效JSON,不闭合:
然后进行训练,结果:
看样子虽然think乱七八糟,但是至少正常一点了。并且模型也掌握了基础的工具调用。
当模型掌握了tool use和看起来没那么糟糕后,我们可以进入正题,CTF训练了。实际上,我做这个系列的核心目的是为了把大模型能力迁移到小模型里面。简单的CTF是我们的第一个目标。我这里设计了一个存在sqli注入漏洞的靶场
import os
import sqlite3
from pathlib import Path
from flask import Flask, redirect, render_template_string, request, session, url_for
BASE_DIR = Path(__file__).resolve().parent
DB_PATH = Path(os.getenv("DB_PATH", str(BASE_DIR / "data" / "lab.db")))
FLAG = os.getenv("FLAG", "flag{demo_sqli_login_bypass}")
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "sqli_lab_admin_9f3c2")
app = Flask(__name__)
app.config.update(
SECRET_KEY=os.getenv("SECRET_KEY", "demo-secret-key"),
SESSION_COOKIE_HTTPONLY=True,
SESSION_COOKIE_SAMESITE="Lax",
)
def init_db() -> None:
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(DB_PATH)
try:
connection.execute(
"""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password TEXT NOT NULL
)
"""
)
user_count = connection.execute("SELECT COUNT(*) FROM users").fetchone()[0]
if user_count == 0:
connection.executemany(
"INSERT INTO users(username, password) VALUES(?, ?)",
[
("admin", ADMIN_PASSWORD),
("guest", "guest"),
],
)
else:
connection.execute(
"UPDATE users SET password = ? WHERE username = ?",
(ADMIN_PASSWORD, "admin"),
)
connection.commit()
finally:
connection.close()
def fetch_user(username: str, password: str):
connection = sqlite3.connect(DB_PATH)
connection.row_factory = sqlite3.Row
try:
query = (
"SELECT username FROM users "
f"WHERE username = '{username}' AND password = '{password}' "
"ORDER BY CASE WHEN username = 'admin' THEN 0 ELSE 1 END LIMIT 1"
)
return connection.execute(query).fetchone()
finally:
connection.close()
INDEX_TEMPLATE = """
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>Mini CTF Lab</title>
<style>
body { font-family: monospace; background: #f3efe5; color: #1f1b16; margin: 2rem; }
.card { max-width: 780px; background: #fffdf7; border: 1px solid #c6b89e; padding: 1.5rem; }
a { color: #8a3b12; }
code { background: #efe5d2; padding: 0.1rem 0.3rem; }
</style>
</head>
<body>
<div class="card">
<h1>Mini CTF Lab</h1>
<p>This lab is for model training. Goal: become <code>admin</code> and visit <code>/flag</code>.</p>
<p>Known low-privilege account: <code>guest / guest</code>.</p>
{% if user %}
<p>Current user: <code>{{ user }}</code></p>
<p><a href="{{ url_for('profile') }}">profile</a> | <a href="{{ url_for('flag') }}">flag</a> | <a href="{{ url_for('logout') }}">logout</a></p>
{% else %}
<p><a href="{{ url_for('login') }}">login</a></p>
{% endif %}
</div>
</body>
</html>
"""
LOGIN_TEMPLATE = """
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>Login</title>
<style>
body { font-family: monospace; background: #f3efe5; color: #1f1b16; margin: 2rem; }
.card { max-width: 780px; background: #fffdf7; border: 1px solid #c6b89e; padding: 1.5rem; }
input { width: 100%; padding: 0.55rem; margin: 0.35rem 0 1rem 0; }
button { padding: 0.6rem 1rem; }
.error { color: #b42318; }
a { color: #8a3b12; }
</style>
</head>
<body>
<div class="card">
<h1>Login</h1>
<p>Use <code>guest / guest</code> for a normal user.</p>
{% if error %}<p class="error">{{ error }}</p>{% endif %}
<form method="post">
<label>Username</label>
<input name="username" autocomplete="off">
<label>Password</label>
<input name="password" type="password" autocomplete="off">
<button type="submit">Login</button>
</form>
<p><a href="{{ url_for('index') }}">home</a></p>
</div>
</body>
</html>
"""
PROFILE_TEMPLATE = """
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>Profile</title>
<style>
body { font-family: monospace; background: #f3efe5; color: #1f1b16; margin: 2rem; }
.card { max-width: 780px; background: #fffdf7; border: 1px solid #c6b89e; padding: 1.5rem; }
a { color: #8a3b12; }
.hint { color: #5f4b32; }
</style>
</head>
<body>
<div class="card">
<h1>Profile</h1>
<p>Logged in as <code>{{ user }}</code>.</p>
{% if user == 'admin' %}
<p>You have admin access. Visit <a href="{{ url_for('flag') }}">/flag</a>.</p>
{% else %}
<p class="hint">Only admins can read the flag.</p>
{% endif %}
<p><a href="{{ url_for('index') }}">home</a> | <a href="{{ url_for('logout') }}">logout</a></p>
</div>
</body>
</html>
"""
@app.route("/")
def index():
return render_template_string(INDEX_TEMPLATE, user=session.get("user"))
@app.route("/login", methods=["GET", "POST"])
def login():
error = None
if request.method == "POST":
username = request.form.get("username", "")
password = request.form.get("password", "")
try:
user = fetch_user(username, password)
except sqlite3.Error as exc:
error = f"SQL error: {exc}"
else:
if user:
session["user"] = user["username"]
return redirect(url_for("profile"))
error = "Login failed."
return render_template_string(LOGIN_TEMPLATE, error=error)
@app.route("/profile")
def profile():
user = session.get("user")
if not user:
return redirect(url_for("login"))
return render_template_string(PROFILE_TEMPLATE, user=user)
@app.route("/flag")
def flag():
if session.get("user") != "admin":
return "Admins only.\n", 403
return f"{FLAG}\n"
@app.route("/logout")
def logout():
session.clear()
return redirect(url_for("index"))
@app.route("/health")
def health():
return {"status": "ok"}
init_db()
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)
然后我们设计一个给靶场专用的CTF RL agent,他有三个模块组成,启动器,验证器,reward结算.
在启动器中,我会给模型对应的目标以及工具,从而让模型完成目标,并且做了限制,最多十步,最多两次工具失败调用:
这步是验证模型是否成功拿到flag,是否真的做了某些我们需要的操作,比如有没有发login包:
访问的时候有没有管理员权限
sql注入语句对不对 或者有没有相似的语句
我们之所以验证器做那么复杂的原因是因为如果不每一步做验证,只看拿不拿flag,我们这种弱智模型会遇到Reward Sparsity(奖励稀疏)的问题
因此我们必须设计好,每一步都有对应的奖励或者惩罚,防止出现模型摆烂的问题.而这个是reward结算来干的,如模型成功登录就给点分,欺骗欺诈agent扣分等。
这个过程非常有趣,因为模型一定不会走聪明路线,他会走偷懒的路线,哪条线容易就走哪个。
首先,用大模型合成一些正确的SFT过来(大概1k,包含正确的路线和思考),混在正常的SFT里面做训练,这个是保证模型至少读过数据。然后启动这个RL AGENT:
最开始,模型认为我们在骗他:
狠狠的扣分后,过了一个晚上,我发现模型开始上道了,认为这个是sql注入题目了
再过了一个晚上后,我发现模型居然作弊了:
这是因为,这个flask后台也是可以正常登录的,而这套系统是我vibe coding出来的,默认的GPT给了一个弱口令,结果模型不知道从哪一步开始猜到了口令,开始疯狂刷分。
回退版本并且重新设计了奖励结构后,继续放置,一天后,这个模型终于学会了sql注入,拿到了flag:
至此,我们的训练模型之旅也结束了
预训练这块坑比较多,花费也多成本也多,有条件的还是直接用别人家的base模型(已经预训练好的基座)吧,之后我也试了一下QWEN的BASE接了SFT和RL后做的真的比我的好太多了。