AI Agent七要素工程实践：从闭环机制到生产级落地

1. 什么是 AI Agent？它不是“更聪明的聊天机器人”，而是能闭环做事的数字员工

很多人第一次听说 AI Agent，脑子里蹦出来的可能是“会自己调用工具的 ChatGPT”——这不算错，但严重低估了它的工程分量。我带团队落地过 17 个生产级 Agent 项目，从金融风控审批流、制造业设备巡检调度，到政务热线智能分派系统，最深的体会是：Agent 的本质不是模型能力的延伸，而是软件工程范式的迁移。它把传统“人写逻辑 → 程序执行 → 人看结果”的线性流程，变成了“目标输入 → 自主规划 → 工具调用 → 状态反馈 → 动态修正”的闭环控制回路。这个回路里，LLM 不是大脑，而是决策中枢里的“认知协处理器”；真正驱动业务运转的，是那七个被反复验证、缺一不可的工程要素。

你刷到的“扣子开发 AI Agent 智能体应用”“FastAPI + LangChain + LangGraph 搭建智慧 Agent”，背后全是这七要素的排列组合。而所谓“七个决策点”，就是工程师在真实场景中必须亲手拍板的关键路口：比如该用 ReAct 还是 Plan-and-Execute？工具发现该走静态注册还是动态反射？记忆该存向量库还是关系型数据库？这些选择没有标准答案，但每个都直接决定系统能否扛住每秒 200+ 并发请求、能否在 3 秒内完成跨 5 个系统（CRM + ERP + 物流 API + 支付网关 + 短信平台）的协同操作。我见过太多团队卡在“Agent 搭建”阶段，不是因为不会写 prompt，而是没想清楚：当用户说“帮我订下周二去上海的高铁票并同步到日历”，这个指令背后要拆解多少层状态机？工具链如何保证支付失败时自动回滚订票动作？沙盒环境怎么隔离不同用户的会话上下文？这些才是让 Agent 从 Demo 走进产线的核心战场。

关键词“AI Agent”“Agent”“LLM”“工具”“循环机制”绝非泛泛而谈——它们精准指向了工程实现的五个硬骨头：语义理解的鲁棒性、工具调用的原子性、状态管理的确定性、循环控制的收敛性、安全边界的可审计性。接下来我会用真实项目中的配置片段、压测数据、错误日志和调试截图，一层层剥开这七个要素如何落地，以及那七个决策点为什么必须由人来定，而不是交给框架自动选。

2. 七要素解构：每个要素都是生产环境里的“生死线”

2.1 目标定义（Goal Definition）：不是写 prompt，而是建模业务契约

目标定义常被简化为“给 LLM 一个清晰指令”，这是最大的认知陷阱。在我们为某银行搭建的信贷审批 Agent 中，目标不是“审核这笔贷款申请”，而是“在 90 秒内，依据《2024 年小微贷风控白皮书》第 3.2 条，完成客户资质校验、抵押物估值、同业征信交叉验证，并生成符合银保监格式要求的审批意见书”。这个目标包含三个刚性约束：时效性（90s）、合规性（白皮书条款）、输出规范性（监管格式）。

实操中，我们用 YAML 定义目标契约：

goal: id: "credit_approval_v2" description: "Automated small business loan approval with regulatory compliance" constraints: - timeout: 90s - compliance: "CBIRC_2024_Q3" - output_format: "XML_schema_v1.2" success_criteria: - "approval_decision in ['APPROVE', 'REJECT', 'PENDING']" - "audit_trail_complete: true"

提示：目标定义必须可验证。我们曾因漏掉audit_trail_complete校验，在上线第三天被风控部叫停——Agent 虽然返回了审批结果，但未记录所有调用的外部 API 响应码，无法满足审计要求。

2.2 规划器（Planner）：LLM 是“参谋”，不是“指挥官”

规划器负责把目标拆解为可执行步骤序列。常见误区是让 LLM 直接输出 JSON 步骤，但在高并发下，LLM 的 token 生成存在概率性抖动。我们采用双阶段规划：第一阶段用轻量级规则引擎（如 Drools）做硬约束预筛，第二阶段才交由 LLM 做柔性决策。

以“期货交易 Agent”为例（注意：仅用于模拟盘，不涉及实盘交易），规划器需处理：

• 硬约束：交易所交易时段（9:00-11:30, 13:30-15:00）、单笔保证金上限（≤账户余额 20%）
• 软约束：根据 LLM 分析的新闻情绪值，动态调整下单量（情绪值 >0.8 时加仓 10%）

我们的规划器配置如下：

# 第一阶段：规则引擎预筛 rules = [ Rule("trading_hours", "now in [9:00, 11:30] or [13:30, 15:00]"), Rule("margin_check", "order_value <= balance * 0.2") ] # 第二阶段：LLM 规划（输入已过滤的候选动作） planning_prompt = """ 你是一个期货交易助手，请基于以下信息生成操作步骤： - 当前时间：{time} - 账户余额：{balance} - 待分析新闻：{news_summary} - 可用合约：[IC2406, IF2406, IM2406] 请严格按 JSON 格式输出，字段：steps:[{action:"buy/sell", contract:"IC2406", quantity:int, price_type:"market/limit"}] """

注意：LLM 输出必须经 schema 校验。我们曾因 LLM 返回"quantity": "ten"（字符串）而非整数，导致下游交易接口崩溃。现在所有 LLM 输出都通过 Pydantic 模型强制转换，失败则触发降级策略（返回默认仓位）。

2.3 工具集成（Tool Integration）：不是“能调用 API”，而是“能兜住失败”

工具集成是 Agent 最易崩塌的环节。“引入工具类”热搜词背后，是无数团队在 HTTP 超时、认证失效、响应格式变更上的血泪史。我们的标准是：每个工具必须自带熔断、重试、降级三件套。

以对接 DBX 数据库工具为例（非 Databricks，指某国产分布式数据库客户端）：

class DBXTool: def __init__(self): self.client = DBXClient( host="dbx-prod.internal", port=8080, # 熔断器：连续3次失败后，10秒内拒绝新请求 circuit_breaker=CircuitBreaker(failure_threshold=3, reset_timeout=10) ) def execute_query(self, sql: str) -> dict: try: # 重试：指数退避，最多3次 for attempt in range(3): try: result = self.client.query(sql, timeout=5) return {"status": "success", "data": result} except TimeoutError: if attempt == 2: raise time.sleep(2 ** attempt) # 1s, 2s, 4s except Exception as e: # 降级：返回空结果或缓存快照 return {"status": "fallback", "data": self._get_cache_snapshot()}

实操心得：工具封装必须隔离 LLM 的“想象空间”。我们禁止在 prompt 中暴露工具内部细节（如“DBX 支持 LIMIT 子句”），所有 SQL 生成由 LLM 完成，工具层只负责执行与兜底。这样即使 DBX 升级导致语法变更，只需改工具层，不影响上层规划逻辑。

2.4 记忆管理（Memory Management）：不是“记住对话”，而是“维护状态一致性”

记忆常被等同于聊天历史，但在生产 Agent 中，它是多维度状态枢纽。我们为政务热线 Agent 设计了三级记忆：

• 短期记忆（Session Memory）：Redis Hash 存储单次会话的上下文（TTL=30min），键为session:{user_id}:{call_id}
• 长期记忆（Knowledge Memory）：向量库（Weaviate）存储政策法规文档，支持语义检索
• 事务记忆（Transaction Memory）：PostgreSQL 表记录每个审批流程的状态机（state: PENDING → VALIDATING → APPROVING → COMPLETED）

关键设计点在于状态同步。当 Agent 调用 CRM 更新客户信息后，必须立即更新事务记忆表，否则下次查询可能读到脏数据。我们采用“事件驱动 + 本地缓存”模式：

# CRM 更新成功后触发 def on_crm_update_success(event): # 1. 更新事务记忆（强一致性） db.execute("UPDATE transaction_memory SET crm_status='UPDATED' WHERE id=?", event.tx_id) # 2. 清除相关会话的 Redis 缓存 redis.delete(f"session:{event.user_id}:{event.call_id}") # 3. 发布事件供其他服务监听 kafka_produce("crm_update_event", {"tx_id": event.tx_id, "timestamp": now()})

踩过的坑：早期用纯 Redis 存储事务状态，遭遇网络分区时出现状态不一致。现在所有关键状态变更都走数据库事务，Redis 仅作读缓存。

2.5 执行器（Executor）：不是“跑代码”，而是“控资源边界”

执行器负责将规划步骤转化为实际动作。难点在于资源隔离与超时控制。我们用 Docker 容器化每个工具调用：

# tools/python-executor/Dockerfile FROM python:3.11-slim COPY requirements.txt . RUN pip install -r requirements.txt CMD ["python", "executor.py"]

执行器启动时接收 JSON 任务：

{ "tool": "dbx_query", "params": {"sql": "SELECT * FROM customers WHERE id=123"}, "timeout": 8, "memory_limit_mb": 256 }

容器启动参数强制限制：

docker run --memory=256m --cpus=0.5 --network=host \ -v /tmp:/tmp \ -e TOOL_CONFIG=/config/dbx.yaml \ tool-executor:latest

关键经验：必须禁用--privileged和--network=host外的网络模式。我们曾因允许容器访问宿主机网络，导致工具脚本意外调用内网监控 API，触发安全告警。

2.6 观察器（Observer）：不是“看返回值”，而是“验业务语义”

观察器解析工具执行结果，但重点不是 JSON 解析，而是业务有效性校验。例如调用短信平台发送验证码，HTTP 200 不代表成功——需检查响应体中的code==0和message=="success"。

我们为每个工具定义观察规则：

OBSERVATION_RULES = { "sms_send": { "http_status": 200, "json_path": "$.code", "expected_value": 0, "retry_on_failure": True, # 失败时重试（非幂等操作需谨慎） "fallback_action": "log_and_notify" # 降级动作 } }

注意：观察器必须区分技术失败与业务失败。技术失败（如网络超时）可重试；业务失败（如短信余额不足）需触发人工干预流程，不能无脑重试。

2.7 循环控制器（Loop Controller）：不是“while True”，而是“有退出条件的状态机”

循环机制是 Agent 的灵魂，也是最易失控的部分。“AI Agent 怎么扛并发”问题本质是循环控制器的设计缺陷。我们采用有限状态机（FSM）+ 最大步数限制：

控制器核心逻辑：

def run_loop(self, goal: Goal): state = "INIT" step_count = 0 max_steps = 15 # 硬性限制，防无限循环 while state not in ["SUCCESS", "FAILURE", "TERMINATED"]: if step_count > max_steps: state = "TERMINATED" break try: if state == "INIT": state = self._plan(goal) elif state == "PLANNING": state = self._execute_plan() elif state == "EXECUTING": state = self._observe_results() step_count += 1 except Exception as e: self._handle_error(e, state) state = "FAILURE"

实测数据：在 500 QPS 压测下，99.99% 的请求在 7 步内完成，平均耗时 2.3s。未设最大步数时，0.1% 的请求因 LLM 生成无效循环（如反复查询同一数据库）导致超时。

3. 七个决策点：工程师必须亲手拍板的工程十字路口

3.1 决策点一：规划策略选型——ReAct 还是 Plan-and-Execute？

ReAct（Reasoning + Acting）让 LLM 在每步行动前生成推理链，适合探索性强的任务（如科研文献综述）；Plan-and-Execute 先生成完整步骤再执行，适合确定性高的流程（如订单履约）。我们在电商售后 Agent 中对比测试：

结论：Plan-and-Execute 更适合高确定性业务。我们最终采用混合模式：先用规则引擎生成主干步骤（Plan），再用 ReAct 处理分支决策（如“用户投诉是否升级至 VIP 通道”）。

3.2 决策点二：工具发现机制——静态注册还是动态反射？

静态注册（预定义工具列表）安全可控，但扩展成本高；动态反射（运行时扫描函数）灵活，但存在安全风险。我们为内部 DevOps Agent 选择动态反射，但加三重防护：

• 白名单装饰器：仅标记@tool_enabled的函数可被发现
• 参数类型校验：反射时强制检查函数签名是否匹配Callable[[dict], dict]
• 执行沙箱：所有反射调用在独立进程运行，超时即 kill

def tool_discovery(): tools = {} for name, func in inspect.getmembers(sys.modules[__name__], inspect.isfunction): if hasattr(func, '_tool_enabled'): # 类型校验 sig = inspect.signature(func) if len(sig.parameters) != 1 or list(sig.parameters.values())[0].annotation != dict: continue tools[name] = func return tools

3.3 决策点三：记忆持久化选型——向量库还是关系型数据库？

向量库（如 Weaviate）擅长语义检索，但不支持 ACID 事务；关系型数据库（如 PostgreSQL）保证强一致性，但语义搜索弱。我们的解法是分层存储：

• 会话上下文、临时变量 → Redis（高性能读写）
• 政策法规、知识文档 → 向量库（语义检索）
• 审批记录、交易流水 → PostgreSQL（事务保障）

关键技巧：用 PostgreSQL 的pgvector扩展支持向量相似度查询，避免跨库 join。例如查询“类似的历史审批案例”：

SELECT id, content, 1 - (embedding <=> '[0.1,0.9,...]') AS similarity FROM policy_cases WHERE 1 - (embedding <=> '[0.1,0.9,...]') > 0.7 ORDER BY similarity DESC LIMIT 3;

3.4 决策点四：循环退出条件——基于步数还是基于目标达成？

单纯限制步数（如max_steps=15）简单粗暴，但可能提前终止；依赖目标达成（如goal.is_satisfied()）精准，但判断逻辑复杂。我们采用双条件触发：

• 主条件：目标达成（调用goal.check_completion()）
• 次条件：步数超限或总耗时超限（step_count > 15 or total_time > 30s）

check_completion()方法包含业务逻辑：

def check_completion(self): # 检查输出文件是否存在且非空 if not os.path.exists(self.output_file): return False if os.path.getsize(self.output_file) == 0: return False # 检查输出格式是否符合 XSD if not self._validate_xsd(self.output_file): return False return True

3.5 决策点五：LLM 选型——通用大模型还是领域微调模型？

通用模型（如 Qwen2-72B）泛化能力强，但领域任务准确率低；微调模型（如金融风控专用 LLM）精度高，但泛化性差。我们在银行项目中采用两阶段 LLM 架构：

• 第一阶段：通用模型做初步规划（成本低，容错高）
• 第二阶段：微调模型做关键决策（如“是否触发反洗钱核查”）

微调数据来自真实审批日志，标注规则：

• 输入：客户资料 + 交易流水 + 当前政策
• 输出：{"decision": "APPROVE/REJECT/PENDING", "reason": "string", "compliance_ref": "CBIRC_2024_Q3_3.2"}

微调后，关键决策准确率从 78% 提升至 94%，误拒率下降 62%。

3.6 决策点六：安全边界设计——沙盒隔离还是 API 网关？

沙盒（如 Docker 容器）提供强隔离，但资源开销大；API 网关（如 Kong）轻量，但依赖服务端配合。我们为政务 Agent 选择沙盒 + 网关双保险：

• 工具调用：全部走 Docker 沙盒（隔离文件系统、网络、内存）
• 敏感操作（如修改用户权限）：额外通过 API 网关鉴权，网关校验 JWT 中的 RBAC 权限

沙盒启动时注入最小权限：

docker run --read-only \ --cap-drop=ALL \ --security-opt=no-new-privileges \ --tmpfs /tmp:rw,size=100m \ sandbox-executor

3.7 决策点七：可观测性方案——日志埋点还是 OpenTelemetry？

日志埋点简单，但难以关联跨服务调用；OpenTelemetry（OTel）提供全链路追踪，但部署复杂。我们采用轻量级 OTel + 关键日志增强：

• 使用 OpenTelemetry Python SDK 自动捕获 HTTP/gRPC 调用
• 在关键节点（规划开始、工具调用、状态变更）手动打点：

from opentelemetry import trace tracer = trace.get_tracer(__name__) with tracer.start_as_current_span("agent_planning") as span: span.set_attribute("goal_id", goal.id) span.set_attribute("llm_model", "qwen2-72b") plan = self.llm.invoke(plan_prompt)

• 日志结构化：所有日志输出 JSON，包含trace_id、span_id、agent_id、step_id字段，便于 ELK 关联分析。

4. 实操全流程：从零搭建一个政务热线 Agent（含完整配置）

4.1 环境准备与依赖安装

我们使用 Python 3.11 + Poetry 管理依赖，确保环境纯净：

# 初始化项目 poetry init -n poetry add langchain==0.1.16 langgraph==0.0.32 weaviate-client==4.4.3 psycopg2-binary==2.9.7 redis==4.6.12 poetry add --group dev pytest==7.4.3 black==23.10.1 # 创建配置目录 mkdir -p config/{llm,tools,memory}

关键依赖说明：

• langchain: 提供基础工具链和回调机制
• langgraph: 实现状态机驱动的循环控制（比原生 while 循环更可靠）
• weaviate-client: 连接向量库，存储政策法规
• psycopg2-binary: PostgreSQL 驱动，保障事务记忆
• redis: 高性能会话缓存

注意：避免使用langchain-community，其工具集成不稳定。我们所有工具都自行封装，确保可控性。

4.2 七要素代码骨架搭建

创建核心模块agent/core.py：

from typing import Dict, Any, Optional from dataclasses import dataclass import redis import psycopg2 @dataclass class AgentState: """Agent 全局状态，贯穿整个循环""" goal: str session_id: str current_step: int = 0 memory: Dict[str, Any] = None tools: Dict[str, callable] = None class GovernmentHotlineAgent: def __init__(self): # 初始化七要素组件 self.planner = Planner() self.tools = self._load_tools() self.memory = MemoryManager() self.executor = Executor() self.observer = Observer() self.loop_controller = LoopController() def _load_tools(self) -> Dict[str, callable]: """加载工具，此处仅展示框架，实际工具见 4.3""" return { "query_policy": self._query_policy_db, "update_case_status": self._update_case_status, "send_sms": self._send_sms } def run(self, user_input: str, session_id: str) -> Dict[str, Any]: """Agent 主入口，遵循七要素流程""" state = AgentState( goal=user_input, session_id=session_id, memory=self.memory.load_session(session_id) ) # 循环执行七要素 while not self.loop_controller.should_exit(state): state = self.planner.plan(state) state = self.executor.execute(state) state = self.observer.observe(state) self.memory.save_session(state.session_id, state.memory) return self._format_output(state)

4.3 工具实现：以“政策查询”为例

agent/tools/policy_tool.py：

import weaviate from weaviate.classes.query import MetadataQuery from config.llm import LLM_CLIENT class PolicyQueryTool: def __init__(self): self.client = weaviate.connect_to_local() self.collection = self.client.collections.get("PolicyDocument") def invoke(self, query: str) -> str: """ 输入：用户自然语言问题，如“残疾人补贴标准是多少？” 输出：最相关的政策条款原文 + 条款编号 """ # 1. LLM 提取关键词（提升检索精度） keywords = LLM_CLIENT.invoke( f"从以下问题中提取2-3个核心关键词，用逗号分隔：{query}" ).strip() # 2. 向量检索 response = self.collection.query.near_text( query=keywords, limit=3, return_metadata=MetadataQuery(distance=True) ) # 3. 格式化结果 results = [] for o in response.objects: results.append(f"【{o.properties['clause_number']}】{o.properties['content']}") return "\n".join(results) # 注册为可调用工具 policy_tool = PolicyQueryTool()

关键细节：我们不用 LLM 直接回答政策问题，而是让它做关键词提取——因为 LLM 对政策条文的记忆可能过时，而向量库中的文档是实时同步的。

4.4 记忆管理实现

agent/memory/memory_manager.py：

import redis import json import psycopg2 from datetime import datetime class MemoryManager: def __init__(self): self.redis_client = redis.Redis(host="localhost", port=6379, db=0) self.pg_conn = psycopg2.connect( "dbname=agent_db user=agent password=secret host=localhost" ) def load_session(self, session_id: str) -> Dict[str, Any]: """加载会话记忆""" # 优先读 Redis cache_key = f"session:{session_id}" cached = self.redis_client.get(cache_key) if cached: return json.loads(cached) # 回源读 PostgreSQL with self.pg_conn.cursor() as cur: cur.execute( "SELECT memory_data FROM session_memory WHERE session_id = %s AND expires_at > %s", (session_id, datetime.now()) ) row = cur.fetchone() if row: data = json.loads(row[0]) # 写入 Redis 缓存 self.redis_client.setex(cache_key, 1800, json.dumps(data)) return data return {} def save_session(self, session_id: str, memory: Dict[str, Any]): """保存会话记忆""" cache_key = f"session:{session_id}" self.redis_client.setex(cache_key, 1800, json.dumps(memory)) # 异步写入 PostgreSQL（避免阻塞） import threading threading.Thread( target=self._save_to_pg, args=(session_id, memory) ).start() def _save_to_pg(self, session_id: str, memory: Dict[str, Any]): with self.pg_conn.cursor() as cur: cur.execute( "INSERT INTO session_memory (session_id, memory_data, expires_at) VALUES (%s, %s, %s) ON CONFLICT (session_id) DO UPDATE SET memory_data = EXCLUDED.memory_data, expires_at = EXCLUDED.expires_at", (session_id, json.dumps(memory), datetime.now().replace(hour=23, minute=59, second=59)) ) self.pg_conn.commit()

4.5 循环控制器实现

agent/loop/controller.py：

from typing import Dict, Any from dataclasses import dataclass import time @dataclass class LoopState: step_count: int = 0 start_time: float = 0.0 max_steps: int = 15 max_time: float = 30.0 class LoopController: def __init__(self): self.state = LoopState() def should_exit(self, agent_state: Dict[str, Any]) -> bool: """判断是否退出循环""" # 条件1：目标达成 if self._is_goal_satisfied(agent_state): return True # 条件2：步数超限 self.state.step_count += 1 if self.state.step_count > self.state.max_steps: return True # 条件3：总耗时超限 if self.state.step_count == 1: self.state.start_time = time.time() if time.time() - self.state.start_time > self.state.max_time: return True return False def _is_goal_satisfied(self, state: Dict[str, Any]) -> bool: """业务目标达成判断""" # 检查输出是否包含必要字段 if "output" not in state or not state["output"].strip(): return False # 检查是否包含政策条款引用（政务 Agent 的硬性要求） if "【" not in state["output"] or "】" not in state["output"]: return False return True

4.6 启动与测试

main.py：

from agent.core import GovernmentHotlineAgent if __name__ == "__main__": agent = GovernmentHotlineAgent() # 模拟用户输入 result = agent.run( user_input="残疾人补贴标准是多少？", session_id="sess_abc123" ) print("Agent 输出：", result["output"]) print("耗时：", result["latency_ms"], "ms") print("调用工具：", result["used_tools"])

运行命令：

poetry run python main.py

预期输出：

Agent 输出： 【津政发〔2023〕15号 第二条】对持有《中华人民共和国残疾人证》的本市户籍残疾人，按月发放生活补贴，标准为每人每月200元。 耗时： 1240 ms 调用工具： ['query_policy']

5. 常见问题与排查技巧实录：来自 17 个项目的血泪总结

5.1 问题速查表

5.2 独家避坑技巧

技巧一：用“影子测试”验证工具变更当升级 DBX 工具版本时，不直接切流，而是：

• 启动影子 Agent，接收相同流量
• 将新旧工具输出做 diff 比对
• 仅当 diff 差异率 <0.1% 时才切流

# 影子测试核心逻辑 def shadow_test(old_result, new_result): # 结构一致性检查 if type(old_result) != type(new_result): return False # 业务字段一致性（忽略 timestamp 等动态字段） ignore_keys = ["timestamp", "request_id"] old_clean = {k:v for k,v in old_result.items() if k not in ignore_keys} new_clean = {k:v for k,v in new_result.items() if k not in ignore_keys} return old_clean == new_clean

技巧二：LLM 输出的“防幻觉校验”对 LLM 生成的 SQL，我们增加三层校验：

1. 语法校验：用sqlparse解析，确保无语法错误
2. 表名校验：比对INFORMATION_SCHEMA.TABLES，确认表存在
3. 列名校验：对每个 SELECT 字段，查询INFORMATION_SCHEMA.COLUMNS

def validate_sql(sql: str) -> bool: # 1. 语法解析 try: parsed = sqlparse.parse(sql)[0] except: return False # 2. 提取表名（简化版） tables = re.findall(r"FROM\s+(\w+)|JOIN\s+(\w+)", sql, re.I) for table in [t[0] or t[1] for t in tables]: if not _table_exists(table): return False return True

技巧三：内存泄漏的“三色标记法”Agent 长时间运行后内存飙升？我们用 Python 的tracemalloc定位：

import tracemalloc tracemalloc.start() # 运行 1000 次 Agent 调用 for i in range(1000): agent.run("测试输入", f"test_{i}") # 获取内存快照 snapshot = tracemalloc.take_snapshot() top_stats = snapshot.statistics('lineno') # 打印前10内存占用 for stat in top_stats[:10]: print(stat)

定位到weaviate-client的连接池未关闭后，我们强制在每次查询后调用client.close()。

5.3 性能调优实战数据

在政务热线 Agent 上，我们通过以下优化将 P99 延迟从 8.2s 降至 1.7s：