企业官网建设流程全解析

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 存于 Redis（TTL=300s），Change Log 存于 Kafka（保留 7 天）。当发生冲突时，以 Log 中最新 commit_id 为准。压测显示，单节点每秒可处理 1200 次 State 更新，而纯内存方案在 200 QPS 时就开始丢状态。

2.3 Memory：记忆不是缓存，而是分层、分权、分粒度的持久化知识网络

Agent 的 Memory 分三层：

• Short-term（对话级）：存于 State 中，生命周期与 Goal 绑定；
• Medium-term（用户级）：存于用户专属向量库（Weaviate），含偏好、历史成功动作、常用工具集，TTL=90 天；
• Long-term（组织级）：存于关系型数据库（PostgreSQL），含 SOP 文档、权限策略、审计日志，永久保存。

关键设计是权限隔离：Medium-term Memory 对每个用户加密隔离（AES-256-GCM），Key 由用户 ID + 租户密钥派生；Long-term Memory 则按 RBAC 控制访问，避免 Agent 误用敏感 SOP。曾有案例：某 Agent 因未隔离 Medium-term Memory，将 A 用户的报销偏好应用到 B 用户，导致审批流程错乱。

2.4 Planning：规划不是生成一段文字，而是生成可验证、可中断、可回滚的执行树

LLM 输出的“先查数据库，再调 API，最后总结”只是伪规划。工程上，Planning 模块必须输出：

• Execution Graph：DAG 结构，节点含 type（sql / http / llm_call）、input_schema、timeout_ms、retry_policy；
• Validation Hook：每个节点执行前校验输入是否符合 schema，不符合则 abort 并返回 error_code；
• Rollback 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类，内置连接池（max=50）、查询超时（3s）、结果行数限制（10000）、SQL 注入检测（基于 sqlparse AST 分析）。线上数据显示，未封装的裸调用在高并发下失败率达 18%，而封装后稳定在 0.3% 以下。

2.6 Observation：观测不是接收字符串，而是结构化解析、可信度加权、噪声过滤的感知层

LLM 返回的 Observation 常含噪声：格式错乱、字段缺失、幻觉内容。Observation 模块必须做三件事：

• Schema Enforcement：用 JSON Schema 强制校验，缺失字段填 default，非法类型转 string；
• Confidence Scoring：对关键字段（如金额、日期）用小模型打分（0~1），低于阈值（0.7）标记为 low_confidence；
• Noise 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 层归一化：

1. 正则清洗：提取数字、日期、单位（正则库覆盖 92% 常见表达）；
2. 同义词映射：维护 business_term.yaml（如“销售额”→“revenue”，“money”→“amount”）；
3. LLM 辅助校验：仅对剩余 5% 模糊 case 调用轻量 LLM（Qwen-1.5B-int4），prompt 严格限定输出 JSON。

SLA 协商指：根据 Goal 复杂度动态分配资源。简单 Goal（查单表）分配 1 个 CPU 核，复杂 Goal（多源关联分析）分配 4 核 + GPU。我们用 Kubernetes Horizontal Pod Autoscaler（HPA）基于 custom metric（goal_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 快照存至 S3（key: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-term（State 内）；
• 再查 Medium-term（Weaviate，query filter:tenant_id == {current}）；
• 最后查 Long-term（PostgreSQL，role-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 检测 cycle（nx.is_directed_acyclic_graph(graph)），存在环则 reject；
• 节点约束检查：每个节点 timeout_ms ≤ 30000，retry_count ≤ 3，input_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.txt）：

fastapi==0.115.0 redis==5.0.7 weaviate-client==4.24.0 networkx==3.3 resilience4j==0.2.0 pydantic==2.8.2 sqlalchemy==2.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(default=1, description="1-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(day=1) - timedelta(days=1)).replace(day=1) end = now.replace(day=1) - timedelta(days=1) elif '昨天' in input.raw_text: end = datetime.now().date() - timedelta(days=1) 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( metric=metric, date_range={"start": str(start), "end": str(end)}, complexity_score=complexity_score )

这个函数做了三件事：日期归一化（避免“上个月”歧义）、指标映射（业务术语标准化）、复杂度打分（驱动资源调度）。audit_log 记录原始输入与归一化结果，用于后续审计。

4.3 State 初始化与租户隔离（决策点二）

services/state_service.py：

import 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( host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=0, decode_responses=True ) def init_state(self, goal_id: str, tenant_id: str, normalized_goal: dict) -> State: # 租户路由：为每个 tenant_id 创建独立 key namespace state_key = f"tenant:{tenant_id}:state:{goal_id}" # 资源预占：调用资源中心 API（此处简化为 Redis 计数） quota_key = f"tenant:{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_id=goal_id, tenant_id=tenant_id, status="initialized", created_at=datetime.utcnow().isoformat(), goal=normalized_goal, history=[] ) # 原子性写入 Redis pipe = self.redis_client.pipeline() pipe.setex(state_key, 300, initial_state.model_dump_json()) # TTL=300s 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.py：

import 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(id="sql_query", type="sql", tool="dbx", input_schema={"table": "sales", "where": "date between ? and ?"}, timeout_ms=3000, retry_count=2), Node(id="llm_summarize", type="llm", tool="qwen", input_schema={"text": "str"}, timeout_ms=5000, retry_count=1) ] edges = [Edge(source="sql_query", target="llm_summarize")] graph = ExecutionGraph(nodes=nodes, edges=edges) # 循环检测 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(f"Node {node.id} timeout too long: {node.timeout_ms}") if node.retry_count > 3: raise ValueError(f"Node {node.id} retry count too high: {node.retry_count}")

_is_dag方法用 networkx 检测循环，这是防止无限执行的底线。_validate_nodes强制约束 timeout 和 retry，避免雪崩。

4.5 Action 执行与熔断（决策点五）

services/action_service.py：

from utils.circuit_breaker import CircuitBreaker from typing import Dict, Any class ActionService: def __init__(self): # 为每个工具初始化熔断器 self.cbs = { "dbx": CircuitBreaker( failure_threshold=5, timeout_duration=60, wait_duration=30 ), "http": CircuitBreaker( failure_threshold=3, timeout_duration=30, wait_duration=15 ) } 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.py：

import 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(f"Missing 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(month=now.month-1)在 1 月会变成 0 月，Python 抛异常后 fallback 到错误逻辑。
• 排查命令：

  # 查看 Goal 归一化日志 kubectl logs -l app=agent-core | grep "goal_normalized" | grep "2024-01" # 检查时区配置 kubectl exec -it deploy/agent-core -- date

• 修复方案：弃用replace(month=...)，改用dateutil.relativedelta：

  from dateutil.relativedelta import relativedelta last_month = datetime.now().date() - relativedelta(months=1) # 自动处理 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

• 修复方案：
  1. 动态连接池：max_connections = min(200, cpu_count * 4)；
  2. 添加连接等待超时：socket_timeout=1000, socket_connect_timeout=1000；
  3. HPA 基于redis_pool_wait_time_ms指标扩容。
• 经验：Redis 连接池不是越大越好，要匹配应用线程数和 Redis 实例规格。

5.3 问题三：Observation 解析失败，LLM 返回的 JSON 格式错乱

• 现象：Action 执行成功，但 Observation 解析报JSONDecodeError，下游流程中断。
• 根因：LLM 在高负载时生成不合法 JSON（如字段名漏引号、尾部逗号），而解析器未做容错。
• 排查命令：

  # 抓取原始 Observation 输出 kubectl logs -l app=agent-core | grep "action_executed" | tail -20 # 用 jq 验证 JSON 合法性 echo '{"total": 100,}' | jq empty # 会报错

• 修复方案：
  1. 用json5库替代json（支持注释、尾逗号）；
  2. 添加 JSON 修复逻辑：jsonrepair包自动修正；
  3. 设置 fallback：解析失败时返回{"error": "json_parse_failed", "raw": ...}。
• 经验：永远假设 LLM 输出是不可信的，Observation 解析层必须比前端更健壮。

5.4 问题四：Reflection 未触发，复杂 Goal 失败后无复盘

• 现象：某 Goal 连续失败 5 次，但 Reflection 日志为空。
• 根因：Reflection 触发条件Action 失败 ≥ 2 次未考虑重试。实际是 1 次 Action 失败 + 2 次重