企业官网建设流程全解析

1. 为什么数据库压测总在“脚本能跑、结果不敢信”之间反复横跳

做 Python 压测 Mysql 和 Doris 这件事，很多人第一次写脚本时都会觉得挺简单：装个 pymysql，开个线程池，循环 insert 就完事了。但真正跑起来之后，问题一个接一个冒出来——连接池被打爆、Doris 的 Stream Load 和普通 insert 行为不一致、压测跑完不知道数据到底写进去多少、想加个“让模型帮我分析压测结果”的环节又发现每个脚本都要单独配一套 Key。

我自己最早压 Doris 的时候，用的是最朴素的pymysql直连，50 个线程每个插 20 条，跑完打印一个总耗时就算完事。结果第一次跑就翻车：连接数直接顶到上限，报Too many connections；第二次把连接池加上，又发现 Doris 的写入延迟和 Mysql 完全不是一个量级，用同一套参数去压两个库，出来的数字根本没法横向对比。

所以这篇要解决的不是“怎么写出一个能跑的压测脚本”，而是怎么搭一套可复用、结果可校验、还能顺手接上大模型做结果分析的数据库压测方案。核心链路分四段：

第一段是压测脚本本身的设计，包括连接池参数、并发模型、写入批次怎么定；第二段是 Mysql 和 Doris 两套目标库的差异化配置，因为这两个库的写入语义差别很大；第三段是结果校验，压测跑完必须能回答“到底成功写了多少条、失败的是哪些、耗时分布长什么样”；第四段是把压测脚本和结果校验都接到一个统一的 Key 上，这样你换模型、换分析脚本时不用到处改配置。

这里说的“统一 Key”，指的是用 TaoToken 这类聚合入口来管理模型调用凭证。压测脚本里如果需要调用模型做结果解读、生成压测报告、或者让模型帮你判断某次压测的 P99 是否异常，都可以走同一个 Base URL 和同一个 Key，不用在 Mysql 压测脚本、Doris 压测脚本、结果分析脚本里各维护一套。

适合谁看：已经会写基础 Python 脚本、但压测结果总是“看着像那么回事、细看又说不清”的后端或数据开发；以及想把压测链路和 AI 辅助分析串起来、又不想在每个脚本里重复配 Key 的人。

下面从环境准备开始，一步步把这条链路搭起来。我会把 Mysql 和 Doris 的配置差异、连接池参数、并发模型、结果校验 SQL、以及 TaoToken 的接入方式都写成可直接复制的形式。你跟着做，最后能拿到一个跑得稳、结果可对账、还能让模型帮你读报告的压测方案。

2. TaoToken 统一 Key 接入：让压测脚本和结果分析共用一套凭证

在正式写压测脚本之前，先把 TaoToken 这一层接进来。原因很简单：如果你的压测链路里只有数据库操作，那确实不需要模型；但一旦你想加“压测结果自动解读”“异常耗时归因”“生成对比报告”这些环节，就会涉及模型调用。与其等到后面再补，不如一开始就把凭证层统一好。

TaoToken 的定位是一个模型调用聚合入口，官网地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 入口是 https://taotoken.net/api 。它的作用是让你用同一个 Base URL 和同一个 API Key，去调用不同的模型，而不需要在每个脚本里分别配置不同厂商的地址和密钥。

对于压测场景来说，这个统一 Key 的价值体现在三个地方：

第一，压测脚本里如果需要模型参与，比如让模型根据压测参数生成测试数据分布、或者根据历史压测结果推荐并发数，可以直接在脚本里读同一个环境变量，不用为每个脚本单独配。

第二，结果校验环节如果要做“智能对账”，比如把 Mysql 和 Doris 的写入条数、耗时分布丢给模型做差异分析，模型调用的凭证和压测脚本用的是同一套。

第三，你后续如果换模型，比如从 A 模型换到 B 模型做报告生成，只需要改模型 ID，Base URL 和 Key 不用动。

接入方式分两步。第一步是拿到 Key，第二步是在压测项目里统一读取。

拿 Key 的入口在控制台的 API Keys 页面，地址是 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 。进去之后创建一个 Key，复制出来。注意这个 Key 只显示一次，建议直接写进项目的.env文件，不要硬编码在脚本里。

第二步是在压测项目根目录建一个.env文件，内容如下：

# .env TAOTOKEN_API_KEY=sk-你的实际key TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL_ID=claude-3-5-sonnet

然后在 Python 里统一读取。我习惯用一个config.py来管理，这样压测脚本、结果校验脚本、报告生成脚本都从这里拿配置：

# config.py import os from dotenv import load_dotenv load_dotenv() TAOTOKEN_API_KEY = os.getenv("TAOTOKEN_API_KEY") TAOTOKEN_BASE_URL = os.getenv("TAOTOKEN_BASE_URL", "https://taotoken.net/api") TAOTOKEN_MODEL_ID = os.getenv("TAOTOKEN_MODEL_ID", "claude-3-5-sonnet") # 数据库配置也统一放这里 MYSQL_CONFIG = { "host": os.getenv("MYSQL_HOST", "127.0.0.1"), "port": int(os.getenv("MYSQL_PORT", 3306)), "user": os.getenv("MYSQL_USER", "root"), "password": os.getenv("MYSQL_PASSWORD", ""), "database": os.getenv("MYSQL_DB", "test"), } DORIS_CONFIG = { "host": os.getenv("DORIS_HOST", "127.0.0.1"), "port": int(os.getenv("DORIS_PORT", 9030)), "user": os.getenv("DORIS_USER", "root"), "password": os.getenv("DORIS_PASSWORD", ""), "database": os.getenv("DORIS_DB", "test"), }

这里有个细节要注意：Doris 的 FE 查询端口默认是 9030，BE 的 HTTP 端口是 8040，Stream Load 走的是 8030。如果你用 pymysql 直连 Doris，连的是 9030，走的是 MySQL 协议。这一点和 Mysql 的 3306 不一样，配置里要区分开。

如果你用的是 Claude Code 这类编码工具来辅助写压测脚本，可以在工具里配置 Anthropic 兼容的 Base URL。Claude Code 的配置入口在 https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claude_code&utm_campaign=rewrite ，里面会告诉你 Base URL 和 Key 怎么填。配置好之后，你在写压测脚本时让工具帮你补全连接池参数、生成校验 SQL，都会走这个统一入口。

模型对话的调试入口在 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite ，你可以先在那里试一下 Key 能不能正常调通，再写进脚本。

接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，里面有不同语言调用方式的说明。Python 侧用 OpenAI SDK 或 Anthropic SDK 都可以，取决于你选的模型。

这一层配好之后，后面压测脚本里如果需要模型参与，直接from config import TAOTOKEN_API_KEY, TAOTOKEN_BASE_URL, TAOTOKEN_MODEL_ID就行。不用在每个脚本里重复写地址和密钥，换模型也只改.env一个地方。

有一点要提醒：TaoToken 是模型调用入口，不是数据库代理。压测脚本连 Mysql 和 Doris 走的还是各自的数据库连接，TaoToken 只负责模型调用那一层。不要把数据库连接也往 TaoToken 上套，那是两回事。

3. 可复制配置：Mysql 与 Doris 压测脚本的连接池、并发与写入参数

这一节是整篇的核心，给出可直接复制的压测脚本配置。我会把 Mysql 和 Doris 分开写，因为这两个库在写入行为上有本质差异，用同一套参数去压会得出误导性结论。

先看 Mysql 的压测脚本。Mysql 的写入是行级事务，每次commit都会落盘，所以并发模型和批次大小对吞吐影响很大。连接池用dbutils的PooledDB，参数如下：

# mysql_bench.py import time import pymysql from dbutils.pooled_db import PooledDB from concurrent.futures import ThreadPoolExecutor, as_completed from config import MYSQL_CONFIG MYSQL_POOL = PooledDB( creator=pymysql, host=MYSQL_CONFIG["host"], port=MYSQL_CONFIG["port"], user=MYSQL_CONFIG["user"], password=MYSQL_CONFIG["password"], database=MYSQL_CONFIG["database"], charset="utf8mb4", connect_timeout=10, read_timeout=30, write_timeout=30, maxconnections=100, mincached=5, maxcached=20, maxshared=10, blocking=True, cursorclass=pymysql.cursors.DictCursor, ) def mysql_insert_batch(thread_id, batch_size, table="tbl_point_query1"): conn = MYSQL_POOL.connection() success = 0 fail = 0 latencies = [] try: with conn.cursor() as cursor: for i in range(batch_size): data_id = thread_id * batch_size + i + 1 start = time.time() try: cursor.execute( f"INSERT INTO {table} (`key`, table_name, column_name, data_type) " f"VALUES (%s, %s, %s, %s)", (data_id, "value1", "value2", "value3"), ) conn.commit() success += 1 except Exception as e: conn.rollback() fail += 1 latencies.append(time.time() - start) finally: conn.close() return {"thread_id": thread_id, "success": success, "fail": fail, "latencies": latencies} def run_mysql_bench(threads=50, batch_size=20): start = time.time() results = [] with ThreadPoolExecutor(max_workers=threads) as executor: futures = [executor.submit(mysql_insert_batch, i, batch_size) for i in range(threads)] for future in as_completed(futures): results.append(future.result()) total_time = time.time() - start total_success = sum(r["success"] for r in results) total_fail = sum(r["fail"] for r in results) all_latencies = [l for r in results for l in r["latencies"]] all_latencies.sort() p50 = all_latencies[len(all_latencies) // 2] if all_latencies else 0 p99 = all_latencies[int(len(all_latencies) * 0.99)] if all_latencies else 0 return { "total_time": round(total_time, 3), "total_success": total_success, "total_fail": total_fail, "qps": round(total_success / total_time, 2) if total_time > 0 else 0, "p50": round(p50, 4), "p99": round(p99, 4), } if __name__ == "__main__": print(run_mysql_bench(threads=50, batch_size=20))

这段脚本和原始 excerpt 的区别在于：原始版本用 f-string 直接拼 SQL，有注入风险且无法复用；这里改成参数化查询。原始版本没有统计成功/失败条数，压测跑完只知道“插了”，不知道“插进去多少”；这里每个线程返回 success 和 fail，最后汇总。原始版本没有延迟分布，这里记录了每条 insert 的耗时，最后算 P50 和 P99。

再看 Doris 的压测脚本。Doris 走 MySQL 协议时，INSERT是同步写入，但 Doris 的写入路径和 Mysql 完全不同——它先写 WAL，再异步刷 BE。所以用同样的线程数和批次去压，Doris 的延迟通常比 Mysql 高，但吞吐可能更稳。Doris 的连接池配置要单独调：

# doris_bench.py import time import pymysql from dbutils.pooled_db import PooledDB from concurrent.futures import ThreadPoolExecutor, as_completed from config import DORIS_CONFIG DORIS_POOL = PooledDB( creator=pymysql, host=DORIS_CONFIG["host"], port=DORIS_CONFIG["port"], user=DORIS_CONFIG["user"], password=DORIS_CONFIG["password"], database=DORIS_CONFIG["database"], charset="utf8mb4", connect_timeout=10, read_timeout=60, write_timeout=60, maxconnections=80, mincached=5, maxcached=15, maxshared=5, blocking=True, cursorclass=pymysql.cursors.DictCursor, ) def doris_insert_batch(thread_id, batch_size, table="tbl_point_query1"): conn = DORIS_POOL.connection() success = 0 fail = 0 latencies = [] try: with conn.cursor() as cursor: for i in range(batch_size): data_id = thread_id * batch_size + i + 1 start = time.time() try: cursor.execute( f"INSERT INTO {table} (`key`, table_name, column_name, data_type) " f"VALUES (%s, %s, %s, %s)", (data_id, "value1", "value2", "value3"), ) success += 1 except Exception as e: fail += 1 latencies.append(time.time() - start) finally: conn.close() return {"thread_id": thread_id, "success": success, "fail": fail, "latencies": latencies} def run_doris_bench(threads=40, batch_size=25): start = time.time() results = [] with ThreadPoolExecutor(max_workers=threads) as executor: futures = [executor.submit(doris_insert_batch, i, batch_size) for i in range(threads)] for future in as_completed(futures): results.append(future.result()) total_time = time.time() - start total_success = sum(r["success"] for r in results) total_fail = sum(r["fail"] for r in results) all_latencies = [l for r in results for l in r["latencies"]] all_latencies.sort() p50 = all_latencies[len(all_latencies) // 2] if all_latencies else 0 p99 = all_latencies[int(len(all_latencies) * 0.99)] if all_latencies else 0 return { "total_time": round(total_time, 3), "total_success": total_success, "total_fail": total_fail, "qps": round(total_success / total_time, 2) if total_time > 0 else 0, "p50": round(p50, 4), "p99": round(p99, 4), } if __name__ == "__main__": print(run_doris_bench(threads=40, batch_size=25))

两个脚本的关键差异我列成表格，方便对照：

这里要特别说 Doris 的 commit 行为。Doris 走 MySQL 协议时，INSERT是自动提交的，你调conn.commit()不会报错但也没有实际事务语义。所以 Doris 脚本里失败重试要谨慎，不能像 Mysql 那样 rollback 后重插，否则可能产生重复数据。我的做法是 Doris 侧失败就记录，不自动重试，压测结束后统一对账。

如果你想让模型帮你根据压测目标推荐参数，比如“我要压 5000 QPS，线程数和批次怎么配”，可以在脚本里加一个调用 TaoToken 的函数：

# llm_helper.py from openai import OpenAI from config import TAOTOKEN_API_KEY, TAOTOKEN_BASE_URL, TAOTOKEN_MODEL_ID client = OpenAI(api_key=TAOTOKEN_API_KEY, base_url=TAOTOKEN_BASE_URL) def suggest_bench_params(target_qps, db_type="mysql"): prompt = f"我要对 {db_type} 做压测，目标 QPS 是 {target_qps}，请给出线程数和批次大小的建议，并说明理由。" resp = client.chat.completions.create( model=TAOTOKEN_MODEL_ID, messages=[{"role": "user", "content": prompt}], temperature=0.3, ) return resp.choices[0].message.content

这个函数用的是 OpenAI SDK 的调用方式，Base URL 指向 TaoToken 的 API 入口。你换成 Anthropic SDK 也可以，取决于.env里配的模型。这样压测脚本和模型调用共用一套凭证，不用额外配。

配置部分到这里就齐了。Mysql 和 Doris 各一套连接池参数、并发模型、写入逻辑，加上一个可选的模型辅助函数。下一节讲怎么验证这些配置真的跑通了，以及结果怎么对账。

4. 验证请求与成功结果：压测跑通后怎么确认数据真的写进去了

压测脚本能跑完不代表数据写对了。我见过太多次“脚本打印 All data inserted，去数据库一查条数对不上”的情况。所以这一节讲两件事：一是怎么验证压测请求本身是通的，二是怎么校验写入结果的一致性。

先验证请求通路。最直接的方式是跑一个小批量，比如 2 个线程各插 5 条，然后去数据库查。Mysql 侧：

SELECT COUNT(*) FROM test.tbl_point_query1; SELECT * FROM test.tbl_point_query1 ORDER BY `key` DESC LIMIT 10;

Doris 侧同样的 SQL，但要注意 Doris 的查询延迟可能比 Mysql 高，刚写完立刻查可能因为 compaction 还没完成而看到旧数据。等几秒再查：

SELECT COUNT(*) FROM test.tbl_point_query1;

如果条数对得上，说明写入通路没问题。如果对不上，先看压测脚本返回的 success 和 fail 计数。success 是脚本认为成功的条数，数据库实际条数是 ground truth，两者不一致通常有三种原因：

第一种是 Doris 的写入可见性延迟。Doris 的 INSERT 返回成功只代表 FE 接受了请求，BE 实际落盘和可见可能有秒级延迟。这种情况等几秒再查就能对上。

第二种是连接池配置问题。如果maxconnections设得太小，blocking=True会让线程排队等连接，压测耗时里包含了等待时间，但 success 计数还是准的。如果blocking=False，拿不到连接会直接抛异常，这时候 fail 计数会上去。

第三种是主键冲突或表结构不匹配。比如key字段是主键，多个线程生成的 data_id 有重叠，就会插入失败。原始 excerpt 里用thread_id * count + i + 1生成 ID，如果线程数和批次变了，ID 生成逻辑要跟着改，否则会撞主键。

验证请求通路的另一个方式是看压测脚本的输出结构。跑一次小批量，输出应该类似：

{ "total_time": 1.234, "total_success": 10, "total_fail": 0, "qps": 8.1, "p50": 0.0123, "p99": 0.0456 }

如果total_fail大于 0，先别急着加线程，先把失败原因打出来。可以在except里加一行print(f"thread {thread_id} insert failed: {e}")，跑一次看报什么错。常见的是Duplicate entry或Table doesn't exist。

请求通路验证完之后，做结果一致性校验。这一步是压测方案里最容易被忽略的。我的做法是压测前后各查一次条数，差值应该等于total_success：

# verify.py import pymysql from config import MYSQL_CONFIG, DORIS_CONFIG def count_rows(config, table="tbl_point_query1"): conn = pymysql.connect(**config, cursorclass=pymysql.cursors.DictCursor) try: with conn.cursor() as cursor: cursor.execute(f"SELECT COUNT(*) AS cnt FROM {table}") return cursor.fetchone()["cnt"] finally: conn.close() def verify_bench(before_mysql, before_doris, mysql_result, doris_result): after_mysql = count_rows(MYSQL_CONFIG) after_doris = count_rows(DORIS_CONFIG) mysql_delta = after_mysql - before_mysql doris_delta = after_doris - before_doris print(f"Mysql 预期写入: {mysql_result['total_success']}, 实际增量: {mysql_delta}") print(f"Doris 预期写入: {doris_result['total_success']}, 实际增量: {doris_delta}") if mysql_delta != mysql_result["total_success"]: print("Mysql 写入不一致，需要排查") if doris_delta != doris_result["total_success"]: print("Doris 写入不一致，可能还在 compaction，稍后重查")

跑完压测后调用这个校验，输出类似：

Mysql 预期写入: 1000, 实际增量: 1000 Doris 预期写入: 1000, 实际增量: 998 Doris 写入不一致，可能还在 compaction，稍后重查

Doris 差 2 条这种情况，等 10 秒再查通常就一致了。如果一直不一致，就要去看 BE 日志，可能是某个 BE 节点写入失败但 FE 没返回错误。

如果你想让模型帮你解读这个校验结果，可以把mysql_result和doris_result拼成 prompt 丢给 TaoToken：

def analyze_result(mysql_result, doris_result): prompt = f""" 以下是 Mysql 和 Doris 的压测结果，请分析差异并给出可能原因： Mysql: {mysql_result} Doris: {doris_result} """ resp = client.chat.completions.create( model=TAOTOKEN_MODEL_ID, messages=[{"role": "user", "content": prompt}], temperature=0.2, ) return resp.choices[0].message.content

这样压测、校验、分析三段都走同一个 Key，不用分别配。

验证环节的核心就一句话：脚本说成功不算成功，数据库条数对得上才算。每次压测跑完，先看 success/fail，再查前后条数差，两个都对上，这次压测结果才可信。

5. 本篇常见错排查：401、local proxy failed、reading choices、OAuth 报错怎么定位

压测链路里报错分两类：数据库侧的报错和模型调用侧的报错。数据库侧的报错相对直观，模型调用侧的报错因为多了一层网络和鉴权，排查起来容易绕弯路。这一节把常见的几类报错和定位方法列出来。

第一类：数据库连接报错

pymysql.err.OperationalError: (2003, "Can't connect to MySQL server on '127.0.0.1'")这种是连不上。先确认端口对不对：Mysql 是 3306，Doris 是 9030。再确认服务有没有起，telnet 127.0.0.1 9030试一下。如果 Doris 的 FE 起了但 BE 没起，连接能建立但写入会失败，报错通常是Failed to find backend。

pymysql.err.OperationalError: (1040, "Too many connections")是连接数打满。把maxconnections调小，或者把blocking设为 True 让线程排队。Doris 侧尤其要注意，FE 的连接数上限默认不高，压测线程别开太多。

pymysql.err.IntegrityError: (1062, "Duplicate entry 'xxx' for key 'PRIMARY'")是主键冲突。检查 data_id 生成逻辑，确保多线程之间不重叠。原始 excerpt 里thread_id * count + i + 1这个公式，如果count和实际批次不一致，就会撞。

第二类：模型调用报错

401 Unauthorized是最常见的。先检查.env里的TAOTOKEN_API_KEY有没有写错，有没有多余空格。然后确认 Base URL 是不是https://taotoken.net/api，注意结尾不要多加/v1，SDK 会自己拼。如果用的是 Anthropic SDK，Base URL 的拼法可能不同，参考接入文档里的说明。

local proxy failed这种报错通常和本地网络环境有关。先确认你的请求是直接发到 TaoToken 的 API 入口，没有经过额外的本地转发。检查.env里有没有残留的HTTP_PROXY或HTTPS_PROXY环境变量，有的话清掉。Python 里可以用os.environ.pop("HTTP_PROXY", None)在脚本开头清一下。

Error reading choices或reading choices这类报错，通常是响应体解析失败。原因可能是模型返回了非预期格式，或者请求被截断。先确认TAOTOKEN_MODEL_ID填的模型名是存在的，去模型对话页面确认一下。然后把temperature调低，max_tokens设大一点，避免返回被截断。

OAuth相关报错，如果你用的是 Claude Code 这类工具，可能是工具的认证方式和 API Key 方式冲突了。Claude Code 的配置入口在 https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claude_code&utm_campaign=rewrite ，按里面的说明配 Base URL 和 Key。如果工具里同时开了 OAuth 和 API Key，可能会互相覆盖，建议只保留一种。

第三类：压测结果异常

压测跑完 QPS 特别低，先看total_time里有没有包含连接池等待时间。如果blocking=True且maxconnections太小，线程大部分时间在等连接，QPS 自然上不去。把maxconnections调到和线程数匹配。

P99 特别高但 P50 正常，通常是偶发的连接重建或 Doris compaction。看压测期间有没有网络抖动，或者 Doris 是不是在后台做 compaction。这种情况多跑几次取稳定值。

total_fail大于 0 但数据库条数对得上，可能是失败后重试成功了。检查脚本里有没有自动重试逻辑，如果有，fail 计数和实际失败数会不一致。建议失败就记录，不自动重试，压测结束后统一对账。

第四类：配置类报错

KeyError: 'TAOTOKEN_API_KEY'是.env没加载。确认python-dotenv装了，load_dotenv()在读取环境变量之前调用了。如果.env文件和脚本不在同一目录，load_dotenv()要传路径。

ModuleNotFoundError: No module named 'dbutils'是依赖没装。pip install dbutils pymysql python-dotenv openai一把装齐。

Doris 侧如果报Unknown table 'tbl_point_query1'，先确认表建了没有，库名对不对。Doris 的库表是大小写敏感的，test.tbl_point_query1和test.TBL_POINT_QUERY1是两张表。

排查的核心思路是：先分清是数据库侧还是模型侧，再看是连接问题还是数据问题，最后看是配置问题还是代码问题。每一类报错都有对应的检查点，按顺序过一遍，大部分问题都能定位到。

6. 把压测、校验、分析串成一条可复用的链路

到这里，Mysql 和 Doris 的压测脚本、连接池配置、结果校验、模型调用都已经能跑通了。最后说一下怎么把这些串成一条可复用的链路，而不是每次压测都手动拼。

我的做法是建一个bench_runner.py，把压测、校验、分析三步串起来：

# bench_runner.py from mysql_bench import run_mysql_bench from doris_bench import run_doris_bench from verify import count_rows, verify_bench from llm_helper import analyze_result from config import MYSQL_CONFIG, DORIS_CONFIG def run_full_bench(mysql_threads=50, mysql_batch=20, doris_threads=40, doris_batch=25): before_mysql = count_rows(MYSQL_CONFIG) before_doris = count_rows(DORIS_CONFIG) mysql_result = run_mysql_bench(mysql_threads, mysql_batch) doris_result = run_doris_bench(doris_threads, doris_batch) verify_bench(before_mysql, before_doris, mysql_result, doris_result) analysis = analyze_result(mysql_result, doris_result) print("模型分析结果：") print(analysis) return {"mysql": mysql_result, "doris": doris_result, "analysis": analysis} if __name__ == "__main__": run_full_bench()

这样每次压测只需要调run_full_bench()，传入线程数和批次，剩下的校验和分析自动完成。模型分析那一步走的是 TaoToken 的统一 Key，和压测脚本共用.env里的配置。

如果你要长期做压测，建议把每次结果存下来，方便对比。可以存成 JSON，字段包括时间戳、线程数、批次、QPS、P50、P99、success、fail。下次压测时把历史结果一起丢给模型，让模型帮你判断这次是变好了还是变差了。

Coding Plan 的入口在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite ，如果你要长期跑压测和 Agent 类任务，可以看一下那边的方案，适合需要持续调用模型的场景。

最后说一个我踩过的坑：Doris 压测时不要用太高的并发去压单 BE 节点，BE 的写入线程池有限，并发过高会排队，QPS 反而下降。我试过把线程从 40 加到 80，QPS 没涨，P99 翻了一倍。后来降到 40，QPS 反而更稳。所以压测参数不是越大越好，找到拐点比堆线程更重要。

整套链路的核心就三件事：压测脚本要能区分 Mysql 和 Doris 的写入语义，结果校验要以数据库实际条数为准，模型调用要统一 Key 避免到处配。这三件做好，压测结果才敢拿去用。