企业官网建设流程全解析

接到一个“用LangGraph多智能体做自动化性能测试”的题目，我第一时间想到的是自己第一次把大模型Agent塞进性能测试流程时的手忙脚乱。后来实践跑顺了才发现，这个组合真正的价值不在“能用AI写压测脚本”，而是把性能测试里最耗人工的分析、决策、报告编写环节拆给了多个各司其职的智能体，让整个流程从“半自动执行+人工判断”变成了“自动执行+自动决策+自动复盘”。这篇文章我会从设计思路、环境选型到核心代码、踩坑实录完整拆一遍，适合有基本性能测试经验、想引入大模型Agent但又不知道从哪下手的同学。

1. 为什么性能测试需要多智能体协作

1.1 传统性能测试流程的痛点

先还原一下大家熟悉的性能测试执行链条：需求分析、场景设计、脚本开发、压测执行、监控采集、结果分析、报告输出。这个链条看起来清晰，实际跑起来全是人工活。

我参与过的项目里，最常见的情况是：业务方提一个“首页查询接口需要支撑1000并发”的需求，测试人员要花大量时间去解析接口文档、设计测试数据、写JMeter脚本或者Locust脚本、设定阶梯加压策略、盯着监控面板看指标、出了瓶颈还要人工去查是DB慢还是线程池满了，最后再花两个小时整理一份从各种平台东拼西凑的报告。这个流程的问题在于：真正创造价值的“分析决策”环节消耗了大量时间在“数据搬运”上，而且不同环节的知识是割裂的。

这里还不提一个更隐蔽的问题：压测过程中改参数往往依赖个人经验。资深测试看一眼系统吞吐率和响应时延的对应关系，能判断应该加线程还是拆热点数据；但一个经验一般的执行者可能只会“继续加压”，结果把系统压到雪崩才发现方向错了。机器执行很擅长，决策判断很稀缺，这就是性能测试自动化的核心矛盾。

1.2 多智能体到底解决什么问题

LangGraph这个项目名字里带着“LangGraph”，核心是它基于图结构编排大模型智能（Agent）。它不是让一个Agent又写脚本又跑压测又写报告，而是把流程拆成多个专职Agent：需求解析Agent、压测执行Agent、监控分析Agent、报告生成Agent，外加一个编排器负责调度和状态管理。

举个例子。传统方式是测试人员手动把接口信息填进压测工具，然后凭经验设置“100并发起步，逐步加到500”，看到响应时间超过800ms了，手动停下来分析。在多智能体体系里，需求解析Agent先把接口文档和性能目标转成机器可执行的配置，压测执行Agent按阶梯加压策略自动跑，监控分析Agent实时对比指标和系统日志，一旦发现异常就触发评估Agent判断是继续加压还是暂停诊断，最后报告生成Agent自动汇总整轮测试的数据和结论。

这套协作机制解决的真正问题，是把“PerfTest专家经验”给结构化复制了。原本分散在团队成员脑子里的决策逻辑，被沉淀成Agent节点的判断标准。实测下来，同样一个“首页详情页”接口的完整压测流程，原先从写脚本到出报告需要大半天，多智能体跑完一轮大概是20分钟，而且每一步都有可追溯的决策日志。

1.3 适用边界与预期收益

不过我得泼一盆冷水，多智能体不是万能钥匙。它最适合的场景是：接口文档结构化程度高、性能指标定义明确、判定逻辑能写成规则或提示词（比如“P99小于200ms”“错误率小于0.1%”）的线上接口或核心链路压测。

对那种探索性很强、连业务流程都说不清楚、或者需要多人线下确认的复杂全链路压测，现阶段不建议直接上多智能体。因为智能体的“智能”依赖清晰的目标输入，目标模糊时它只会自作主张，做一个看起来正确但实际跑偏的决定。

我自己的预期收益参考：脚本生成时间减少80%，压测执行中的人工介入次数从五六次降到一次，报告从“人工整理2小时”变成“自动生成5分钟”。当然，这不是白来的，前期要把Agent的提示词和判定规则调明白，这个投入大概需要两三天。

2. 技术选型：为什么是LangGraph

2.1 工作流引擎选型对比

决定做多智能体性能测试平台的时候，我在工作流编排方案上有几个备选：直接写Python状态机、用LangChain的AgentExecutor、用LangGraph、或者自研一个基于队列的任务调度。

最简单的状态机方案我在早期Demo里试过，问题在于扩节点要改核心逻辑，状态分担不清晰，Add一个“监控异常回调”就够受的。LangChain的AgentExecutor更适合单Agent多工具场景，但多Agent协作时它的循环和终止条件控制不够细，很难表达“压测Agent执行完，监控Agent分析，如果指标不达标返回压测Agent继续调整”这种带环路的流程。

下面这个对比表是我当时整理的选型依据：

LangGraph胜在“图计算”这套抽象，正好能承接性能测试的复杂决策流。它的StateGraph把整个流程定义为一张有向图，每个节点就是一个处理逻辑（可以是调用LLM，也可以是纯Python函数），节点间的边决定了数据怎么流转。这让“压测—分析—是否重压—生成报告”这个带条件分支的流程，表达起来非常直观。

2.2 状态管理与图执行模型

另一个让我下决心的是LangGraph的State设计。它内部的状态是一个TypedDict，所有节点都可以读写，节点函数接收当前的state快照，返回一个字典作为state的增量更新。比如压测执行节点返回了一个load_test_result字段，后续的监控节点就能直接读取，不需要像在Java里那样设计一堆DTO和DAO。

这个设计放到性能测试场景里尤其舒服。因为性能压测天然是一个“多次迭代、逐步逼近”的过程：先跑一轮小并发，拿到基线数据，再翻倍并发，系统扛不住就缩回去找拐点。每一轮的结果都需要成为下一轮决策的输入，这种状态堆积用LangGraph的state叠加操作（比如用operator.add把多轮结果追加成列表）非常顺手。对比之下，传统Runner模式每轮都重置上下文的方案，在动态调参上就很笨拙。

另外一个实操体验：LangGraph的Graph可以单独运行每个节点并打印state变化，这在调试多智能体系统时价值巨大。我调试压测智能体时经常出现某一步产出不符合预期，直接打印那一步的输入输出state，一两分钟就能定位问题出在提示词还是工具调用。这种可观测性在Agent系统里比什么都重要。

3. 整体架构与智能体职责划分

3.1 四个智能体的分工

我实现的多智能体性能压测系统里，一共四个业务Agent和一个编排器。

需求解析Agent（需求分析角色）：输入是接口文档（OpenAPI/Swagger JSON）、压测目标描述、历史压测记录。输出是标准化的“压测任务书”，包括接口路径、请求方法、请求体模板、压测指标阈值（如P95 < 500ms）、预估并发量、持续时长、测试数据约束。

压测执行Agent（压测工程师角色）：接收任务书，动态生成Locust压测脚本（我用Locust做执行引擎），设置阶梯加压策略，启动压测。它会根据上一轮结果决定下一步的加压幅度，因此需要具备“读指标定参数”的能力。

监控分析Agent（性能分析师角色）：在压测过程中采集系统侧和应用侧指标（QPS、RT、错误率、CPU、内存、GC、线程池活跃数），同时关联日志关键词，输出异常根因线索。

报告生成Agent（报告撰写者角色）：把压测执行结果、监控分析结论、系统调优建议组织成结构化测试报告，包含趋势图数据、结论和可执行的优化清单。

这四个Agent串起来的数据流可以用一句话概括：需求解析Agent产出任务书，压测执行Agent按任务书跑并产出原始结果，监控分析Agent把原始结果翻译成问题和根因，报告生成Agent把问题和数据组织成决策者能看懂的文档。

3.2 状态流转与循环重试机制

LangGraph里最核心的设计是带条件的循环。不是简单的一条线走完，而是压测执行和监控分析之间可以形成环路。系统描述如下：

入口 -> 需求解析 -> 压测执行 -> 监控分析 -> 结果评估 | 如果指标不达标且未达到最大迭代次数 -> 回到压测执行（调整参数） | 如果指标达标或达到最大迭代次数 -> 报告生成 -> 结束

这里的“结果评估”节点不需要单独做一个Agent，它可以是一个轻量LLM调用或者纯规则判断器。我实际用的是“规则优先，LLM兜底”的策略：如果错误率超过1%或者P99超过阈值3倍，直接走“停止加压转诊断”分支，不需要LLM参与判断；只有边界情况才让评估Agent做一次带上下文的分析。

这个循环在LangGraph里实现得很干净，就是定义节点之间的条件边（conditional_edges），根据state里的评估结果决定下一跳是回到压测执行节点还是走向报告节点。这种图结构表达了真实压测过程里“试错—修正—再试”的螺旋，让系统从纯执行工具变成了能自主寻找性能瓶颈的测试系统。

4. 实战实现：从状态定义到图编排

4.1 环境准备与依赖安装

我当时的环境是Python 3.11，官方建议3.9以上。核心依赖有这么几个：

pip install langgraph langchain-openai locust psutil jinja2

• langgraph：图编排核心。
• langchain-openai：大模型接入，兼容OpenAI协议的接口都可以配，base_url指向你们自己的LLM服务就行。
• locust：压测执行引擎，纯Python写压测脚本，动态生成脚本最方便。
• psutil：采集本机资源指标，我拿它看压测机自身的CPU内存，避免执行机成为瓶颈。
• jinja2：报告模板渲染，生成最终HTML报告。

如果被测系统的监控数据要从云端监控平台拉，那就加一个对应的API客户端库。我这边的方案是直接读应用暴露的/metrics端点取Prometheus格式数据。

4.2 状态结构定义

先定义全局状态对象，这个对象就是“Agent之间的共享工单”。我定义得比较宽，不同节点各取所需：

from typing import TypedDict, Annotated, List import operator class PerfTestState(TypedDict): # 阶段一：需求解析产出 api_docs: dict # OpenAPI文档 perf_goal: str # 原始性能目标描述 test_plan: dict # 标准化压测任务书 # 阶段二：压测执行信息 load_config: dict # 当前压测参数：并发数、时长、阶梯策略 load_result: dict # 压测结果：QPS、RT分布、错误率 # 阶段三：监控分析信息 metrics: List[dict] # 系统监控指标序列 analysis: dict # 根因分析结论 # 阶段四：报告与过程记录 report: str # 最终报告 history: Annotated[List[dict], operator.add] # 每轮压测的关键参数与结果 errors: List[str] # 执行过程中的异常 iteration: int # 循环压测的轮次计数

history用Annotated[List, operator.add]标注，这是LangGraph里的Reducer机制，意思是每次节点返回的history不是覆盖而是追加。它帮我保留了每次压测的参数和结果，最后做趋势分析非常有用。

4.3 需求解析智能体

这个节点接收原始输入，调用LLM把非结构化的性能需求解析成结构化压测任务书。提示词我调了好几版，核心是让模型按JSON Schema输出，并且必须包含“阈值类型、目标值、并发参考、持续时间、压测脚本语言”这些字段。

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm = ChatOpenAI( model="your-llm-model", temperature=0, # 解析任务不需要创造性，必须近似确定性输出 base_url="http://your-llm-service:8000/v1", api_key="sk-xxx" ) REQ_ANALYZE_PROMPT = ChatPromptTemplate.from_messages([ ("system", """你是一名资深性能测试架构师。 你的任务是把用户提供的性能测试需求整理成标准压测任务书。 请严格按以下JSON格式输出： {{ "endpoint": "接口路径", "method": "HTTP方法", "request_body_template": "请求体模板，可为空", "target_metrics": {{ "type": "P95或P99或AVG", "threshold_ms": 数字, "error_rate_limit": 百分数 }}, "suggested_concurrency": 数字, "duration_sec": 数字, "ramp_up_steps": [["并发数", "持续秒数"], ...] }} 只输出JSON，不要输出任何解释。"""), ("human", "接口文档：{api_docs}\n性能目标：{perf_goal}") ]) def requirement_analyzer_node(state: PerfTestState) -> PerfTestState: chain = REQ_ANALYZE_PROMPT | llm raw = chain.invoke({ "api_docs": state["api_docs"], "perf_goal": state["perf_goal"] }).content # 解析JSON，加异常处理 import json try: plan = json.loads(raw.replace("```json", "").replace("```", "").strip()) except json.JSONDecodeError as e: state["errors"].append(f"需求解析输出不合法: {e}, 原始输出: {raw}") plan = default_plan(state["api_docs"]) state["test_plan"] = plan return state

这里有一个关键细节：temperature=0。需求解析这步最怕LLM自由发挥，把用户随口说的“越快越好”解读成P99小于100ms，这会让后面的压测执行Agent做出完全错误的决策。我甚至建议在提示词里加一句“如果目标不明确，请按行业默认值：P95小于500ms”，保证一定有输出。

4.4 压测执行与监控智能体

压测执行节点是系统里最“硬”的环节，它不能被LLM直接控制，必须走工具调用。我用的是Function Calling方案：让LLM根据任务书和上一轮结果决定“调用哪个压测工具函数、传什么参数”。

from langchain_core.tools import tool @tool def run_locust_load_test(script_content: str, concurrency: int, run_seconds: int) -> dict: """执行Locust压测，返回统计结果。""" import subprocess, json, tempfile, os with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: f.write(script_content) script_path = f.name # 执行 locust 压测，用 --headless 模式，关闭Web UI cmd = [ "locust", "-f", script_path, "--headless", "-u", str(concurrency), "-r", str(max(1, concurrency // 10)), # 每秒启动用户数 "-t", f"{run_seconds}s", "--csv", "/tmp/perf_result", "--csv-full-history" ] subprocess.run(cmd, capture_output=True, text=True, timeout=run_seconds + 60) # 解析CSV结果，取最新的stats结果 ... return result_dict

这里有几个我踩过坑的细节。Locust的-r参数是每秒启动用户数，如果并发数设置得很大比如5000，-r也是5000，很容易瞬间打爆被测系统，出现“预热即崩溃”的假象。所以我在工具函数里把启动速率限制在并发数的10%，同时设置一个最低2每秒、最高500每秒的上下限。

压测执行Agent拿到工具返回值后，会把结果聚合进state。同时它要判断“上一轮结果是否正常”来决定下一轮参数。这部分我们放在评估节点做，压测执行Agent只负责“按参数跑并返回结果”。

监控分析节点做的事就杂了：读CPU内存（psutil）、拉接口RT分布、扫应用日志关键错误。我把它做成工具集合，同样通过Function Calling让Agent调用：

@tool def collect_system_metrics() -> dict: """采集压测机自身及被测系统的基础指标。""" import psutil return { "cpu_percent": psutil.cpu_percent(interval=1), "memory_percent": psutil.virtual_memory().percent, "load_avg": psutil.getloadavg(), } @tool def query_application_metrics(endpoint: str) -> dict: """从Prometheus端点拉取应用指标。""" import requests # 实际逻辑：请求 /metrics，解析文本格式 # 返回{http_requests_total, http_request_duration_seconds}等 return aggregated_metrics

监控分析Agent的提示词我会要求它输出两个字段：conclusion（结论：是否达标/是否有瓶颈）和evidence（证据：哪些指标支撑这个结论）。这个结构的反向力量比想象中大——没有evidence字段的时候，这个Agent经常泛泛而谈“系统基本正常”，加了强制证据要求后，准确性明显提升。

4.5 评估与报告智能体

评估节点我上面说了，是一个“规则优先、LLM兜底”的决策器。伪代码如下：

def evaluator_node(state: PerfTestState) -> PerfTestState: it = state.get("iteration", 0) res = state.get("load_result", {}) th = state.get("test_plan", {}).get("target_metrics", {}) error_rate = res.get("error_rate", 0) p99 = res.get("p99_ms", 0) threshold = th.get("threshold_ms", 500) # 规则一：错误率超过2%，直接判定异常，停止加压 if error_rate > 0.02: state["analysis"] = {"verdict": "failed", "reason": "错误率过高", "next_action": "investigate"} return state # 规则二：P99超过阈值3倍，立即停止，避免雪崩 if p99 > threshold * 3: state["analysis"] = {"verdict": "failed", "reason": "响应时间严重超时", "next_action": "investigate"} return state # 规则三：P99未超过阈值，且iteration小于最大次数，继续加压 if p99 < threshold and it < state["test_plan"].get("max_iteration", 5): state["analysis"] = {"verdict": "pass_continue", "reason": f"当前P99={p99}ms，低于阈值{threshold}ms，尝试加压", "next_action": "ramp_up", "next_concurrency": res.get("concurrency", 0) * 2} return state # 规则四：其他情况交给LLM深度分析 ...

我只在规则三的边界情况才调LLM深度评估（比如P99在阈值的0.8到1.2倍之间波动，介于“明显达标”和“明显超限”之间）。这样做的好处是：90%的决策是确定性的，成本低、速度快、可复现；只有10%需要上下文理解，才消耗Token。

报告生成节点我会在第五节用一段完整实例展示。

4.6 图的组装与条件边

把所有节点连起来，是LangGraph最体现“图思维”的部分：

from langgraph.graph import StateGraph, END g = StateGraph(PerfTestState) g.add_node("requirement_analyzer", requirement_analyzer_node) g.add_node("load_generator", load_generator_node) g.add_node("monitor_analyzer", monitor_node) g.add_node("evaluator", evaluator_node) g.add_node("report_generator", report_generator_node) g.set_entry_point("requirement_analyzer") g.add_edge("requirement_analyzer", "load_generator") g.add_edge("load_generator", "monitor_analyzer") g.add_edge("monitor_analyzer", "evaluator") def route_after_evaluator(state: PerfTestState) -> str: analysis = state.get("analysis", {}) action = analysis.get("next_action", "finish") if action == "ramp_up" and state.get("iteration", 0) < state["test_plan"].get("max_iteration", 5): state["iteration"] = state.get("iteration", 0) + 1 return "load_generator" # 回到压测执行节点继续加压 return "report_generator" # 否则写报告收尾 g.add_conditional_edges("evaluator", route_after_evaluator, { "load_generator": "load_generator", "report_generator": "report_generator" }) g.add_edge("report_generator", END) app = g.compile()

这里需要注意add_conditional_edges返回的节点必须直接映射到字符串或节点名，我一开始误把路由函数返回成了evaluator自己，结果出现了无限循环。LangGraph虽然支持循环，但循环必须要有退出条件，否则就是死循环。我加了max_iteration这个状态字段作为“熔断”机制，防止某个异常情况让压测永不停歇地跑下去。

5. 跑通一个真实场景

5.1 被测系统与测试目标

为了验证这套系统，我拿了一个“某模拟商城项目的订单查询接口”作为被测对象。接口是GET /api/v1/orders/{order_id}，从Redis和MySQL联合读取，算是一个典型的读写混合接口。性能目标用一句话描述：“支持200并发稳定运行，P95响应时间小于300ms，错误率不超过0.5%”。

调用入口代码：

initial_state: PerfTestState = { "api_docs": { "openapi": "3.0.0", "paths": {"/api/v1/orders/{order_id}": {"get": {"parameters": [...]}}} }, "perf_goal": "支持200并发稳定运行，P95响应时间小于300ms，错误率不超过0.5%", "test_plan": {}, "load_config": {}, "load_result": {}, "metrics": [], "analysis": {}, "report": "", "history": [], "errors": [], "iteration": 0 } result = app.invoke(initial_state, config={"recursion_limit": 50})

5.2 配置与启动

压测脚本是压测执行Agent动态生成的。它根据接口文档里的参数定义，生成一个Locust的User类。核心逻辑是先看test_plan里的request_body_template，如果非空就自动生成POST类任务，否则生成GET类。这里有一个容易翻车的地方：接口文档里的order_id是个路径参数，Agent第一次生成脚本时没处理，直接把/api/v1/orders/{order_id}当成整串URL发出去，结果所有请求都404。

我加了下面这段修复代码，让Agent在生成URL前做参数的显式替换：

def normalize_url(template_url: str) -> str: """把路径参数替换成测试数据。""" import re # 扫描 {xxx} 片段，替换为模拟ID，这里用固定前缀+随机数 def repl(match): return f"ORD{random.randint(10000, 99999)}" return re.sub(r"\{(\w+)\}", repl, template_url)

日常接口压测里这类“路径参数未处理”的问题高频出现，多智能体也不例外，它只是把错误从“脚本运行时报错”提前到了“生成时主动修复”。所以我的建议是：压测执行Agent不要直接生成最终脚本，而是要先生成脚本再调用静态检查工具函数，把“生成”和“校验”拆成两步。

5.3 一次典型执行流程的观察

用LangGraph跑起来后，我可以逐步拉出每一轮的决策记录。这里展示一次真实的迭代轨迹（部分字段省略）：

这个例子很有教学意义。第一次跑到200并发时P95=312ms是明显超标的，系统立刻降回150并发，而不是继续愣头青地加压。这个决策完全自动完成，不需要人介入。最后它还能自己报出结论：“系统稳定支撑并发在175左右，200并发时P95劣化明显，瓶颈疑似在订单查询的DB连接池”。中间那句“瓶颈疑似”是监控分析Agent结合GC日志和DB慢查询日志得出的，比单纯看压测曲线要准确得多。

5.4 压测强度参数的调优心得

我想重点分享三个我在调参时候体会到的关键点。

第一，阶梯加压的步长不要固定，要看上一轮的错误率变化率。如果错误率从0%跳到0.3%，说明系统已经接近临界点，下一步的步长应该减半；如果错误率一直是0%，可以激进一点，按当前并发的80%步进。这个逻辑我直接写在了压测执行Agent的提示词里，比固定“每轮加50并发”科学得多。

第二，持续时间至少要覆盖慢请求的尾部。有些接口的P95好看，P99却突然劣化，原因是慢速请求要到第40秒才出现。所以我强制最小压测持续时间为90秒，哪怕任务书要求“快速跑一下”，也必须在90秒以上，否则采集到的P99数据不可信。

第三，压测机的CPU和内存指标必须进入监控Agent的分析范围。我之前遇到过一次奇怪的结果：QPS上不去且P95飙升，误判为“应用性能差”，结果后来发现是压测机自己CPU被打满了，Locust的event loop处理不过来了。加上自我监控后，系统能在报告中标注“压测机资源使用率过高，本次结果可能受压测机性能影响”，这个标注在产出报告时非常关键。

6. 常见问题与排查实录

6.1 LLM输出不稳定导致的“乱改参数”

最让我头疼的问题是LLM在生成压测参数时“自以为是”。有一次明明上一轮200并发P95已经超标，评估Agent也判定失败了，压测执行Agent在生成下一轮脚本时居然把并发悄悄改成500，结果把被测系统给压宕机了。查日志发现是Agent看到了原始需求里“支持2000并发”的字样，自作主张按2000去配置了。

解决办法是给压测执行Agent的工具函数加上严格的参数范围校验，在函数内部做参数白名单：

CONCURRENCY_LIMITS = {"min": 1, "max": 500} @tool def run_locust_load_test(script_content: str, concurrency: int, run_seconds: int) -> dict: concurrency = max(CONCURRENCY_LIMITS["min"], min(CONCURRENCY_LIMITS["max"], concurrency)) if concurrency > 300: # 超过300强制要求人工确认标记，否则拒绝执行 return {"error": "并发数超过安全阈值，拒绝执行"}

这个教训的本质是：LLM生成的参数必须经过确定性代码的约束，不能直接信任。Agent的建议只能作为“需求侧”输入，真正控制物理资源消耗的责任必须落在工具函数的护栏代码上。

6.2 并发压测与状态共享冲突

第二个坑是LangGraph的状态在节点执行时是串行的，但Locust压测本身是并发的。我一开始设计的是“单个压测执行节点跑一个Locust进程”，压测结束返回结果，流程没问题。但后来想加速，让压测执行节点同时启动多个Locust进程测不同的接口，就发现Locust的CSV结果文件路径冲突了，两个进程写同一个文件，最后解析出来的数据全是乱码。

解决办法很简单也老土：每个Locust进程用独立的临时目录和文件前缀，用tempfile.mkdtemp()生成唯一目录：

import tempfile workdir = tempfile.mkdtemp(prefix="locust_") csv_prefix = os.path.join(workdir, "perf")

同一个现象如果在线程模型里会表现为“共享一个全局变量导致结果互相覆盖”，也就是“状态持有者”错乱。在多智能体系统里，这个问题的解法一致：任何有副作用的外部资源（文件、数据库连接、端口）都要做隔离，不能因为Agent层是逻辑独立的就默认物理资源也独立。

6.3 长时间任务超时与心跳

一轮压测如果是200并发、持续90秒，加上分析和决策，整个图执行时间可能超过3分钟。LangGraph的Node默认没有超时控制，我遇到过一次压测子进程因为被测系统卡住而hang住，整个压测流程等了两小时才报错。这个体验非常恶劣。

解决方案是通过FastAPI或asyncio wrapper给工具函数加超时，或者用信号机制：

import signal class TimeoutError(Exception): pass def timeout_handler(signum, frame): raise TimeoutError("压测进程超时") signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(run_seconds + 120) # 压测时间 + 120秒缓冲 try: subprocess.run(...) finally: signal.alarm(0)

另外，如果顶层也接了一个异步Web框架，最好把LangGraph的invoke放到线程池里执行，这样HTTP请求不至于因为压测阻塞而断连。实际部署时可以把整条Graph封装成一个FastAPI的/run_test接口，同步转异步调用，前端轮询状态。

6.4 成本控制与Token消耗

多智能体系统最容易被忽略的是Token成本。我统计过一轮完整压测的Token消耗：需求解析约5000 token，压测执行Agent每次脚本生成约8000 token，每轮2次即16000，监控分析约15000，LLM兜底评估约5000，报告生成约6000。一轮5轮压测下来将近100k token。如果跑频繁，成本是按天飙升的。

压成本有几个有效做法：

• 优先用规则代替LLM，评估节点最多调一次LLM，其他时候走规则。
• 用长上下文模型时注意把历史记录压缩，不在每次调用时都把完整history发给大模型，可以只传最近两轮的关键指标摘要。
• 压测脚本生成这种“文本转代码”的任务，可以用小参数模型做初稿，再用大模型做校验。

我用一个小技巧：把历史压测结果从“完整JSON列表”压缩成“每行一个指标汇总字符串”。比如"round2: 并发100, P95=198ms, err=0.0%"，这段内容让模型理解的成本和完整JSON一样，但Token数省了60%。

最后的一些实话

这套多智能体性能测试系统跑通之后，我最深的感受是：它的智能程度没有宣传的那么神，但工程化收益是实实在在的。以前一个性能测试工程师一天只能完整跑透一两个接口的压测，现在能让Agent夜里自动跑二十个接口的回归压测，第二天早上直接看报告。这背后的核心不是Agent有多聪明，而是把“压测领域知识”拆成了可组合的节点，再用LangGraph的图模型把它固化下来。

如果你也想尝试，我建议先别上来就做四个Agent的完整系统，先把需求解析和报告生成这两个“纯文本处理型Agent”跑通，替换你原来手工整理文档的环节。等熟悉了LangGraph的State和Conditional Edge，再把压测执行和监控分析的Agent接进去。这个路线投入产出比最高，也不至于一开始就被工程复杂度劝退。

最后再分享一个实操小技巧：LangGraph的状态里可以放一个human_in_the_loop标志，当评估Agent判定“failed”且原因无法归类时，不要自动重试，而是暂停并通知人。性能测试这东西，自动化再厉害，有些线上系统的压测还是需要人来兜底的。留一扇人工介入的门，系统才真正可靠。