企业官网建设流程全解析

1. 项目概述：从“agency-agents”这个词组看智能体系统设计的本质逻辑

“agency-agents”不是某个具体软件、平台或开源库的官方名称，而是一个高度凝练的技术概念组合词——它把agency（能动性、主体性）和agents（智能体、代理）两个核心术语并置在一起，形成一种语义强化结构。这个词组在2023年底至2024年初开始频繁出现在技术社区、论文预印本和前沿工程实践讨论中，尤其在LLM应用架构、自主工作流编排、多智能体协同决策等方向被反复提及。它不指向某款现成工具，而是在提示一个根本性转向：我们正在从“调用模型API”走向“构建具备目标导向行为能力的数字主体”。

我第一次在某高校实验室的模拟项目X中系统性实践这个思路，是为了解决一个看似简单却长期卡点的问题：让一组文本处理任务（比如从会议纪要中提取待办、同步到日程、生成跟进邮件、自动预约下一次会议）不再依赖人工串联脚本，而是让每个环节“知道自己该做什么、什么时候做、做到什么程度算完成”。当时我们试过纯Prompt链、LangChain的SequentialChain、甚至用Celery搭异步任务队列，但都绕不开一个瓶颈——任务之间缺乏状态共识，失败后无法自主恢复，更谈不上根据环境反馈调整策略。直到把整个流程重新建模为一组具有“agency”的agents，问题才真正松动。

这个词组背后实际承载的是三层递进需求：第一层是功能封装——把能力模块化为可调度的agent；第二层是行为建模——赋予agent目标识别、计划生成、工具调用、结果评估的闭环能力；第三层是关系治理——定义多个agent如何协商、分工、仲裁冲突、共享记忆。它解决的不是“能不能跑通”，而是“能不能稳态运行、适应变化、持续进化”。适合正在搭建RAG增强问答系统、自动化客服工单分派、科研文献综述辅助、跨系统数据同步管道的工程师；也适合想跳出“写Prompt-看输出-改Prompt”循环、真正理解大模型落地底层逻辑的产品与架构师。如果你还在用硬编码if-else处理模型输出的不确定性，或者为每次新增一个业务环节就要重写整条流水线，“agency-agents”就是你该认真拆解的下一阶段范式。

2. 核心设计思路拆解：为什么必须同时强调“agency”与“agents”

2.1 “agents”只是载体，“agency”才是灵魂

很多初学者看到“agents”就立刻联想到LangChain的AgentExecutor、LlamaIndex的ReActAgent，或者AutoGen的ConversableAgent——这没错，但容易陷入一个关键误区：把agent当成“会调用工具的LLM封装器”。这种理解停留在第一层。真正的分水岭在于是否赋予其agency。Agency在认知科学和人工智能哲学中特指：一个实体能基于自身目标，在不确定环境中感知、决策、行动，并对结果负责的能力。它包含四个不可分割的要素：

• Goal-directedness（目标导向）：不是被动响应输入，而是主动维持一个内部目标状态（例如：“确保用户所有待办事项在24小时内进入CRM系统”）；
• Autonomy（自主性）：能在最小外部干预下启动、监控、终止行为（例如：检测到邮件解析失败后，自动切换备用解析规则，而非抛出异常等待人工介入）；
• Reactivity（反应性）：实时感知环境变化（如API限流、数据库连接中断、用户中途修改需求）并调整行为；
• Pro-activeness（主动性）：预判潜在障碍并提前准备（例如：在批量处理前主动检查存储空间余量，不足时触发清理流程）。

我在某公司搭建的合同条款比对系统中就吃过亏：早期版本用三个独立agent分别处理PDF解析、条款抽取、差异标定，每个都“能干活”，但当PDF解析因扫描件模糊失败时，后续两个agent直接挂起，整个流程卡死。后来给每个agent注入agency机制——解析agent失败后，不仅记录错误，还会主动触发图像增强子agent重试，并将临时增强后的文件路径写入共享状态池，抽取agent轮询到新路径后自动续跑。这个转变不是加几行代码，而是重构了整个系统的责任边界。

2.2 拒绝“伪agency”：警惕三种常见设计陷阱

实践中，有三类典型方案看似实现了agency，实则埋下长期隐患，必须提前识别：

陷阱一：目标硬编码在Prompt里
典型表现：在system prompt中写“你的目标是……”，然后靠LLM自己推理。问题在于LLM的“目标理解”极不稳定——微小的prompt扰动、上下文长度变化、甚至温度参数调整，都会导致目标漂移。我测试过同一份会议纪要，当上下文塞入额外500字无关日志时，原本稳定的“提取待办”agent有37%概率转而开始总结会议主题。真正可靠的目标锚定，必须通过结构化字段（如JSON Schema定义的goal_object）+ 运行时校验（每次决策前强制比对当前action是否服务于goal_object）来实现。

陷阱二：状态全靠LLM记忆
常见做法：把历史对话、中间结果全塞进prompt上下文。这在短流程中可行，但一旦涉及多跳推理（比如A agent查数据库→B agent分析趋势→C agent生成报告），上下文会指数级膨胀。更致命的是，LLM的“记忆”没有事务一致性——B agent看到的A agent结果，可能已是过期快照。我们在某跨平台数据同步项目中发现，当两个agent并发读写同一份缓存时，因LLM无法保证原子性，出现过12%的数据覆盖丢失。正确解法是引入轻量级状态管理器（如Redis Hash或SQLite WAL模式），所有agent只读写该中心状态，自身prompt里只保留必要上下文摘要。

陷阱三：协作靠“聊天”模拟
AutoGen的conversational pattern很诱人，但真实业务中，agent间“对话”本质是低效的序列化通信。我们曾用纯对话模式实现采购审批流：采购员agent发请求→主管agent审阅→财务agent核价→最终agent归档。测试发现，仅消息序列化/反序列化+LLM解析就占去63%的端到端延迟，且任何一环prompt微调都需全链路回归。后来改为事件驱动架构：每个agent监听特定topic（如“approval.requested”），处理完发布“approval.reviewed”事件，下游agent订阅即可。延迟下降至原来的1/5，更重要的是，事件日志天然形成可审计的行为链。

2.3 架构选型逻辑：为什么推荐“状态机+事件总线+轻量LLM”组合

综合上述教训，我目前在所有新项目中统一采用三层架构：

• 最底层：确定性状态机（State Machine）
  用Python的transitions库或Node.js的xstate实现。它不处理业务逻辑，只定义合法状态迁移（如“pending → processing → success”或“pending → processing → failed → retrying”）。所有agent的生命周期、失败回退、超时熔断都由状态机驱动。好处是：行为完全可预测、可测试、可回放。我们曾用状态机日志精准复现了一个凌晨发生的偶发性内存溢出，定位到是某个agent在retrying状态未释放临时文件句柄。

• 中间层：事件总线（Event Bus）
  选用NATS JetStream（轻量、持久化、支持流式消费）或RabbitMQ（成熟、运维友好）。agent间不直接调用，只发布/订阅事件。每个事件携带结构化payload（含timestamp、trace_id、payload_schema_version），天然支持异步、解耦、流量削峰。某次促销活动期间，订单处理agent突发流量激增，事件总线自动缓冲了87%的瞬时峰值，下游库存校验agent平稳消化，零故障。

• 最上层：LLM执行单元（LLM Executor）
  每个agent对应一个独立的LLM执行上下文，但严格限定其职责：只做三件事——解析输入事件、调用预注册工具（数据库查询、API调用、文件操作）、生成标准格式输出事件。绝不允许LLM自行决定下一步调用哪个agent——那属于状态机的管辖范围。这样既发挥LLM的语义理解优势，又规避其不可控性。实测下来，这种切分使单agent平均响应时间稳定在1.2s内（P95），且错误率低于0.3%。

这个组合不是理论最优，而是经过十多个项目验证的“稳态生产解”。它放弃了一部分LLM的“灵性”，换来了可运维性、可观测性和可扩展性——这才是agency-agents落地的生命线。

3. 核心细节解析与实操要点：从概念到可运行代码的关键跨越

3.1 Agent的最小可行定义：五个必含字段与两个隐式契约

一个真正具备agency的agent，代码层面必须显式声明以下五个字段，缺一不可。这不是框架要求，而是领域逻辑的刚性约束：

1. id: str—— 全局唯一标识，用于日志追踪、状态隔离、事件路由。禁止用随机UUID，必须带业务语义（如"contract-parser-v2"）。某次线上事故追溯中，正是靠ID中的v2后缀，快速定位到是新旧版本agent共存导致的状态冲突。

2. goal: Dict[str, Any]—— 结构化目标描述，必须包含objective（终极目标）、constraints（硬性限制，如“耗时<5s”、“不得修改原始PDF”）、success_criteria（成功判定规则，如“输出JSON含3个以上valid_clause字段”）。我们用Pydantic V2定义schema，启动时强制校验，避免运行时才发现目标缺失。

3. tools: List[Tool]—— 预注册工具列表，每个Tool必须声明name、description、input_schema（JSON Schema）、output_schema。重点：不允许agent在运行时动态加载未注册工具。这看似限制灵活性，实则杜绝了“LLM幻觉调用不存在API”的经典故障。某金融项目中，因开放了动态tool加载，LLM曾虚构出bank_transfer_v3接口并尝试调用，触发了沙箱告警。

4. state_manager: StateManager—— 状态管理器实例，负责读写该agent专属状态（如last_processed_timestamp、retry_count）。我们封装了Redis-backed的StateManager，所有读写操作自动包裹分布式锁，避免并发冲突。实测在1000QPS压力下，状态一致性达100%。

5. event_bus: EventBus—— 事件总线客户端，用于发布agent_started、agent_completed、agent_failed等生命周期事件。关键设计：每个agent启动时自动订阅{self.id}.controltopic，接收暂停、终止、参数热更新指令——这是实现动态治理的基础。

两个隐式契约则关乎协作可靠性：

• 契约一：事件幂等性
  所有agent必须假设同一事件可能被重复投递（网络重传、消费者崩溃重启）。因此，process_event()方法内必须先校验事件ID是否已处理（查状态库），再执行业务逻辑。我们用Redis SETNX + TTL实现轻量去重，成本几乎为零。

• 契约二：状态终局性
  Agent完成任务后，必须将最终状态（含结果、耗时、错误码）写入状态库，并发布completed事件。绝不允许“静默退出”。这使得监控系统能实时计算各环节SLA，某次发现invoice-validatoragent的completion率突然跌至82%，排查发现是上游OCR服务升级后返回格式变更，立即触发告警。

3.2 状态机设计：用有限状态覆盖95%的业务异常流

agency-agents系统最大的价值，往往体现在异常处理环节。我们摒弃了传统“try-catch-throw”的线性错误处理，转而用状态机建模所有可能的失败路径。以一个典型的“客户信息同步agent”为例，其状态图包含7个核心状态：

这个设计的关键在于：每个失败状态都有明确的、可配置的退出策略。fetch_failed状态不是简单重试，而是内置退避算法（首次1s，二次3s，三次10s），且重试前强制检查CRM服务健康度（调用其/health端点）。processing_failed状态不终止流程，而是降级处理——比如当地址标准化服务不可用时，跳过地址清洗，仅保留原始字段入库。这些策略全部外置为JSON配置，无需重启服务即可调整。

我们曾用此状态机捕获一个隐藏很深的时区bug：pushing状态向ERP提交的时间戳未转换为ERP所在时区，导致所有夜间订单被标记为“明日处理”。状态机在completed前强制校验时间戳有效性（对比本地时间差值>2h则告警），第一时间暴露了问题。

3.3 工具注册与调用：让LLM只做它最擅长的事

LLM在agency-agents中不是万能胶，而是精密的“意图解析器”和“工具协调器”。它的核心价值在于：从非结构化输入（如用户自然语言请求、邮件正文、日志片段）中，精准提取结构化参数，并选择正确的工具组合。为此，工具注册必须遵循三项铁律：

铁律一：工具必须声明副作用范围
每个Tool的description中，必须用固定前缀标明影响域：

• [READ]：只读操作（如get_customer_by_id）
• [WRITE]：写操作（如update_order_status）
• [SIDE_EFFECT]：有外部影响（如send_email、trigger_webhook）

LLM在规划步骤时，会优先考虑[READ]工具探查现状，再决定是否调用[WRITE]。某次误将[SIDE_EFFECT]的send_notification标记为[WRITE]，导致LLM在测试环境中多次触发真实短信，我们立即增加了调用前的双因素确认（需匹配预设的test_modeflag）。

铁律二：输入输出强Schema约束
不用自然语言描述参数，而用JSON Schema精确声明：

{ "name": "search_knowledge_base", "description": "[READ] 在知识库中检索与query最相关的3个文档片段", "input_schema": { "type": "object", "properties": { "query": {"type": "string", "minLength": 2}, "max_results": {"type": "integer", "minimum": 1, "maximum": 10} }, "required": ["query"] }, "output_schema": { "type": "array", "items": { "type": "object", "properties": { "doc_id": {"type": "string"}, "snippet": {"type": "string"}, "relevance_score": {"type": "number", "minimum": 0, "maximum": 1} } } } }

LLM调用时，框架自动校验输入是否符合schema，输出是否满足output_schema。不符合则拒绝执行并记录违规事件——这比任何prompt约束都可靠。

铁律三：工具调用必须带可追溯凭证
每次LLM发起工具调用，框架自动生成tool_call_id，并绑定到当前事件的trace_id。所有工具执行日志必须包含这两个ID。当某次payment_gateway.charge调用超时，我们通过trace_id在ELK中秒级关联出：上游order_validation耗时异常升高→触发了风控规则→增加支付验证步骤→导致整体超时。没有这个凭证链，根本无法定位根因。

4. 实操过程与核心环节实现：手把手搭建一个可运行的agency-agents系统

4.1 环境准备与依赖安装：精简到极致的运行时

我们坚持“最小可行依赖”原则。一个基础agency-agents系统，仅需以下6个核心包（以Python为例）：

pip install \ transitions==0.10.0 \ # 状态机引擎，轻量无依赖 nats-py==2.10.0 \ # NATS客户端，支持JetStream持久化 redis==4.6.0 \ # 状态存储，比SQLite更适合高并发 pydantic==2.7.1 \ # 数据验证，v2版性能提升3倍 python-dotenv==1.0.0 \ # 环境变量管理，安全分离配置 structlog==23.3.0 # 结构化日志，天然支持trace_id注入

提示：坚决不安装langchain、llama-index等大而全框架。它们抽象层过厚，掩盖了agency的核心矛盾。我们用原生requests调用LLM API，用raw SQL操作数据库——控制力远胜于便利性。

环境变量.env文件模板：

# LLM配置 LLM_API_URL=https://api.example.com/v1/chat/completions LLM_API_KEY=sk-xxx LLM_MODEL_NAME=gpt-4-turbo-preview LLM_TIMEOUT=30 # 状态存储 REDIS_URL=redis://localhost:6379/0 REDIS_PASSWORD= # 事件总线 NATS_URL=nats://localhost:4222 NATS_STREAM_NAME=agency_events NATS_RETENTION_POLICY=limits # 按消息数保留，防磁盘爆满 # 监控 LOG_LEVEL=INFO TRACE_SAMPLING_RATE=0.1 # 10%请求开启全链路追踪

关键配置说明：NATS_RETENTION_POLICY=limits是生产必备——无限保留消息会撑爆磁盘。我们设置max_msgs=1000000，配合max_age=24h，确保既满足审计需求，又不拖垮系统。

4.2 定义第一个Agent：合同解析器（contract-parser）

以下是contract-parseragent的核心代码，完整可运行（已脱敏）：

# agents/contract_parser.py from typing import Dict, Any, List, Optional from pydantic import BaseModel, Field from transitions import Machine import json import logging import structlog from utils.state_manager import RedisStateManager from utils.event_bus import NatsEventBus from utils.llm_client import LLMClient logger = structlog.get_logger() class ContractParseGoal(BaseModel): objective: str = "Extract structured clauses from contract PDF" constraints: Dict[str, Any] = Field(default_factory=lambda: { "max_processing_time_sec": 15, "allowed_file_types": ["pdf"], "output_format": "json" }) success_criteria: Dict[str, Any] = Field(default_factory=lambda: { "required_fields": ["parties", "effective_date", "termination_clause"], "min_clauses_extracted": 5 }) class ContractParseInput(BaseModel): pdf_url: str = Field(..., description="Publicly accessible URL of the PDF") document_id: str = Field(..., description="Unique ID for audit trail") class ContractParseOutput(BaseModel): document_id: str clauses: List[Dict[str, Any]] parsing_quality_score: float = Field(ge=0.0, le=1.0) raw_text_preview: str = Field(max_length=500) class ContractParserAgent: def __init__(self, agent_id: str = "contract-parser-v2"): self.id = agent_id self.state_manager = RedisStateManager(agent_id) self.event_bus = NatsEventBus() self.llm_client = LLMClient() # 定义状态机 self.states = ['idle', 'fetching', 'fetch_failed', 'parsing', 'parsing_failed', 'validating', 'completed', 'failed'] self.transitions = [ {'trigger': 'start', 'source': 'idle', 'dest': 'fetching'}, {'trigger': 'fetch_success', 'source': 'fetching', 'dest': 'parsing'}, {'trigger': 'fetch_failure', 'source': 'fetching', 'dest': 'fetch_failed'}, {'trigger': 'parse_success', 'source': 'parsing', 'dest': 'validating'}, {'trigger': 'parse_failure', 'source': 'parsing', 'dest': 'parsing_failed'}, {'trigger': 'validate_success', 'source': 'validating', 'dest': 'completed'}, {'trigger': 'validate_failure', 'source': 'validating', 'dest': 'failed'}, {'trigger': 'retry_fetch', 'source': 'fetch_failed', 'dest': 'fetching'}, {'trigger': 'retry_parse', 'source': 'parsing_failed', 'dest': 'parsing'}, ] self.machine = Machine(model=self, states=self.states, transitions=self.transitions, initial='idle') # 注册工具 self.tools = [ { "name": "download_pdf", "description": "[READ] Download PDF from given URL and return base64 encoded content", "input_schema": {"type": "object", "properties": {"url": {"type": "string"}}, "required": ["url"]}, "output_schema": {"type": "object", "properties": {"content_b64": {"type": "string"}}} }, { "name": "parse_pdf_to_text", "description": "[READ] Extract text from base64 PDF using OCR if needed", "input_schema": {"type": "object", "properties": {"content_b64": {"type": "string"}}, "required": ["content_b64"]}, "output_schema": {"type": "object", "properties": {"text": {"type": "string"}}} } ] def process_event(self, event: Dict[str, Any]) -> None: """主入口：处理来自事件总线的事件""" try: # 1. 解析输入 input_data = ContractParseInput(**event.get("payload", {})) # 2. 初始化状态 self.state_manager.set("input", input_data.dict()) self.state_manager.set("start_time", time.time()) # 3. 启动状态机 self.start() # 4. 状态流转 while self.state not in ['completed', 'failed']: if self.state == 'fetching': self._handle_fetching(input_data.pdf_url) elif self.state == 'parsing': self._handle_parsing() elif self.state == 'validating': self._handle_validating() except Exception as e: logger.error("agent_process_error", agent_id=self.id, error=str(e), event_id=event.get("id")) self.to_failed() def _handle_fetching(self, pdf_url: str) -> None: try: # 调用download_pdf工具 tool_result = self.llm_client.call_tool( tool_name="download_pdf", tool_input={"url": pdf_url} ) # 保存原始PDF内容 self.state_manager.set("pdf_content_b64", tool_result["content_b64"]) self.fetch_success() except Exception as e: logger.warning("fetch_failed", url=pdf_url, error=str(e)) self.fetch_failure() def _handle_parsing(self) -> None: try: pdf_content = self.state_manager.get("pdf_content_b64") # 构造LLM提示：要求输出JSON格式的条款列表 prompt = f"""你是一个专业的法律文档解析器。请从以下PDF文本中提取所有关键条款，按JSON格式输出，包含字段：parties（签约方列表）、effective_date（生效日期）、termination_clause（终止条款）、governing_law（适用法律）、payment_terms（付款条款）。文本：{pdf_content[:10000]}...""" response = self.llm_client.chat_completion( messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"} ) clauses = json.loads(response["choices"][0]["message"]["content"]) self.state_manager.set("raw_clauses", clauses) self.parse_success() except Exception as e: logger.warning("parse_failed", error=str(e)) self.parse_failure() def _handle_validating(self) -> None: try: raw_clauses = self.state_manager.get("raw_clauses") # 基于success_criteria做硬校验 if len(raw_clauses) < 5: raise ValueError(f"Too few clauses extracted: {len(raw_clauses)}") required_fields = ["parties", "effective_date", "termination_clause"] missing = [f for f in required_fields if f not in raw_clauses] if missing: raise ValueError(f"Missing required fields: {missing}") # 计算质量分（简单规则：字段完整性+文本长度） completeness = len([k for k in raw_clauses.keys() if k in required_fields]) / len(required_fields) quality_score = min(0.9, completeness + 0.1 * (len(str(raw_clauses)) / 1000)) output = ContractParseOutput( document_id=self.state_manager.get("input")["document_id"], clauses=raw_clauses, parsing_quality_score=quality_score, raw_text_preview=str(raw_clauses)[:500] ) self.state_manager.set("output", output.dict()) self.validate_success() except Exception as e: logger.warning("validate_failed", error=str(e)) self.validate_failure() def on_enter_completed(self) -> None: """状态机进入completed时的钩子""" output = self.state_manager.get("output") # 发布完成事件 self.event_bus.publish("contract.parsed", { "document_id": output["document_id"], "clauses": output["clauses"], "quality_score": output["parsing_quality_score"] }) # 清理临时状态 self.state_manager.delete("pdf_content_b64") self.state_manager.delete("raw_clauses") logger.info("agent_completed", agent_id=self.id, document_id=output["document_id"]) # 使用示例 if __name__ == "__main__": parser = ContractParserAgent() # 模拟接收事件 test_event = { "id": "evt_123", "type": "contract.parse.request", "payload": { "pdf_url": "https://example.com/contracts/nda_v2.pdf", "document_id": "doc_789" } } parser.process_event(test_event)

这段代码体现了agency-agents的核心思想：状态机驱动流程，LLM专注意图解析，工具提供确定性能力。它没有一行代码在“指挥”另一个agent，所有协作都通过事件总线完成。

4.3 启动事件总线与状态管理：三分钟部署生产级基础设施

NATS JetStream和Redis的部署，我们坚持Docker Compose一键启停，配置文件docker-compose.yml如下：

version: '3.8' services: nats: image: nats:2.10.14 ports: - "4222:4222" # Client port - "8222:8222" # Monitoring port - "7777:7777" # JetStream management command: | -js --store_dir /data/jetstream --max_payload 10MB --max_connections 1000 --max_subscriptions 10000 volumes: - ./nats-data:/data/jetstream restart: unless-stopped redis: image: redis:7.2-alpine command: redis-server --save 60 1 --loglevel warning --maxmemory 512mb --maxmemory-policy allkeys-lru ports: - "6379:6379" volumes: - ./redis-data:/data restart: unless-stopped # 可选：监控面板 prometheus: image: prom/prometheus:latest ports: - "9090:9090" volumes: - ./prometheus.yml:/etc/prometheus/prometheus.yml

配套的prometheus.yml监控配置：

global: scrape_interval: 15s scrape_configs: - job_name: 'nats' static_configs: - targets: ['nats:8222'] - job_name: 'redis' static_configs: - targets: ['redis:6379'] - job_name: 'agency-agents' static_configs: - targets: ['host.docker.internal:8000'] # 假设你的agent服务暴露在此

注意：host.docker.internal是Docker Desktop的特殊DNS，Linux需替换为宿主机IP。实测这套组合在4核8G服务器上，可稳定支撑500+并发agent实例，CPU占用率常年低于40%。

启动命令：

docker-compose up -d # 等待30秒，检查服务健康 curl http://localhost:8222/streamz # 应返回JetStream状态 curl http://localhost:6379/health # 应返回Redis健康信息

此时，你的agency-agents系统底座已就绪。接下来只需编写更多agent（如clause-validator、risk-assessor），它们自动接入同一套状态机和事件总线，形成有机整体。

5. 常见问题与排查技巧实录：那些文档里不会写的实战经验

5.1 典型问题速查表：从现象到根因的快速定位路径

5.2 我踩过的五个深坑及独家避坑技巧

坑一：LLM的“自信幻觉”导致工具调用参数错误
现象：search_knowledge_base工具要求max_results为整数，但LLM有时输出"max_results": "5"（字符串）。虽然JSON Schema校验会捕获，但LLM在重试时可能继续犯同样错误。
独家技巧：在LLM调用工具前，框架层自动做类型强制转换。对max_results字段，无论LLM输出"5"还是5.0，都转为int(5)。我们封装了safe_cast函数，覆盖所有常见类型（int, float, bool, list, dict），并记录转换日志供审计。实测将此类错误率从12%降至0.2%。

坑二：事件总线消息堆积引发OOM
现象：某次全量数据迁移，上游agent爆发式发布10万条sync.request事件，NATS内存暴涨，容器OOM被K8s杀死。
独家技巧：在NATS Stream配置中启用storage: file（而非memory），并设置max_bytes=10GB。更重要的是，在agent发布事件前，增加背压控制：if nats_stream_size > 50000: sleep(0.1)。这个简单的sleep，让系统在峰值时自动降速，平滑吞吐量。我们称之为“软熔断”。

坑三：状态机状态不一致导致流程断裂
现象：parsing状态agent崩溃重启后，状态机仍处于parsing，但pdf_content_b64已丢失，无法续跑。
独家技巧：实现状态机的“自愈”能力。在on_enter_parsing钩子中，先检查`pdf_content