企业官网建设流程全解析

1. 这不是“造轮子”，是重新理解Agent的起点

我从2023年夏天开始写第一个能自主调用天气API、再把结果整理成周报发到钉钉的Agent，到现在手头维护着6个生产级Agent系统——覆盖电商客服意图路由、工业图纸参数校验、医疗报告结构化提取、法务合同条款比对、教育题库动态生成、以及内部IT工单自动分派。它们底层没有共用一行框架代码，但所有核心逻辑都跑在一个不到800行的Python模块里。这个模块，就是标题里说的“通用内核”。

你可能已经看过太多LangChain、LlamaIndex、Dify、CrewAI的教程，也试过用AutoGen搭个“会聊天的Agent”，但真正上线后，90%的问题根本不在LLM调用或Prompt工程上：而是当用户说“把上周三下午三点的订单导出成Excel发给财务”时，你的Agent卡在“周三”到底是日历上的哪一天；当它调用ERP接口失败，错误码是rpc error (-1): empty sid and service name，却连该重试几次、要不要降级到缓存数据、是否该通知运维都无从判断；更别说插件加载时路径错位、记忆冲突、工具权限越界、状态漂移这些连日志都难定位的幽灵问题。

这14个设计决策，不是理论推演出来的，是我在47次线上故障复盘、132次压测失败、287次用户投诉归因后，用删掉又重写的代码换来的。它们不教你怎么调用OpenAI API，而是告诉你：Agent不是“会调用工具的LLM”，而是一个有状态、有边界、有契约、有退路的软件实体。所谓“复利”，不是模型能力的指数增长，而是每做一个正确决策，后续所有Agent开发都能少踩一次同类坑——比如第3条“状态快照隔离机制”，直接让新项目接入时间从5天压缩到4小时；第7条“插件契约声明”，让第三方插件集成成功率从61%提升到98%；第12条“安全沙箱分级”，使合规审计周期缩短70%。

如果你正在用现成框架快速搭建Demo，这篇内容可能显得“过度设计”；但如果你的目标是交付一个能扛住双十一流量、经得起金融级审计、被业务方当作基础设施依赖的Agent系统，那这14个点，每一个都是绕不开的窄门。它们不分语言（Rust/Python/Go都适用）、不绑定框架（和LangChain可共存）、不依赖特定云服务——只关乎你如何定义一个Agent该“是什么”，而不是“能做什么”。

2. 14个设计决策：为什么必须在编码前就拍板

2.1 决策1：Agent生命周期必须显式建模，而非隐式依赖LLM上下文

几乎所有主流框架默认把Agent生命周期等同于一次LLM对话轮次：用户提问→LLM思考→调用工具→返回结果→结束。但真实业务中，一个Agent任务可能持续数小时（如“跟踪这批货从下单到签收的全链路”），中间经历多次人工介入、系统中断、状态回滚。我们曾用LangChain的ConversationBufferMemory支撑物流Agent，结果发现：当网络抖动导致某次工具调用超时，整个对话历史被截断，Agent丢失了已获取的运单号、承运商ID、预计到达时间三个关键状态，重启后只能重问用户“您的运单号是多少？”，用户体验崩坏。

我们的解法：在通用内核中强制定义AgentState对象，包含task_id（全局唯一）、status（RUNNING/PENDING/FAILED/COMPLETED）、created_at、updated_at、retry_count、last_step（最后执行的步骤名）等字段，并要求所有插件调用、记忆读写、日志记录都必须绑定此状态。状态持久化不依赖LLM上下文，而是直写Redis（带TTL）+本地SQLite（兜底）。

提示：task_id不是UUID，而是{business_domain}_{timestamp}_{hash_of_input}格式（如logistics_20240521142301_8a3f），便于业务方按领域快速筛选日志，也避免跨域状态污染。

2.2 决策2：拒绝“万能Agent”，每个Agent必须声明其能力边界（Capability Contract）

很多团队陷入误区：用一个Agent处理所有客服问题。结果当用户问“帮我查下账户余额”时，Agent调用支付网关；问“推荐新款手机”时，它又去查商品库；问“我的快递到哪了”，它再切到物流系统。表面看很智能，实则埋下三颗雷：

• 资源争抢：支付网关和物流API共用同一连接池，高并发时互相拖垮；
• 权限失控：客服Agent本不该有调用支付接口的权限，但代码里没做隔离；
• 调试地狱：一个500 Internal Error报错，你得翻遍所有插件源码才能定位是哪个环节崩了。

我们的解法：在Agent初始化时，必须传入capability_contract字典，明确声明它能调用哪些插件、能访问哪些数据源、能修改哪些状态字段。例如：

logistics_agent = Agent( capability_contract={ "plugins": ["track_shipment", "estimate_delivery"], "data_sources": ["tms_db", "carrier_api"], "writable_states": ["delivery_status", "estimated_time"] } )

内核在每次插件调用前校验契约，违反即抛CapabilityViolationError并记录审计日志。这个契约不是装饰，而是运行时强制检查——我们甚至用mypy插件在编译期做静态校验。

2.3 决策3：状态快照必须隔离，禁止跨任务共享内存

这是最隐蔽也最致命的坑。某次大促期间，客服Agent出现诡异现象：用户A咨询“订单123退款进度”，Agent返回“已打款”；用户B立刻问“订单456退款进度”，Agent竟也返回“已打款”，尽管B的订单还在审核中。排查三天才发现，两个任务共用了同一个dict对象存储临时状态，当任务A结束时清空了state["refund_status"]，但任务B的引用还指向同一内存地址。

我们的解法：通用内核中所有状态操作都基于copy.deepcopy()创建隔离副本，且规定：

• 插件函数签名必须为def plugin_func(state: dict, **kwargs) -> dict，严禁修改入参state；
• 内核在调用插件前自动深拷贝，调用后用新state覆盖旧状态；
• 对性能敏感场景（如高频OCR识别），提供state_view只读视图，但写操作仍走深拷贝流程。

注意：Python的deepcopy对嵌套datetime、Decimal等类型有坑，我们封装了safe_deepcopy()，对常见类型做特殊处理，实测比原生快3.2倍。

2.4 决策4：工具调用必须带超时熔断与降级策略，而非简单重试

框架默认的“重试3次”策略在生产环境毫无意义。我们曾遇到物流API因上游系统升级，连续5分钟返回503 Service Unavailable，LangChain配置的max_retries=3让Agent在3秒内发起9次请求，不仅没解决问题，反而触发对方限流，导致整个物流模块雪崩。

我们的解法：每个插件注册时必须声明timeout_sec、circuit_breaker_threshold（连续失败阈值）、fallback_func（降级函数）。例如：

@plugin( timeout_sec=8.0, circuit_breaker_threshold=5, fallback_func=lambda state: {"status": "unknown", "message": "物流系统维护中"} ) def track_shipment(state): # 实际调用逻辑

内核内置熔断器，当失败达阈值后自动跳过该插件5分钟，并执行fallback_func返回兜底结果。降级函数可访问state，因此能返回“根据历史数据预估，预计明早送达”。

2.5 决策5：记忆（Memory）必须分层，且每层有独立TTL与淘汰策略

把所有记忆塞进一个向量库是新手最爱犯的错。“用户昨天问过iPhone价格”和“用户三年前注册时填的身份证号”，安全等级、更新频率、保留时长天差地别。我们曾用ChromaDB统一存所有记忆，结果发现：

• 身份证号这类敏感信息被LLM不经意间泄露在调试日志里；
• 高频更新的购物车记忆挤占了向量库空间，导致冷门产品描述检索变慢；
• 没有按业务域隔离，法务Agent的合同条款记忆和客服Agent的用户投诉记忆混在一起，LLM经常张冠李戴。

我们的解法：定义三层记忆：

2.6 决策6：插件必须声明输入/输出Schema，禁止自由传参

“写个插件调用天气API”看似简单，但实际落地时，90%的故障源于参数错位：前端传city_name="北京"，插件却期待location="beijing"；LLM生成{"date": "2024-05-21"}，插件解析成datetime.date对象后，序列化回JSON时变成{"date": "2024-05-21T00:00:00"}，下游系统无法识别。

我们的解法：强制插件使用Pydantic v2定义InputModel和OutputModel，内核在调用前自动校验并转换：

class WeatherInput(BaseModel): city: str = Field(..., description="城市中文名，如'上海'") date: Optional[date] = None class WeatherOutput(BaseModel): temperature: float condition: Literal["sunny", "rainy", "cloudy"] @plugin(input_model=WeatherInput, output_model=WeatherOutput) def get_weather(input_data: WeatherInput) -> WeatherOutput: # 实现逻辑

校验失败时返回结构化错误（含字段名、期望类型、实际值），而非抛ValidationError让整个Agent崩溃。

2.7 决策7：插件契约必须包含权限声明（Permission Contract）

这是Agent安全的基石。很多团队把“调用数据库”当成原子操作，但现实中：

• 客服Agent可读取用户订单，但绝不能修改支付状态；
• 法务Agent能读合同全文，但不能导出PDF附件；
• IT运维Agent可重启服务，但不能删除数据库。

我们的解法：每个插件注册时声明permissions列表，格式为{resource}:{action}：

@plugin( permissions=["order:read", "user:read_basic"] ) def get_order_detail(state): pass @plugin( permissions=["payment:write_status"] ) def update_payment_status(state): pass

内核在调用前检查当前Agent的capability_contract是否包含所需权限，缺失则拒绝执行并记录PermissionDenied事件。权限模型支持继承：父Agent声明["order:*"]，子Agent可细化为["order:read", "order:cancel"]。

2.8 决策8：错误处理必须结构化，禁止裸抛Exception

框架默认的Exception堆栈对运维毫无价值。“KeyError: 'shipment_id'”这种错误，你得翻3层调用栈才知道是哪个插件、哪个字段缺失、影响了多少用户。我们曾因一个未捕获的requests.Timeout，导致2000+物流查询任务卡死，监控告警却只显示“Agent Execution Terminated Due to Error”。

我们的解法：定义统一错误码体系，所有插件必须返回Result对象：

from typing import Union, Optional class Result: def __init__(self, success: bool, data: Optional[dict] = None, error_code: Optional[str] = None, error_msg: Optional[str] = None): self.success = success self.data = data self.error_code = error_code # 如 "PLUGIN_TIMEOUT", "VALIDATION_FAILED" self.error_msg = error_msg # 插件必须返回Result @plugin def track_shipment(state) -> Result: try: # 业务逻辑 return Result(True, {"status": "delivered"}) except requests.Timeout: return Result(False, error_code="PLUGIN_TIMEOUT", error_msg="Carrier API timeout")

内核据此生成结构化日志，ELK中可直接按error_code聚合分析，MTTR（平均修复时间）从47分钟降至8分钟。

2.9 决策9：Agent必须支持热重载插件，无需重启进程

业务需求迭代极快。某次大促前夜，运营要求客服Agent新增“一键领取优惠券”功能，按传统方式需停服部署，但我们用热重载在2分钟内上线：上传新插件代码→内核校验签名→动态加载→灰度1%流量→确认无误后全量。

我们的解法：

• 插件以.py文件形式存放，文件名即插件ID（如coupon_grant.py）；
• 内核监听插件目录inotify事件，检测到新文件后：
  1. 用ast.parse()静态分析代码，确保无os.system、eval等危险函数；
  2. 计算文件SHA256，匹配白名单签名；
  3. importlib.util.spec_from_file_location()动态导入；
  4. 调用plugin.validate()检查契约完整性。
• 灰度通过state["task_id"].startswith("gray_")实现，不影响主流量。

2.10 决策10：日志必须包含可追溯的因果链（Causal Trace）

当用户投诉“Agent说我的订单已发货，实际还没出库”，你需要快速定位：是物流API返回假数据？还是Agent解析逻辑有bug？抑或LLM幻觉编造了状态？传统日志只有时间戳和文本，无法关联。

我们的解法：每个Agent任务生成唯一trace_id，所有日志、数据库操作、API调用、LLM请求都携带此ID。更关键的是，内核自动注入parent_span_id：

• LLM调用日志的span_id作为父ID；
• 插件调用日志的parent_span_id设为LLM的span_id；
• 插件内HTTP请求日志的parent_span_id设为插件的span_id。
  这样在Jaeger中可展开完整调用树，一眼看出是LLM → track_shipment → carrier_api哪一环出了问题。

2.11 决策11：LLM调用必须封装为可插拔的Adapter，而非硬编码API

今天用OpenAI，明天可能切到Claude或本地Qwen，后天要加多模型投票。若代码里写死openai.ChatCompletion.create()，每次切换都是灾难。我们曾为迁移到Claude，重写了17个Agent的LLM调用层。

我们的解法：定义LLMAdapter抽象基类：

class LLMAdapter(ABC): @abstractmethod def chat(self, messages: List[Dict], **kwargs) -> Dict: pass class OpenAIAdapter(LLMAdapter): def chat(self, messages, model="gpt-4", **kwargs): return openai.ChatCompletion.create(model=model, messages=messages) class ClaudeAdapter(LLMAdapter): def chat(self, messages, model="claude-3-opus-20240229", **kwargs): return anthropic.Anthropic().messages.create(model=model, messages=messages)

Agent初始化时注入适配器实例，内核调用adapter.chat()即可。切换模型只需改一行配置，零代码修改。

2.12 决策12：安全沙箱必须分级，按风险动态启用

“Agent安全”不是加个防火墙就行。我们把插件分为四级：

2.13 决策13：必须内置可观测性探针（Observability Probe）

很多团队等出问题才看日志，但Agent的异常往往是渐进式的：LLM响应延迟从800ms升到1200ms，插件成功率从99.2%降到98.7%，这些微小变化预示着更大故障。

我们的解法：内核内置三类探针：

• 延迟探针：统计每个插件、每次LLM调用的P50/P95/P99延迟，超过阈值触发告警；
• 质量探针：对LLM输出做规则校验（如“必须包含订单号”、“状态值必须在枚举中”），失败率超5%告警；
• 资源探针：监控进程内存增长、线程数、Redis连接数，防内存泄漏。
  所有指标推送到Prometheus，Grafana看板实时展示。

2.14 决策14：必须支持离线模式（Offline Mode）与确定性回放

当网络中断或LLM服务不可用时，Agent不能瘫痪。我们要求所有Agent在offline_mode=True时：

• 跳过LLM调用，用规则引擎（如Drools）处理确定性逻辑；
• 插件调用转为本地Mock，返回预设结果；
• 所有操作记录到offline_log，网络恢复后自动重放。
  更重要的是，内核支持replay_trace(trace_id)：输入任意历史trace_id，内核能100%复现当时所有决策路径，用于测试、审计、教学。

3. 通用内核的四种复利：为什么800行代码能撑起6个系统

3.1 复利1：架构复利——新Agent接入时间从5天压缩到4小时

过去开发一个电商客服Agent，要：

• 选框架（LangChain vs Dify）；
• 配置LLM Adapter；
• 写插件（订单查询、退货申请、物流跟踪）；
• 设计记忆存储；
• 接入监控告警；
• 编写测试用例；
• 压测调优。
  平均耗时5个工作日。

现在，通用内核已固化上述所有能力：

• Agent基类自带capability_contract校验、状态管理、插件调度；
• Plugin装饰器自动处理Schema校验、权限检查、熔断降级；
• LLMAdapter抽象层屏蔽模型差异；
• MemoryManager统一接口对接三层记忆；
• Tracer自动生成因果链日志；
• Probe实时上报可观测指标。

新项目只需：

1. 创建my_agent.py，继承Agent；
2. 用@plugin写3个业务插件（如get_order,apply_refund,track_logistics）；
3. 在capability_contract中声明能力；
4. 运行python my_agent.py --start。
   4小时内完成，且天然具备生产级健壮性。我们最近上线的医疗报告Agent，从需求评审到上线仅用3.5小时。

3.2 复利2：安全复利——合规审计周期缩短70%，零重大漏洞

某金融客户要求Agent通过等保三级审计，重点查：

• 敏感数据不落盘；
• 权限最小化；
• 操作全程留痕；
• 故障可回滚。

若每个Agent单独实现，审计需2周/个。而通用内核的四大安全支柱：

• 分层记忆：身份证号等PII数据只存Business层，且加密存储；
• 权限契约：所有插件调用经PermissionChecker拦截，审计日志含task_id+plugin_id+user_id；
• 因果链日志：trace_id贯穿所有操作，满足“谁在何时做了什么”的审计要求；
• 离线模式：网络中断时自动降级，避免业务停滞。

结果：6个Agent共用同一套安全实现，审计团队只验证内核一次，耗时3天，通过率100%。上线至今，0次数据泄露、0次越权访问、0次监管处罚。

3.3 复利3：运维复利——MTTR（平均修复时间）从47分钟降至8分钟

过去故障定位像侦探破案：

• 查Nginx日志找请求；
• 翻Agent日志找报错；
• 进数据库查状态；
• 抓包看API调用。
  平均耗时47分钟。

现在，trace_id让一切变得简单：

1. 运维在Kibana输入trace_id: logistics_20240521142301_8a3f；
2. 一眼看到调用树：LLM → track_shipment → carrier_api；
3. 点击carrier_api节点，看到HTTP状态码503、响应体{"error":"Service Unavailable"}；
4. 切到Prometheus，发现carrier_api_latency_p95曲线在14:22陡升；
5. 结论：物流供应商故障，非Agent问题，立即通知对方。
   整个过程8分钟，且90%的故障无需登录服务器。

3.4 复利4：演进复利——支持从Rule-Based到LLM-Native平滑迁移

很多团队担心“All-in-LLM”风险太大，想先用规则引擎，再逐步引入LLM。但传统方案要么规则和LLM割裂，要么强行把规则塞进Prompt，维护成本爆炸。

通用内核的HybridExecutor解决了这个问题：

• 同一插件可同时注册rule_func和llm_func；
• 内核按confidence_score自动路由：高置信度走规则（快、准、可解释），低置信度走LLM（灵活、泛化）；
• 所有路径共享同一state和memory，无缝衔接。

例如退货申请插件：

• 规则路径：if order.status == "shipped" and reason in ["defective", "wrong_item"]: return {"approved": True}；
• LLM路径：当用户描述模糊（如“东西不好用”），LLM分析商品评论、售后记录，给出批准建议。
  上线后，规则路径处理82%的请求（平均延迟120ms），LLM路径处理18%（平均延迟2100ms），整体体验优于纯LLM方案，且业务方随时可调整规则权重。

4. 实操：用通用内核5分钟搭建一个物流Agent

4.1 环境准备与内核安装

通用内核开源在GitHub（agent-core），支持Python 3.9+：

pip install agent-core==0.8.3 # 或从源码安装（推荐，可查看实现细节） git clone https://github.com/your-org/agent-core.git cd agent-core && pip install -e .

依赖极简：redis,psycopg2,pydantic>=2.0,requests，无LLM厂商绑定。

注意：内核默认使用SQLite做本地状态存储，生产环境务必配置Redis（REDIS_URL=redis://localhost:6379/0）和PostgreSQL（DATABASE_URL=postgresql://user:pass@localhost:5432/agentdb）。

4.2 定义物流Agent能力契约

创建logistics_agent.py：

from agent_core import Agent from agent_core.plugin import plugin # 声明能力边界 capability_contract = { "plugins": ["track_shipment", "estimate_delivery", "get_carrier_info"], "data_sources": ["tms_db", "carrier_api"], "writable_states": ["delivery_status", "estimated_time", "carrier_name"] } class LogisticsAgent(Agent): def __init__(self, *args, **kwargs): super().__init__( capability_contract=capability_contract, # 指定LLM适配器，此处用OpenAI，可随时切换 llm_adapter="openai", # 安全沙箱级别：物流API属Level 1，用restrictedpython security_level=1, **kwargs )

4.3 编写第一个插件：物流跟踪

import requests from pydantic import BaseModel, Field from agent_core.plugin import plugin class TrackInput(BaseModel): shipment_id: str = Field(..., description="运单号，如SF123456789") class TrackOutput(BaseModel): status: str = Field(..., description="当前状态，如'已签收'") location: str = Field(..., description="最新位置") updated_at: str = Field(..., description="更新时间，ISO格式") @plugin( input_model=TrackInput, output_model=TrackOutput, timeout_sec=10.0, circuit_breaker_threshold=3, fallback_func=lambda state: {"status": "未知", "location": "系统维护中", "updated_at": "2024-01-01T00:00:00Z"}, permissions=["tms_db:read", "carrier_api:read"] ) def track_shipment(input_data: TrackInput) -> TrackOutput: try: # 调用物流API（此处为示意） resp = requests.get( f"https://api.carrier.com/track/{input_data.shipment_id}", timeout=8.0 ) resp.raise_for_status() data = resp.json() return TrackOutput( status=data["status"], location=data["location"], updated_at=data["updated_at"] ) except requests.Timeout: raise Exception("Carrier API timeout") except Exception as e: # 内核会捕获并转为Result raise e

4.4 启动Agent并测试

# logistics_agent.py末尾添加 if __name__ == "__main__": # 初始化Agent agent = LogisticsAgent( task_id="logistics_test_001", state={"shipment_id": "SF123456789"} ) # 执行插件 result = agent.run_plugin("track_shipment", {"shipment_id": "SF123456789"}) print(result) # 输出结构化结果 # 或启动HTTP服务（内核内置FastAPI） # agent.serve(host="0.0.0.0", port=8000)

运行：

python logistics_agent.py # 输出： # Result(success=True, data={'status': '已签收', 'location': '北京市朝阳区', 'updated_at': '2024-05-21T15:30:22Z'}, error_code=None, error_msg=None)

实操心得：第一次运行时，内核会自动创建SQLite表、初始化Redis连接池。若报ConnectionRefusedError，请先启动Redis和PostgreSQL。我们建议用Docker Compose一键拉起：

version: '3.8' services: redis: image: redis:7-alpine ports: ["6379:6379"] postgres: image: postgres:15 environment: POSTGRES_DB: agentdb POSTGRES_USER: user POSTGRES_PASSWORD: pass ports: ["5432:5432"]

4.5 集成LLM实现自主决策

现在让Agent自己决定调用哪个插件。创建prompt_template.j2：

你是一个物流专家Agent，请根据用户问题选择最合适的动作： - 如果用户问运单状态，调用track_shipment插件； - 如果用户问预计送达时间，调用estimate_delivery插件； - 如果用户问承运商信息，调用get_carrier_info插件。 用户问题：{{ user_input }} 输出JSON：{"plugin": "插件名", "input": {"参数名": "参数值"}}

在Agent中：

from agent_core.llm import LLMAdapter # 使用OpenAI适配器 adapter = LLMAdapter("openai", model="gpt-4-turbo") # 构造Prompt prompt = render_template("prompt_template.j2", user_input="我的顺丰单SF123456789到哪了？") # 调用LLM response = adapter.chat([{"role": "user", "content": prompt}]) # 解析JSON，得到{"plugin": "track_shipment", "input": {"shipment_id": "SF123456789"}} # 自动调用对应插件 result = agent.run_plugin(response["plugin"], response["input"])

至此，一个具备自主决策能力的物流Agent诞生，全程5分钟，代码不足100行，却拥有生产级可靠性。

5. 常见问题与避坑指南：那些文档不会告诉你的真相

5.1 问题1：LLM返回的JSON格式总不规范，json.loads()频繁报错

现象：LLM常返回{"plugin": "track", "input": {...}}，但实际需要{"plugin": "track_shipment", "input": {...}}，字段名错一位就失败。

根因：LLM的“自由发挥”特性与程序的“确定性”要求冲突。

解决方案：

• 前置约束：在Prompt中明确要求“只输出JSON，无任何其他字符”，并用json.dumps()校验；
• 后置修复：内核自动尝试3种修复：
  1. 去除首尾空白和json标记；
  2. 补全缺失引号（如{plugin: "track"}→{"plugin": "track"}）；
  3. 字段名模糊匹配（"track"→"track_shipment"，基于插件注册名相似度）。
     实测修复成功率92.7%，剩余7.3%交由fallback_func兜底。

5.2 问题2：插件热重载后，旧版本函数还在内存里，导致状态混乱

现象：修改track_shipment.py后，新调用仍执行旧逻辑，print("v1")没变print("v2")。

根因：Python的import缓存机制，importlib.reload()不清理所有引用。

解决方案：

• 内核热重载时，不仅reload()模块，还清空sys.modules中相关缓存；
• 强制所有插件函数用functools.lru_cache(maxsize=None)装饰，避免闭包变量残留；
• 添加plugin_version字段，每次重载递增，内核对比版本号决定是否执行。
  我们在plugin.py中封装了safe_reload()，比原生reload()稳定10倍。

5.3 问题3：多Agent并发时，Redis状态被覆盖

现象：两个物流Agent同时处理不同运单，state["shipment_id"]相互覆盖。

根因：Redis Key设计不当，用了固定Key如agent:state。

解决方案：

• Key格式为agent:{task_id}:state，task_id全局唯一；
• 使用RedisHASH结构存state，避免单Key大Value；
• 并发更新用HSET+HGETALL，而非GET/SET，减少竞态。
  内核的StateStore类已封装此逻辑，开发者无需关心。

5.4 问题4：LLM调用超时，但插件已执行成功，造成重复操作

现象：调用支付插件扣款，网络超时，Agent重试，结果用户被扣两次款。

根因：缺乏幂等性设计，插件未声明idempotent=True。

解决方案：

• 内核为每次插件调用生成execution_id（{task_id}_{plugin_name}_{timestamp}）；
• 插件函数签名增加execution_id: str参数；
• 支付插件在执行前查DB，若execution_id已存在，直接返回上次结果；
• 所有Level 2+插件强制要求幂等性声明。
  我们为此专门写了IdempotentPlugin基类，继承即可。

5.5 问题5：向量记忆检索不准，LL