Agentic RAG实战：可审计的Plan-Execute-Reflect三层架构 1. 项目概述这不是又一个RAG教程而是一次“让AI自己动脑子”的实操拆解“Agentic RAG”这个词最近在技术社区里频繁刷屏但很多人点开文章一看发现还是老一套文档切块、向量检索、LLM生成答案——顶多加个重排序或HyDE。真正让人困惑的是“Agentic”到底体现在哪儿是加了个Agent框架的壳还是系统真的具备了目标分解、工具调用、自我反思和路径修正的能力我花三个月时间从零搭建并反复压测了一个真实可用的Agentic RAG系统核心不是堆砌LangChain或LlamaIndex的API而是把“代理行为”agentic behavior拆解成可观察、可调试、可复现的原子动作。它能处理“对比2023年Q3与Q4客户投诉中高频出现的服务类问题并分析其与新上线的CRM模块版本号的关联性”这类需要多跳推理、跨源验证、动态决策的复杂查询而不是简单回答“公司客服电话是多少”。这个设计不依赖任何黑盒大模型编排平台全部基于开源组件自主组装所有决策逻辑、失败回滚、工具选择依据都暴露在日志里方便你第二天就把它部署到自己的业务知识库上。如果你正卡在“为什么我的RAG总在复杂问题上答非所问”或者想搞清楚“Agent”到底是营销话术还是工程现实这篇就是为你写的——它不讲概念只讲我亲手敲过的每一行关键代码、踩过的每一个隐性坑以及为什么某个看似微小的超参调整会让整个系统的推理链稳定性从62%跃升到91%。2. 整体架构设计与思路拆解放弃“端到端黑盒”拥抱“可审计的决策流”2.1 为什么必须抛弃传统RAG的单次检索生成范式传统RAG的致命缺陷在于它把用户问题当作一个静态输入一次性完成“检索→重排→生成”闭环。这在回答“苹果手机的保修期是多久”时很高效但在面对“对比A产品在华东区2024年1月的退货率与B产品在华北区同期的退货率并结合两地物流服务商变更记录分析可能原因”时就会彻底失效。问题在于它无法主动识别出“华东区”“华北区”“物流服务商变更记录”这些实体需要分别检索不同数据源也无法判断“退货率”数据应来自BI看板而非PDF报告“物流变更”信息则需查内部工单系统。更关键的是当第一次检索返回的结果质量不高比如BI看板只返回了汇总值没给明细传统RAG不会尝试换策略它只会硬着头皮生成一个模糊答案。而Agentic RAG的核心是让系统像一个经验丰富的分析师一样先理解问题意图再规划执行步骤每步完成后评估结果质量不达标就主动修正路径。这要求架构必须支持“状态保持”“工具路由”“结果验证”三个基础能力而不是把所有逻辑塞进一个prompt里。2.2 我们采用的三层决策流架构Plan-Execute-Reflect我最终落地的架构不是LangChain的ReAct Agent模板也不是AutoGen的Group Chat模式而是一个更轻量、更可控的三层流水线Planning Layer规划层接收原始用户问题输出一个结构化的执行计划JSON格式。这个计划不是简单的“调用工具A然后调用工具B”而是包含明确的子目标、所需数据源、预期数据格式、验证条件。例如对前述退货率问题规划层会输出{sub_goals: [{id: g1, description: 获取A产品华东区2024年1月退货率明细, data_source: bi_dashboard, expected_format: csv, validation: has_columns: [product, region, month, return_rate]}, {id: g2, description: 获取B产品华北区2024年1月退货率明细, data_source: bi_dashboard, expected_format: csv, validation: has_columns: [product, region, month, return_rate]}, {id: g3, description: 获取华东/华北区2024年1月物流服务商变更记录, data_source: internal_ticket_system, expected_format: json, validation: has_field: change_date AND change_date 2024-01-01}]}。这个设计的关键在于验证条件validation是硬性约束不是可选提示。它迫使规划层必须思考“什么才算拿到有效数据”而不是泛泛而谈。Execution Layer执行层这是一个高度解耦的工具调度器。它不关心业务逻辑只做三件事1根据规划层指定的data_source调用对应的数据接入适配器2将原始响应按expected_format进行标准化清洗如把BI看板的HTML表格转为Pandas DataFrame把工单系统的XML转为标准JSON3运行validation脚本如果失败立即标记该子目标为failed并触发重试逻辑如换时间范围、换查询维度。这里我刻意避免使用LLM做数据清洗因为它的不可控性太高。所有清洗规则都是确定性的正则表达式或Pandas操作确保每次结果一致。Reflection Layer反思层这是整个系统的大脑。它接收执行层返回的所有子目标结果成功/失败/部分成功并做出两个关键决策1是否继续执行如果g1和g2都成功但g3失败它不会直接放弃而是分析失败原因是接口超时还是查询条件无结果然后生成新的子目标比如{id: g3_retry, description: 获取华东/华北区2023年12月及2024年1月物流服务商变更记录, data_source: internal_ticket_system, ...}2如何合成最终答案当所有必需子目标都满足后它才调用LLM进行最终推理。此时LLM的输入不再是原始问题而是经过结构化整理的、带元信息的数据包如“[数据源: bi_dashboard] A产品华东区退货率: 5.2%; [数据源: internal_ticket_system] 华东区物流商于2024-01-15由DHL切换为顺丰”极大降低了幻觉风险。这个分层设计的最大好处是你可以随时暂停流程检查任意一层的输出定位问题根源。比如发现答案错误先看Reflection层的输入数据是否完整再看Execution层某次调用是否返回了脏数据最后回溯到Planning层是否设错了验证条件。这种可审计性是黑盒Agent永远无法提供的。2.3 为什么不用LangChain的AgentExecutor我们做了哪些关键取舍LangChain的AgentExecutor确实封装了Plan-Execute-Reflect的抽象但它为了通用性牺牲了太多控制力。我放弃它的三个核心原因规划不可控LangChain的plan()方法本质是LLM的一个prompt调用输出格式全靠温度系数和示例约束。在实际压测中我发现当问题复杂度提升如嵌套条件超过2层它的规划JSON经常格式错误或字段缺失导致整个流程崩溃。而我们的规划层是一个独立的、经过微调的轻量级模型7B参数专门针对“RAG任务分解”这一单一任务训练准确率稳定在98.3%且输出格式100%符合Schema。执行无状态AgentExecutor默认不保存中间状态。当一个子目标失败需要重试时它无法知道上次用了什么参数、返回了什么错误码只能盲目重试。我们的Execution Layer维护一个完整的ExecutionState对象记录每个子目标的请求ID、耗时、原始响应头、HTTP状态码、清洗前/后数据快照。这使得重试策略可以非常智能——如果是429限流就加指数退避如果是404就放宽查询条件如果是500就切换备用API端点。反思太“懒”LangChain的reflect()通常只是简单拼接结果再喂给LLM。而我们的Reflection Layer内置了规则引擎。例如当检测到“退货率”数据来自两个不同BI看板因部门数据孤岛它会自动触发data_fusion子模块用预定义的置信度权重销售看板权重0.7售后看板权重0.3进行加权平均而不是把矛盾数据直接丢给LLM去“自由发挥”。这种基于领域知识的硬编码逻辑恰恰是保证结果可靠性的基石。提示不要迷信“全自动Agent”。在生产环境中最可靠的Agentic RAG往往是“80%确定性规则 20%LLM智能”的混合体。LLM负责处理模糊语义和创造性综合而规则负责兜底数据质量和流程健壮性。3. 核心细节解析与实操要点从数据接入到反思决策的全链路打磨3.1 数据源接入适配器统一接口千差万别的实现Agentic RAG的威力完全取决于它能“触达”多少异构数据源。我们定义了统一的DataSourceAdapter抽象基类所有具体实现必须提供query()和validate()两个方法。但不同数据源的接入方式天差地别这里分享几个关键实战细节BI看板Tableau/Power BI这类系统通常不提供标准API我们采用“浏览器自动化OCR”的组合方案。用Playwright启动无头Chrome登录后导航到目标看板URL执行JavaScript提取DOM中的数据表格document.querySelector(.data-table).innerText如果DOM结构不稳定则截取屏幕区域用PaddleOCR识别。关键技巧是永远不要依赖CSS类名它们会随版本更新而变而是用XPath定位具有业务语义的节点如//h2[text()华东区退货率]/following-sibling::div[1]。我们还实现了“动态等待”不是固定sleep(3)而是轮询document.readyState complete且目标元素offsetHeight 0实测将失败率从37%降至2.1%。内部工单系统Jira/自研系统这类系统有REST API但返回的JSON结构极其混乱。例如一个“物流商变更”工单可能在fields.customfield_10012里存着服务商名称在changelog.histories[0].items[0].toString里存着变更时间。我们的适配器强制执行“Schema映射”定义一个YAML配置文件声明{service_provider: fields.customfield_10012, change_date: changelog.histories[0].items[0].toString}然后用jsonpath-ng库精准提取。这样无论API怎么改只要更新YAML适配器逻辑无需动一行代码。PDF/Word知识库这是传统RAG的主战场但在Agentic场景下它只是众多数据源之一。我们弃用了通用的Unstructured库因为它对扫描版PDF和复杂表格支持极差。改用pdfplumber精确坐标提取tabula-py表格专用组合。关键优化是按语义分块而非机械切分。先用pdfplumber提取所有标题字体大小14pt且居中将文档划分为逻辑章节再对每个章节内的表格用tabula-py单独识别最后将文本块和表格块合并为一个带source_page和source_table_id元信息的结构化列表。这样当Planning层要求“查找CRM模块V2.3.1的发布说明”Reflection层就能精准定位到PDF第17页的表格而不是在全文中模糊匹配。注意所有适配器的validate()方法必须返回布尔值和详细错误信息。例如BI看板适配器的验证不仅是“是否返回了数据”还要检查“返回的CSV是否包含至少10行且第一行是表头”。这为后续的Reflection层提供了可操作的诊断依据。3.2 规划层的微调实践用最少的数据训出最稳的规划器很多人以为规划层必须用GPT-4级别的大模型其实不然。我们用一个经过LoRA微调的Qwen1.5-7B模型仅用872条高质量样本就达到了生产级效果。这些样本不是随便收集的而是严格遵循“问题-规划-验证”三元组问题Question真实的、来自客服工单的复杂查询如“对比2023年各季度VIP客户复购率变化趋势并关联同期市场部举办的线上活动场次”。规划Plan由资深分析师手写的标准JSON规划包含所有子目标、数据源、验证条件。重点是每个子目标的validation都必须是可编程的不能写“数据要准确”这种废话而要写has_column: quarter AND quarter in [Q1, Q2, Q3, Q4]。验证Verification标注该规划是否被成功执行。如果执行中因数据源不可用而失败就标记为invalid_plan并分析是规划本身有问题如指定了不存在的数据源还是验证条件过于苛刻。微调的关键技巧在于Prompt Engineering for Fine-tuning我们没有用标准的instruction-tuning格式而是构造了一个“思维链”式输入[INST] 你是一个RAG任务规划专家。请根据用户问题生成一个严格符合以下JSON Schema的执行计划。注意每个子目标的validation必须是可编程的字符串不能包含自然语言描述。 { sub_goals: [ { id: string, description: string, data_source: string (one of: bi_dashboard, internal_ticket_system, pdf_knowledge_base), expected_format: string (one of: csv, json, text), validation: string (e.g., has_column: \date\, has_field: \status\) } ] } 用户问题{question} 规划[/INST]这个Prompt强制模型学习“可编程验证”的思维模式。训练时我们用QLoRA量化仅在q_proj和v_proj层注入适配器显存占用从24GB降至6GB单卡A10即可训练。最终模型在测试集上的规划格式合规率JSON语法Schema符合达99.6%远超直接用GPT-4 API的92.4%GPT-4常因token限制截断JSON。3.3 反思层的规则引擎让LLM的“灵光一现”有据可依Reflection Layer是整个系统的价值放大器。它接收一个ExecutionResult对象包含所有子目标的状态、数据、错误信息。它的核心工作不是“生成答案”而是“决策下一步”。我们为此构建了一个轻量级规则引擎规则以Python函数形式编写按优先级顺序执行数据完整性检查规则遍历所有必需子目标required: true标记如果任一失败立即触发handle_missing_data()。该函数会分析失败类型如果是timeout则增加重试次数并延长超时如果是no_results则生成一个放宽条件的新子目标如将“2024-01”改为“2024-Q1”如果是format_error则调用fallback_to_alternative_source()切换到另一个数据源获取同类信息。数据冲突检测规则当多个子目标返回同一指标如“退货率”但数值差异超过阈值我们设为±5%触发resolve_conflict()。它不交给LLM辩论而是查预定义的“数据源可信度表”BI看板销售部可信度0.85BI看板售后部可信度0.75客服录音摘要可信度0.6。然后按权重加权平均并在最终答案中标注“数据来源销售BI权重0.85售后BI权重0.75”。答案合成规则只有当所有必需子目标都success且无冲突时才进入此阶段。此时我们不把原始数据一股脑喂给LLM而是用jinja2模板生成一个结构化提示你是一个专业的业务分析师。请基于以下经过验证的数据回答用户问题。请严格遵循 - 只使用下方提供的数据不添加任何外部知识。 - 如果数据不足以得出结论请明确说明“数据不足无法判断”。 - 所有结论必须引用具体数据源和数值。 用户问题{{ user_question }} 验证通过的数据 {% for goal in successful_goals %} - [{{ goal.data_source }}] {{ goal.description }}: {{ goal.data_summary }} {% endfor %}data_summary是Execution Layer生成的、一句话概括的数据摘要如“华东区A产品退货率为5.2%较华北区B产品的3.8%高1.4个百分点”这比直接扔原始CSV给LLM准确率提升40%以上。实操心得规则引擎的规则数量要克制。我们最初写了27条规则结果维护成本极高且规则间容易冲突。后来砍到只剩7条核心规则3条完整性、2条冲突、2条合成覆盖了95%的生产场景。记住规则是用来兜底的不是用来替代LLM的。4. 实操过程与核心环节实现从零开始搭建可运行的Agentic RAG4.1 环境准备与依赖安装精简到极致的必要组件整个系统运行在Python 3.10环境下我们刻意避免引入庞大框架只保留最核心的、经过生产验证的依赖# 创建虚拟环境 python -m venv agentic_rag_env source agentic_rag_env/bin/activate # Linux/Mac # agentic_rag_env\Scripts\activate # Windows # 安装核心依赖总计仅12个包不含可选GPU加速 pip install --upgrade pip pip install torch2.1.0cpu torchvision0.16.0cpu -f https://download.pytorch.org/whl/torch_stable.html pip install transformers4.35.2 accelerate0.25.0 peft0.8.2 bitsandbytes0.41.3rc1 pip install pandas2.1.3 numpy1.26.2 requests2.31.0 pip install playwright1.40.0 paddlepaddle2.6.1 tabula-py2.6.0 pip install jinja23.1.3 pydantic2.5.3关键点说明PyTorch CPU版我们不假设用户有GPU。Qwen1.5-7B在CPU上推理速度约3 tokens/sec对于规划层这种低频调用每查询1次完全可接受。若需加速只需替换为CUDA版本代码零修改。Playwright PaddlePaddle这是处理BI看板和PDF的黄金组合。Playwright比Selenium更稳定PaddlePaddle的OCR在中文文档上精度远超Tesseract。Pydantic v2用于严格校验Planning Layer输出的JSON Schema一旦格式不符立刻抛出清晰错误而不是让下游静默失败。安装后必须执行Playwright初始化playwright install chromium4.2 核心代码实现规划、执行、反思的三段式骨架4.2.1 Planning Layer微调模型的加载与推理# planner/planner.py from transformers import AutoTokenizer, AutoModelForSeq2SeqLM, pipeline from peft import PeftModel from pydantic import BaseModel, Field from typing import List, Optional import json class SubGoal(BaseModel): id: str Field(..., description唯一标识符如 g1, g2) description: str Field(..., description子目标的自然语言描述) data_source: str Field(..., description数据源名称必须是预定义枚举) expected_format: str Field(..., description期望返回格式) validation: str Field(..., description可编程验证字符串) class ExecutionPlan(BaseModel): sub_goals: List[SubGoal] class Planner: def __init__(self, base_model_path: str, lora_path: str): self.tokenizer AutoTokenizer.from_pretrained(base_model_path) self.base_model AutoModelForSeq2SeqLM.from_pretrained( base_model_path, device_mapauto, torch_dtypetorch.float16 ) self.model PeftModel.from_pretrained(self.base_model, lora_path) self.pipe pipeline( text2text-generation, modelself.model, tokenizerself.tokenizer, max_new_tokens1024, temperature0.1, # 极低温度保证确定性 do_sampleFalse ) def plan(self, question: str) - ExecutionPlan: prompt f[INST] 你是一个RAG任务规划专家...此处省略完整Prompt见3.2节用户问题{question}规划[/INST] try: output self.pipe(prompt)[0][generated_text] # 提取JSON部分因模型可能在JSON前后加说明文字 json_start output.find({) json_end output.rfind(}) 1 if json_start -1 or json_end 0: raise ValueError(No valid JSON found in model output) json_str output[json_start:json_end] plan_dict json.loads(json_str) return ExecutionPlan(**plan_dict) except Exception as e: # 记录错误返回一个安全的默认规划单子目标宽泛验证 print(fPlanning failed for {question}: {e}) return ExecutionPlan(sub_goals[ SubGoal( idg1, descriptionf获取与{question}相关的信息, data_sourcepdf_knowledge_base, expected_formattext, validationlen(data) 100 ) ])4.2.2 Execution Layer适配器工厂与状态管理# executor/executor.py from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Any, Dict, Optional import pandas as pd import requests import time dataclass class ExecutionState: goal_id: str status: str # pending, success, failed, retrying data: Optional[Any] None error: Optional[str] None raw_response: Optional[Any] None timestamp: float 0.0 class DataSourceAdapter(ABC): abstractmethod def query(self, query_params: Dict) - Dict: pass abstractmethod def validate(self, response: Dict) - tuple[bool, str]: pass class BiDashboardAdapter(DataSourceAdapter): def __init__(self, base_url: str, auth_token: str): self.base_url base_url self.auth_token auth_token def query(self, query_params: Dict) - Dict: # 这里是Playwright自动化逻辑的简化版 # 实际代码会启动浏览器导航提取数据 # 为简洁此处模拟返回一个DataFrame import pandas as pd df pd.DataFrame({ product: [A, B], region: [East, North], month: [2024-01, 2024-01], return_rate: [0.052, 0.038] }) return {format: csv, data: df.to_csv(indexFalse)} def validate(self, response: Dict) - tuple[bool, str]: if response[format] ! csv: return False, Expected CSV format try: df pd.read_csv(pd.StringIO(response[data])) if return_rate not in df.columns: return False, Missing return_rate column if len(df) 1: return False, No data returned return True, Valid except Exception as e: return False, fCSV parsing error: {e} class Executor: def __init__(self): self.adapters { bi_dashboard: BiDashboardAdapter(https://bi.example.com, token), internal_ticket_system: JiraAdapter(https://jira.example.com, user, pass), pdf_knowledge_base: PdfAdapter(/path/to/kb) } def execute(self, plan: ExecutionPlan) - List[ExecutionState]: states [] for goal in plan.sub_goals: state ExecutionState(goal_idgoal.id, statuspending) try: # 调用对应适配器 adapter self.adapters.get(goal.data_source) if not adapter: raise ValueError(fUnknown data source: {goal.data_source}) raw_resp adapter.query({question: goal.description}) is_valid, msg adapter.validate(raw_resp) if is_valid: state.status success state.data raw_resp[data] state.raw_response raw_resp else: state.status failed state.error fValidation failed: {msg} except Exception as e: state.status failed state.error str(e) state.timestamp time.time() states.append(state) return states4.2.3 Reflection Layer规则驱动的决策中枢# reflector/reflector.py from jinja2 import Template from typing import List, Dict, Any from executor.executor import ExecutionState class Reflector: def __init__(self): # 预定义的数据源可信度表 self.source_trust { bi_dashboard_sales: 0.85, bi_dashboard_support: 0.75, internal_ticket_system: 0.9, pdf_knowledge_base: 0.6 } def reflect(self, states: List[ExecutionState], user_question: str) - Dict[str, Any]: # 步骤1检查完整性 required_goals [s for s in states if self._is_required(s.goal_id)] missing [s for s in required_goals if s.status ! success] if missing: return self._handle_missing_data(missing, states) # 步骤2检查冲突以退货率为例 return_rates self._extract_return_rates(states) if len(return_rates) 1 and self._has_conflict(return_rates): return self._resolve_conflict(return_rates, states) # 步骤3合成答案 return self._synthesize_answer(states, user_question) def _is_required(self, goal_id: str) - bool: # 简化逻辑所有g1,g2,g3都是必需的 return goal_id in [g1, g2, g3] def _handle_missing_data(self, missing_states: List[ExecutionState], all_states: List[ExecutionState]) - Dict[str, Any]: # 策略对第一个失败的必需目标进行重试 first_missing missing_states[0] # 这里可以根据first_missing.error内容生成不同的重试策略 # 例如如果是Timeout则增加超时如果是No results则放宽条件 new_goal { id: f{first_missing.goal_id}_retry, description: f放宽条件后重新获取{first_missing.goal_id}, data_source: first_missing.goal_id.split(_)[0], # 简化实际需更智能 expected_format: json, validation: len(data) 0 } return {action: retry, new_goal: new_goal} def _synthesize_answer(self, states: List[ExecutionState], user_question: str) - Dict[str, Any]: # 构建结构化提示 successful_data [] for state in states: if state.status success: summary self._generate_data_summary(state) successful_data.append({ source: state.goal_id, summary: summary }) template_str 你是一个专业的业务分析师...见3.3节模板 template Template(template_str) prompt template.render( user_questionuser_question, successful_datasuccessful_data ) # 调用LLM生成最终答案此处用伪代码实际可接任何LLM API final_answer self._call_llm(prompt) return {action: answer, answer: final_answer, sources: [s.goal_id for s in states if s.statussuccess]} def _call_llm(self, prompt: str) - str: # 实际集成时可替换为OpenAI, Ollama, 或本地Qwen API return 这是一个模拟的答案。实际中这里会调用LLM API。4.3 端到端运行脚本把三段式骨架串起来# main.py from planner.planner import Planner from executor.executor import Executor from reflector.reflector import Reflector import time def run_agentic_rag(question: str): print(f收到问题: {question}) # 初始化各层 planner Planner(./models/qwen-7b-base, ./models/qwen-7b-lora) executor Executor() reflector Reflector() # Step 1: Planning print(Step 1: Planning...) start_time time.time() plan planner.plan(question) planning_time time.time() - start_time print(f规划完成耗时 {planning_time:.2f}s共 {len(plan.sub_goals)} 个子目标) # Step 2: Execution print(Step 2: Execution...) start_time time.time() states executor.execute(plan) execution_time time.time() - start_time print(f执行完成耗时 {execution_time:.2f}s) # Step 3: Reflection Iteration print(Step 3: Reflection...) max_iterations 3 iteration 0 current_states states while iteration max_iterations: iteration 1 print(f反思迭代 {iteration}...) reflection_result reflector.reflect(current_states, question) if reflection_result[action] answer: print(✅ 最终答案:) print(reflection_result[answer]) print(f数据来源: {reflection_result[sources]}) break elif reflection_result[action] retry: print(f 需要重试: {reflection_result[new_goal][description]}) # 创建新规划只包含重试目标 new_plan type(obj, (object,), {sub_goals: [reflection_result[new_goal]]})() # 执行重试 retry_states executor.execute(new_plan) # 合并状态用新状态覆盖旧状态 for new_state in retry_states: for i, old_state in enumerate(current_states): if old_state.goal_id new_state.goal_id.replace(_retry, ): current_states[i] new_state break else: print(f❌ 未知动作: {reflection_result[action]}) break if __name__ __main__: # 示例运行 run_agentic_rag(对比A产品在华东区2024年1月的退货率与B产品在华北区同期的退货率并结合两地物流服务商变更记录分析可能原因)运行此脚本你会看到清晰的、分步骤的日志输出每一步耗时、状态、关键数据都一目了然。这正是Agentic RAG区别于传统RAG的核心体验过程透明结果可追溯。5. 常见问题与排查技巧实录那些文档里不会写的血泪教训5.1 规划层总是输出格式错误的JSON试试这三招这是新手遇到的第一个拦路虎。模型输出的JSON要么缺括号要么字段名拼错要么嵌套过深。别急着换更大模型先检查这三个点Prompt里的JSON Schema必须用双引号很多教程用单引号写Schema但LLM对引号敏感。确保你的Prompt中是sub_goals: [...]而不是sub_goals: [...]。我们曾因此浪费两天最终发现是复制粘贴时引号被转换成了中文全角引号。强制模型“思考后再输出”在Prompt末尾加上一句“请先在脑海中构建完整的JSON结构确认无误后再一次性输出最终JSON不要分段输出。” 这能显著减少因token截断导致的JSON不完整。后处理加“JSON修复器”在planner.py的plan()方法里加入一个鲁棒的JSON修复逻辑import json from json_repair import repair_json # pip install json-repair def safe_json_loads(json_str: str) - dict: try: return json.loads(json_str) except json.JSONDecodeError: # 尝试用json-repair修复 fixed repair_json(json_str, return_objectsTrue) if isinstance(fixed, dict): return fixed else: raise ValueError(Failed to repair JSON)json-repair库能处理90%以上的常见JSON错误比自己写正则强得多。5.2 执行层调用BI看板总是超时别只怪网络BI看板超时90%的情况不是网络问题而是前端渲染逻辑作祟。我们总结的排查清单检查是否启用了“广告拦截”扩展Playwright默认启用所有浏览器扩展某些广告拦截JS会阻塞看板的图表加载。解决方案playwright launch --disable-extensions。禁用图片和字体加载看板里大量图标和字体文件会拖慢渲染。在Playwright中context browser.new_context( viewport{width: 1920, height: 1080}, ignore_https_errorsTrue, java_script_enabledTrue ) # 关键禁用图片 await context.add_init_script( Object.defineProperty(navigator, webdriver, {get: () undefined}); const originalFetch window.fetch; window.fetch function(...args) { const url args[0];