1. 什么是 AI Agent它不是“更聪明的聊天机器人”而是可执行、可调度、可容错的软件实体你可能已经用过扣子Coze、Dify 或 LangChain 搭建过一个“能查天气、能写周报、能读PDF”的智能体但很快会发现它在测试环境跑得飞快一上生产就卡死用户连续发三条指令第二条就丢失上下文调用数据库工具时偶尔返回空结果却不会重试或降级甚至某次模型返回格式错乱的 JSON整个流程直接崩掉——这不是模型不行是 Agent 的工程骨架没搭稳。AI Agent 的本质不是 LLM 加个提示词而是以大语言模型为认知中枢、以结构化决策流为神经回路、以工具调用为四肢末端、以状态记忆为短期海马体、以错误恢复为自主免疫系统的完整软件系统。它必须像传统后端服务一样经得起并发压测、扛得住输入噪声、容得下工具故障、留得住关键状态。热搜里反复出现的“ai agent 怎么扛并发”“agent安全”“llm request failed: provider rejected the request schema or tool payload”背后全是工程落地的真实痛感。我从 2022 年底开始在金融风控和工业运维场景落地 Agent做过 7 个千级 QPS 的线上智能体服务也踩过把“循环机制”写成无限递归导致 CPU 100% 的坑。今天这篇不讲概念、不画架构图、不堆术语只拆解一个真实可复现的 Agent 工程实现路径从七要素Goal、State、Memory、Planning、Action、Observation、Reflection出发映射到七个必须由工程师亲手编码、调试、压测、监控的决策点。这七个点就是 Agent 从 Demo 走向 Production 的分水岭。无论你用 Python 还是 Rust用 LangChain 还是自研框架只要想让 Agent 真正干活就绕不开它们。下面每一节我都配了真实日志片段、参数计算依据、压测数据对比和线上熔断配置你可以直接抄作业。2. 七要素不是理论模型而是七个必须落地的工程模块很多教程把七要素讲成抽象概念比如“Memory 是记忆模块”“Planning 是规划模块”。但工程上每个要素都对应一段有明确输入输出、可单元测试、可埋点监控、可独立替换的代码。我们逐个拆解它们在真实系统中的物理形态和设计约束。2.1 Goal目标不是一句自然语言而是带校验规则的结构化契约用户说“帮我分析上季度销售数据异常”这不能直接当 Goal。工程上Goal 必须解析为结构化对象包含业务域标识sales_anomaly、时间范围2024-Q2、数据源约束仅限 CRM 和 ERP 表、输出格式要求Markdown 关键指标表格、SLA 承诺≤15 秒响应。否则后续所有 Planning 和 Action 都会因目标模糊而失控。我们采用两级校验第一层是 Schema 校验JSON Schema拒绝非法字段第二层是语义校验轻量规则引擎例如检测到“上季度”但当前是 1 月则自动修正为 2023-Q4并记录 warning 日志。实测发现37% 的线上失败源于 Goal 解析阶段的歧义比如用户输入“最近三天”系统需明确是“最近 72 小时”还是“最近三个工作日”。提示不要依赖 LLM 做 Goal 解析。我们用 spaCy 自定义规则库做前置标准化LLM 只负责最终意图确认。LLM 解析 Goal 的平均延迟是 820ms而规则引擎是 12ms且 100% 可控。2.2 State状态不是全局变量而是带版本、带 TTL、带变更溯源的分布式快照Agent 在执行中需要维护对话状态、任务进度、工具调用历史等。常见错误是把 State 存在内存里或简单塞进 Redis Hash。但真实场景中一个 Goal 可能跨多个微服务、多个容器、多次重试State 必须满足一致性同一 Goal 的所有子任务看到相同 State 版本时效性过期 State 自动清理如用户 5 分钟无操作则释放资源可追溯每次 State 更新记录变更者planning_module / action_executor、变更原因tool_success / timeout_retry、变更前值diff 记录。我们采用“State Snapshot Change Log”双存储模式Snapshot 存于 RedisTTL300sChange Log 存于 Kafka保留 7 天。当发生冲突时以 Log 中最新 commit_id 为准。压测显示单节点每秒可处理 1200 次 State 更新而纯内存方案在 200 QPS 时就开始丢状态。2.3 Memory记忆不是缓存而是分层、分权、分粒度的持久化知识网络Agent 的 Memory 分三层Short-term对话级存于 State 中生命周期与 Goal 绑定Medium-term用户级存于用户专属向量库Weaviate含偏好、历史成功动作、常用工具集TTL90 天Long-term组织级存于关系型数据库PostgreSQL含 SOP 文档、权限策略、审计日志永久保存。关键设计是权限隔离Medium-term Memory 对每个用户加密隔离AES-256-GCMKey 由用户 ID 租户密钥派生Long-term Memory 则按 RBAC 控制访问避免 Agent 误用敏感 SOP。曾有案例某 Agent 因未隔离 Medium-term Memory将 A 用户的报销偏好应用到 B 用户导致审批流程错乱。2.4 Planning规划不是生成一段文字而是生成可验证、可中断、可回滚的执行树LLM 输出的“先查数据库再调 API最后总结”只是伪规划。工程上Planning 模块必须输出Execution GraphDAG 结构节点含 typesql / http / llm_call、input_schema、timeout_ms、retry_policyValidation Hook每个节点执行前校验输入是否符合 schema不符合则 abort 并返回 error_codeRollback Path标注哪些节点支持回滚如 SQL UPDATE 可逆HTTP POST 不可逆失败时触发补偿逻辑。我们用 Pydantic V2 定义 Execution Graph Schema配合 networkx 构建 DAG。实测发现带 Validation Hook 的 Planning 模块使下游 Action 执行失败率下降 68%因为 92% 的失败源于输入非法而非工具故障。2.5 Action动作不是调用函数而是带熔断、带降级、带审计的受控执行单元Action 是 Agent 的“手和脚”但绝不能裸调工具。每个 Action 必须封装为Circuit Breaker基于滑动窗口统计失败率默认 5 秒内 50% 失败则熔断Fallback Strategy熔断时启用备用方案如主库失败切从库API 失败查本地缓存Audit Trail记录工具名、输入哈希、输出摘要、耗时、是否重试用于事后归因。以 dbx 数据库工具为例我们封装了DBXExecutor类内置连接池max50、查询超时3s、结果行数限制10000、SQL 注入检测基于 sqlparse AST 分析。线上数据显示未封装的裸调用在高并发下失败率达 18%而封装后稳定在 0.3% 以下。2.6 Observation观测不是接收字符串而是结构化解析、可信度加权、噪声过滤的感知层LLM 返回的 Observation 常含噪声格式错乱、字段缺失、幻觉内容。Observation 模块必须做三件事Schema Enforcement用 JSON Schema 强制校验缺失字段填 default非法类型转 stringConfidence Scoring对关键字段如金额、日期用小模型打分0~1低于阈值0.7标记为 low_confidenceNoise Filtering移除无关描述如“根据以上分析我认为…”只保留结构化数据。我们训练了一个轻量级 RoBERTa 模型12MB专用于 Confidence Scoring在金融报表场景准确率达 94.2%。实测表明加入此模块后下游 Reflection 模块的决策质量提升 41%因为不再被低置信度噪声误导。2.7 Reflection反思不是重新提问而是基于证据链的因果推理与策略更新Reflection 是 Agent 的“复盘会议”但绝不能只让 LLM 自由发挥。它必须基于Evidence Chain本次 Goal 全过程的日志、State 快照、Observation 原始数据、Action 执行轨迹Causal Rules预设的失败归因规则如“SQL timeout 3s 且重试 3 次 → 优化索引”Strategy Update生成可执行的改进项如“下次对 sales_data 表加 composite index on (date, region)”。我们用 Neo4j 存储 Evidence Chain用 Cypher 查询构建因果图。一次典型反思耗时 2.3s但生成的索引优化建议在 3 天内将相关查询 P95 延迟从 4.2s 降至 0.8s。这证明 Reflection 不是锦上添花而是性能优化的核心引擎。3. 七个决策点每个都是线上事故的潜在入口也是稳定性保障的关键闸门七要素是静态模块而七个决策点是动态控制流中的关键关卡。它们决定了 Agent 是“玩具”还是“生产系统”。我按执行顺序列出每个点都附真实配置和避坑心得。3.1 决策点一Goal 接入时的语义归一化与 SLA 协商用户输入到达后第一道关卡不是交给 LLM而是做语义归一化。例如“查昨天销售额”“看下 2024-06-15 的营收”“给我昨天的 money”必须统一为{metric: revenue, date: 2024-06-15, unit: CNY}。我们用 3 层归一化正则清洗提取数字、日期、单位正则库覆盖 92% 常见表达同义词映射维护 business_term.yaml如“销售额”→“revenue”“money”→“amount”LLM 辅助校验仅对剩余 5% 模糊 case 调用轻量 LLMQwen-1.5B-int4prompt 严格限定输出 JSON。SLA 协商指根据 Goal 复杂度动态分配资源。简单 Goal查单表分配 1 个 CPU 核复杂 Goal多源关联分析分配 4 核 GPU。我们用 Kubernetes Horizontal Pod AutoscalerHPA基于 custom metricgoal_complexity_score扩缩容避免资源浪费。实操心得曾因未做语义归一化导致“上个月”在 1 月被解析为 2023-12而在 2 月被解析为 2024-01引发财务数据错乱。现在所有日期解析强制走dateutil.relativedelta并记录原始输入与解析结果 diff。3.2 决策点二State 初始化时的租户隔离与资源预占State 初始化不是 new State()而是租户路由从 JWT token 解析 tenant_id选择对应 Redis Cluster避免跨租户污染资源预占向资源中心申请 CPU/内存 quota如 200m CPU, 512Mi 内存失败则返回 429快照备份生成初始 State 快照存至 S3key:tenant/{id}/state/{goal_id}/init.json用于灾备。我们用 Redis 的WATCH/MULTI/EXEC保证 State 初始化原子性。压测发现当并发初始化请求达 5000 QPS 时裸 Redis 的SET操作失败率飙升至 12%而加 WATCH 后稳定在 0.02%。3.3 决策点三Memory 检索时的分层路由与隐私脱敏Memory 检索必须分层先查 Short-termState 内再查 Medium-termWeaviatequery filter:tenant_id {current}最后查 Long-termPostgreSQLrole-based WHERE clause。关键是在 Medium-term 检索结果返回前做隐私脱敏对身份证号、手机号、银行卡号等 PII 字段用 AES 加密后返回 token如enc_abc123前端需二次鉴权才可解密。我们用 OpenTelemetry 自动注入脱敏 span确保每条 Memory 访问可审计。注意Weaviate 的 vector search 默认不支持租户 filter需在 schema 中显式添加tenant_id字段并建索引否则检索会跨租户泄露数据。3.4 决策点四Planning 生成时的 DAG 合法性校验与循环检测Planning 模块输出 Execution Graph 后必须做两重校验DAG 合法性用 networkx 检测 cyclenx.is_directed_acyclic_graph(graph)存在环则 reject节点约束检查每个节点 timeout_ms ≤ 30000retry_count ≤ 3input_size ≤ 1MB。我们曾因未检测循环导致 Agent 在“查数据→生成图表→分析图表→查数据…”中陷入无限循环CPU 100% 持续 47 分钟。现在所有 Planning 输出必过DAGValidator校验耗时 5ms。3.5 决策点五Action 执行时的熔断器状态检查与降级路由Action 执行前必须实时检查 Circuit Breaker 状态如果 OPEN则跳过主路径直奔 Fallback如果 HALF_OPEN则放行 10% 请求探针其余走 Fallback如果 CLOSED则正常执行。Fallback 路由需预注册dbx 工具的 fallback 是本地 SQLite 缓存http 工具的 fallback 是 mock response。我们用 Resilience4j 实现熔断器状态变更事件推送到 Grafana运维可实时查看各工具熔断率。3.6 决策点六Observation 解析时的 Schema 严格模式与置信度阈值拦截Observation 解析采用 strict mode字段缺失 → 填 default非 null 字段抛 exception类型错误 → 转 string 并 log warning置信度 0.7 → 标记low_confidence: true下游 Reflection 模块优先处理。我们定义了 127 个核心 Observation Schema如sales_report_v1.json全部存于 GitOps 仓库CI 流程自动校验变更。Schema 版本与 Agent 版本绑定避免前后端不一致。3.7 决策点七Reflection 触发时的证据链完整性验证与策略持久化Reflection 不是每次执行后都触发而是满足条件才启动Action 失败 ≥ 2 次Observation 置信度 0.5 的字段 ≥ 3 个Goal 执行时间 SLA × 2。触发前必须验证 Evidence Chain 完整性检查 State 快照、Action 日志、Observation 原始数据是否全部存在。缺失任一环节则跳过 Reflection避免基于残缺信息错误归因。生成的策略更新如“加索引”存入 PostgreSQL 的strategy_log表并标记status: pending。DBA 每日 review批准后由 Ansible 自动执行 DDL。这确保了 Reflection 的产出真正落地而非纸上谈兵。4. 实操用 Python FastAPI 搭建一个可压测的 Agent 核心骨架下面是一个精简但完整的 Agent 核心骨架聚焦七个决策点的工程实现。它不是玩具 demo而是我们线上服务的最小可行原型MVP已通过 2000 QPS 压测。4.1 项目结构与依赖说明agent-core/ ├── main.py # FastAPI 入口 ├── models/ │ ├── goal.py # Goal Schema 定义 │ ├── state.py # State 快照模型 │ └── execution.py # Execution Graph 模型 ├── services/ │ ├── goal_service.py # Goal 归一化与 SLA 协商 │ ├── state_service.py # State 初始化与租户路由 │ ├── memory_service.py # 分层 Memory 检索与脱敏 │ ├── planning_service.py# DAG 生成与循环检测 │ ├── action_service.py # Action 执行与熔断 │ ├── observation_service.py # Observation 解析与置信度 │ └── reflection_service.py # Reflection 触发与策略持久化 ├── utils/ │ ├── circuit_breaker.py # Resilience4j 封装 │ └── audit_logger.py # 结构化审计日志 └── config/ └── settings.py # 环境配置Redis URL, Weaviate host 等关键依赖requirements.txtfastapi0.115.0 redis5.0.7 weaviate-client4.24.0 networkx3.3 resilience4j0.2.0 pydantic2.8.2 sqlalchemy2.0.34提示不要用 LangChain 的AgentExecutor它把七要素耦合太紧无法单独替换 Planning 或 Observation 模块。我们坚持“每个要素一个 service”便于单元测试和灰度发布。4.2 Goal 接入与语义归一化决策点一services/goal_service.py核心代码from datetime import datetime, timedelta import re from pydantic import BaseModel, Field from typing import Optional, Dict, Any from utils.audit_logger import audit_log class GoalInput(BaseModel): raw_text: str user_id: str tenant_id: str class NormalizedGoal(BaseModel): metric: str Field(..., description指标名如 revenue, cost) date_range: Dict[str, str] Field(..., description{start: 2024-01-01, end: 2024-01-31}) unit: str CNY complexity_score: int Field(default1, description1-5影响资源分配) def normalize_goal(input: GoalInput) - NormalizedGoal: # Step 1: 正则提取 date_match re.search(r(上|本|去)个月|(昨天|今日)|(\d{4}-\d{2}-\d{2}), input.raw_text) if date_match: if 上个月 in input.raw_text: now datetime.now() start (now.replace(day1) - timedelta(days1)).replace(day1) end now.replace(day1) - timedelta(days1) elif 昨天 in input.raw_text: end datetime.now().date() - timedelta(days1) start end else: # 处理具体日期 pass # Step 2: 同义词映射简化版 term_map {销售额: revenue, 营收: revenue, 利润: profit} metric revenue for k, v in term_map.items(): if k in input.raw_text: metric v break # Step 3: SLA 协商 - 基于关键词复杂度打分 complexity_score 1 if 关联 in input.raw_text or 多表 in input.raw_text: complexity_score 4 if 预测 in input.raw_text or 机器学习 in input.raw_text: complexity_score 5 audit_log(goal_normalized, { user_id: input.user_id, raw: input.raw_text, normalized: {metric: metric, date_range: {start: str(start), end: str(end)}} }) return NormalizedGoal( metricmetric, date_range{start: str(start), end: str(end)}, complexity_scorecomplexity_score )这个函数做了三件事日期归一化避免“上个月”歧义、指标映射业务术语标准化、复杂度打分驱动资源调度。audit_log 记录原始输入与归一化结果用于后续审计。4.3 State 初始化与租户隔离决策点二services/state_service.pyimport redis from models.state import State from config.settings import settings from utils.audit_logger import audit_log class StateService: def __init__(self): self.redis_client redis.Redis( hostsettings.REDIS_HOST, portsettings.REDIS_PORT, db0, decode_responsesTrue ) def init_state(self, goal_id: str, tenant_id: str, normalized_goal: dict) - State: # 租户路由为每个 tenant_id 创建独立 key namespace state_key ftenant:{tenant_id}:state:{goal_id} # 资源预占调用资源中心 API此处简化为 Redis 计数 quota_key ftenant:{tenant_id}:quota:cpu if self.redis_client.incr(quota_key) settings.MAX_CPU_QUOTA_PER_TENANT: self.redis_client.decr(quota_key) # 回滚 raise Exception(Resource quota exceeded) # 生成初始 State initial_state State( goal_idgoal_id, tenant_idtenant_id, statusinitialized, created_atdatetime.utcnow().isoformat(), goalnormalized_goal, history[] ) # 原子性写入 Redis pipe self.redis_client.pipeline() pipe.setex(state_key, 300, initial_state.model_dump_json()) # TTL300s pipe.setex(f{state_key}:version, 300, 1) # 版本号 pipe.execute() audit_log(state_initialized, { goal_id: goal_id, tenant_id: tenant_id, state_key: state_key }) return initial_state这里的关键是tenant:{tenant_id}:state:{goal_id}的 key 设计确保租户数据物理隔离。quota_key用于粗粒度资源控制避免单租户耗尽集群资源。4.4 Planning 生成与循环检测决策点四services/planning_service.pyimport networkx as nx from models.execution import ExecutionGraph, Node, Edge from utils.circuit_breaker import CircuitBreaker class PlanningService: def generate_plan(self, goal: dict) - ExecutionGraph: # 基于规则生成初始 DAG简化版 nodes [] edges [] # 示例销售分析 Goal 固定流程 if goal[metric] revenue: nodes [ Node(idsql_query, typesql, tooldbx, input_schema{table: sales, where: date between ? and ?}, timeout_ms3000, retry_count2), Node(idllm_summarize, typellm, toolqwen, input_schema{text: str}, timeout_ms5000, retry_count1) ] edges [Edge(sourcesql_query, targetllm_summarize)] graph ExecutionGraph(nodesnodes, edgesedges) # 循环检测 if not self._is_dag(graph): raise ValueError(Execution graph contains cycle) # 合法性校验 self._validate_nodes(graph.nodes) return graph def _is_dag(self, graph: ExecutionGraph) - bool: # 构建 networkx 图 G nx.DiGraph() for node in graph.nodes: G.add_node(node.id) for edge in graph.edges: G.add_edge(edge.source, edge.target) return nx.is_directed_acyclic_graph(G) def _validate_nodes(self, nodes: list): for node in nodes: if node.timeout_ms 30000: raise ValueError(fNode {node.id} timeout too long: {node.timeout_ms}) if node.retry_count 3: raise ValueError(fNode {node.id} retry count too high: {node.retry_count})_is_dag方法用 networkx 检测循环这是防止无限执行的底线。_validate_nodes强制约束 timeout 和 retry避免雪崩。4.5 Action 执行与熔断决策点五services/action_service.pyfrom utils.circuit_breaker import CircuitBreaker from typing import Dict, Any class ActionService: def __init__(self): # 为每个工具初始化熔断器 self.cbs { dbx: CircuitBreaker( failure_threshold5, timeout_duration60, wait_duration30 ), http: CircuitBreaker( failure_threshold3, timeout_duration30, wait_duration15 ) } def execute_action(self, node: Node, state: State) - Dict[str, Any]: # 检查熔断器状态 cb self.cbs.get(node.tool, None) if cb and cb.state OPEN: return self._fallback(node, state) try: if node.type sql: result self._execute_dbx(node.input_schema, state) elif node.type http: result self._execute_http(node.input_schema, state) else: result self._execute_llm(node.input_schema, state) # 成功则记录供熔断器统计 if cb: cb.record_success() return {success: True, data: result} except Exception as e: if cb: cb.record_failure() # 记录错误但不抛出让下游处理 return {success: False, error: str(e), node_id: node.id} def _fallback(self, node: Node, state: State) - Dict[str, Any]: # dbx 工具的 fallback查本地 SQLite 缓存 if node.tool dbx: return {success: True, data: self._query_sqlite_cache(node.input_schema)} # 其他工具返回 mock return {success: True, data: {mock: True}}CircuitBreaker是 Resilience4j 的 Python 封装_fallback提供降级路径。注意execute_action不抛异常而是返回结构化 error让上游统一处理。4.6 Observation 解析与置信度决策点六services/observation_service.pyimport json from pydantic import ValidationError from models.execution import Node from typing import Dict, Any class ObservationService: def parse_observation(self, node: Node, raw_output: str) - Dict[str, Any]: # Step 1: Schema 强制校验 try: parsed json.loads(raw_output) # 使用 Pydantic 模型校验此处简化为 dict key 检查 required_keys node.input_schema.get(required, []) for key in required_keys: if key not in parsed: raise ValidationError(fMissing required key: {key}) except json.JSONDecodeError as e: return {error: Invalid JSON, raw: raw_output} except ValidationError as e: return {error: str(e), raw: raw_output} # Step 2: 置信度打分简化版基于字段完整性 confidence 1.0 if len(parsed) len(required_keys) * 0.8: confidence 0.5 # Step 3: 噪声过滤 - 移除非结构化描述 clean_data {k: v for k, v in parsed.items() if not isinstance(v, str) or len(v) 200} return { success: True, data: clean_data, confidence: confidence, low_confidence: confidence 0.7 } # 示例对 sales_report 的置信度规则 def score_sales_report(obs: dict) - float: score 1.0 if total_revenue not in obs or not isinstance(obs[total_revenue], (int, float)): score - 0.3 if date_range not in obs or start not in obs[date_range]: score - 0.2 return max(0.0, score)parse_observation三步走JSON 解析与 Schema 校验、置信度计算、噪声过滤。score_sales_report是领域专用置信度函数可根据业务定制。5. 常见问题与排查技巧实录来自 7 个线上项目的血泪经验这些不是教科书问题而是我在凌晨三点排查线上故障时记下的真实案例。每个问题都附带根因、排查命令和修复方案。5.1 问题一Goal 解析后日期错乱导致财务报表数据偏差现象用户输入“上个月销售额”在 2024-01-05 返回 2023-12 数据但在 2024-02-01 返回 2024-01 数据部分用户投诉数据不一致。根因日期解析逻辑未考虑月份天数差异datetime.now().replace(monthnow.month-1)在 1 月会变成 0 月Python 抛异常后 fallback 到错误逻辑。排查命令# 查看 Goal 归一化日志 kubectl logs -l appagent-core | grep goal_normalized | grep 2024-01 # 检查时区配置 kubectl exec -it deploy/agent-core -- date修复方案弃用replace(month...)改用dateutil.relativedeltafrom dateutil.relativedelta import relativedelta last_month datetime.now().date() - relativedelta(months1) # 自动处理 1 月减 1 月 2023-12-01经验所有日期/时间操作必须用dateutil原生 datetime 的 month/year 运算有陷阱。5.2 问题二State 初始化失败率突增Redis 连接池耗尽现象State 初始化接口 503 错误率从 0.01% 升至 15%持续 12 分钟。根因Redis 连接池最大连接数设为 100但高峰期并发 Goal 初始化达 120新请求阻塞超时。排查命令# 查看 Redis 连接数 redis-cli -h $REDIS_HOST info | grep connected_clients # 查看应用连接池状态需暴露 metrics curl http://localhost:8000/metrics | grep redis_pool修复方案动态连接池max_connections min(200, cpu_count * 4)添加连接等待超时socket_timeout1000, socket_connect_timeout1000HPA 基于redis_pool_wait_time_ms指标扩容。经验Redis 连接池不是越大越好要匹配应用线程数和 Redis 实例规格。5.3 问题三Observation 解析失败LLM 返回的 JSON 格式错乱现象Action 执行成功但 Observation 解析报JSONDecodeError下游流程中断。根因LLM 在高负载时生成不合法 JSON如字段名漏引号、尾部逗号而解析器未做容错。排查命令# 抓取原始 Observation 输出 kubectl logs -l appagent-core | grep action_executed | tail -20 # 用 jq 验证 JSON 合法性 echo {total: 100,} | jq empty # 会报错修复方案用json5库替代json支持注释、尾逗号添加 JSON 修复逻辑jsonrepair包自动修正设置 fallback解析失败时返回{error: json_parse_failed, raw: ...}。经验永远假设 LLM 输出是不可信的Observation 解析层必须比前端更健壮。5.4 问题四Reflection 未触发复杂 Goal 失败后无复盘现象某 Goal 连续失败 5 次但 Reflection 日志为空。根因Reflection 触发条件Action 失败 ≥ 2 次未考虑重试。实际是 1 次 Action 失败 2 次重