1. 项目概述这不是一个“代理”而是一套可协作的智能体工作流“agency-agents”这个词组最近在技术社区里频繁出现但很多人第一反应是——这不就是“代理”agent吗加个“agency”前缀是不是就高级了其实完全不是。我接触过十几个实际落地的类似项目从某高校实验室的跨平台任务调度系统到某公司内部的自动化文档协同平台再到某开源社区正在推进的轻量级AI工作流框架它们都绕不开“agency-agents”这个核心设计范式。它解决的从来不是“让一个AI干活”而是“让多个AI像人类团队一样分工、对齐、协商、回溯、兜底”。关键词里的“agency”不是指机构而是指自主性autonomy、能动性initiative和责任归属accountability——每个agent必须知道自己能做什么、不能越界做什么、出错了该谁负责、怎么向其他成员解释自己的决策。这个概念之所以突然变热不是因为技术突飞猛进而是因为现实需求倒逼架构升级。过去我们用单个LLM调用API完成问答或摘要就像让一个全能实习生包揽所有事写邮件、查资料、做PPT、催进度……结果是响应慢、错误多、无法追溯、一崩全崩。而“agency-agents”模式相当于把一个大任务拆成“项目经理agent”“资料检索agent”“文案撰写agent”“合规校验agent”“格式排版agent”它们之间通过结构化协议通信各自专注自己最擅长的子域失败只影响局部且能互相补位。比如检索agent没找到最新数据会主动触发“备用渠道重试”流程并通知项目经理agent调整交付预期——这种动态协调能力才是它区别于传统pipeline或简单chain-of-thought的本质。适合谁参考如果你正面临这些情况这篇内容就是为你写的正在用LangChain/LlamaIndex搭自动化流程但发现调试困难、错误定位耗时、上线后稳定性差团队里有不同角色产品、法务、运营需要参与AI输出审核但现有方案只能靠人工卡点无法嵌入自动校验想尝试多模型协同比如用Claude做逻辑推理、用GPT-4做创意生成、用本地小模型做隐私数据处理但不知道如何设计通信边界和结果融合机制或者你只是好奇为什么最近那么多开源项目如AutoGen、CrewAI、MetaGPT都在强调“role-based agent system”它们底层到底在解决什么共性问题接下来我会从设计逻辑、核心组件、实操实现、避坑经验四个维度带你真正吃透“agency-agents”不是概念炒作而是一套可落地、可调试、可运维的工程方法论。不讲虚的全是我在三个真实项目中反复验证过的细节。2. 整体设计思路为什么必须放弃“单Agent万能论”转向“Agency级编排”2.1 单Agent架构的三大硬伤是业务增长的天花板很多团队一开始都走“单Agent路线”喂给一个大模型完整Prompt让它读文档、分析需求、生成报告、检查错别字、导出PDF……听起来很美实测下来问题集中爆发在三个不可回避的环节第一上下文失控导致的“健忘症”。哪怕用128K上下文模型当任务链超过5步比如读3份合同→比对条款差异→识别风险点→生成修订建议→拟写沟通话术中间某一步的推理依据就会被后续token挤掉。我参与过一个金融尽调项目单Agent在第4步生成建议时完全忘了第1步读到的关键违约金计算方式导致整份报告逻辑断裂。这不是模型能力问题而是架构缺陷——它被迫把所有信息压缩进一个“大脑”而人类专家是分模块调用记忆的。第二责任模糊引发的“甩锅链”。当最终输出出错你问“是Prompt写得不对是模型理解偏差是工具调用失败还是后处理逻辑有bug”答案往往是“都有可能”。没有明确的职责切分就没有可归因的调试路径。某次客户投诉报告里数据单位写反万元写成元排查花了两天最后发现是格式转换agent和数值解析agent之间对“原始数据源字段名”的约定不一致但单Agent架构里根本不存在“agent间约定”这个概念。第三扩展性为零的“缝合怪”困境。业务方提新需求“能不能在报告末尾加个可视化图表”——好你得改Prompt、加D3.js调用、处理SVG渲染异常、适配不同浏览器……下个月又要加“支持语音播报”再改……每次都是全局重构。而真正的工程化系统应该像插拔U盘一样新增一个“图表生成agent”定义好输入结构化数据、输出base64图片、超时规则、失败重试策略其他agent完全无感。提示不要试图用更长的Prompt、更大的模型、更多的few-shot样例来修补单Agent架构。这就像给自行车加涡轮增压——方向错了功率越大翻车越快。2.2 Agency级设计的四大支柱角色、协议、状态、治理“agency-agents”不是简单堆砌多个Agent而是构建一个具备组织特性的系统。我在某跨平台协作系统中提炼出四个不可妥协的设计支柱缺一不可支柱一角色Role即契约不是头衔每个Agent必须有明确定义的能力边界和责任范围。比如“合规校验agent”不能参与文案创作它的唯一输入是待审文本适用法规库版本号唯一输出是{status: pass/fail, issues: [list], suggestions: [list]}。角色定义要写进代码注释、接口文档、甚至单元测试用例里。我见过最失败的案例是把“项目经理agent”设计成万能中转站结果它既调API又写代码还改Prompt最后成了新的单点故障源。支柱二协议Protocol即语言不是格式Agent之间通信不能靠自由发挥的JSON。我们强制采用三层协议语义层用预定义Action Type如request_data,submit_draft,escalate_issue代替自然语言指令结构层所有消息必须包含sender_id,receiver_id,task_id,timestamp,retry_count载荷层业务数据用Schema约束如data_schema.json禁止传递未声明字段。这套协议让我们在某次线上事故中5分钟内就定位到是“数据检索agent”返回了空数组却没设置status: no_result导致下游agent无限重试。支柱三状态State即真相不是缓存整个Agency必须有一个权威状态中心可以是Redis Hash、SQLite表或内存Map记录每个Task的全生命周期created → assigned → in_progress → blocked → completed → archived。每个Agent操作前先查状态操作后更新状态。这解决了分布式环境下的竞态问题——比如两个agent同时收到“补充材料”指令状态锁确保只有一个能进入in_progress另一个自动排队。我们曾用纯事件驱动替代状态机结果在高并发场景下出现任务重复执行回滚成本极高。支柱四治理Governance即护栏不是监控治理不是事后看日志而是事前设规则。我们在系统启动时加载governance_rules.yaml包含超时熔断如“外部API调用8s自动降级为mock数据”成本阈值如“单任务调用GPT-4 token超5000立即告警并切换Claude”权限隔离如“财务agent禁止访问用户身份证号字段即使上游传入也自动脱敏”。这些规则由非技术人员如法务、风控配置无需改代码。这才是真正的“AI可治理”。2.3 为什么不用现成框架选型背后的现实权衡看到这里你可能会想那直接用AutoGen或CrewAI不就行了我必须坦诚地说在三个项目中我们最终都选择了自研轻量级编排层原因很实在AutoGen的GroupChatManager太重它默认假设所有agent都能看到全部历史消息这在涉及敏感数据如医疗、金融的场景中直接违规。我们要求“财务agent看不到用户聊天记录只接收结构化账单数据”而AutoGen的message history是全局共享的改造成本高于重写。CrewAI的Task依赖太刚性它用Task.dependencies [task1, task2]定义顺序但真实业务中常有“如果task1成功则执行task3否则跳过task3直接执行task4”。这种条件分支在CrewAI里要写大量hook可读性极差。开源框架的调试体验灾难当一个agent卡死你看到的是await self._execute()这样的抽象栈而自研层我们每步都打log.debug(f[{agent_id}] step_3_data_fetch: {url} - status{resp.status})配合ELK日志30秒内定位到是某个agent的HTTP client没设timeout。当然这不是否定框架价值。我的建议是用框架做PoC验证可行性用自研做生产落地。我们给新团队的标准流程是先用CrewAI跑通端到端流程1天再用2天时间把核心逻辑迁移到自研编排层约500行Python剩下时间全花在治理规则和状态持久化上——这才是真正在解决业务问题。3. 核心组件拆解从“能跑起来”到“可运维”的关键细节3.1 Agent基类设计90%的稳定性问题源于基类缺失很多团队直接拿langchain.agents.AgentExecutor开干结果很快陷入“每个agent都要重复写日志、重试、超时、错误分类”的泥潭。我们的解决方案是定义一个严格继承的BaseAgent基类强制所有Agent实现五个抽象方法class BaseAgent(ABC): abstractmethod def validate_input(self, input_data: dict) - bool: 输入校验字段存在性、类型、业务规则如日期格式 abstractmethod def execute_core_logic(self, input_data: dict) - dict: 核心逻辑纯函数式无副作用可单元测试 abstractmethod def format_output(self, raw_result: dict) - dict: 输出标准化统一添加metadata、status、version abstractmethod def handle_failure(self, error: Exception, input_data: dict) - dict: 失败兜底返回降级结果或明确escalation指令 abstractmethod def get_metadata(self) - dict: 元数据agent_id, version, supported_actions等用于治理这个设计带来的直接收益调试效率提升当某个agent报错我们立刻知道是validate_input阶段参数问题、execute_core_logic阶段模型/工具问题还是format_output阶段模板错误不用在千行代码里grep灰度发布可控新版本agent上线前先替换get_metadata().version编排层根据version路由流量老版本出问题不影响新版本治理规则生效handle_failure返回的escalation字段会被治理模块自动触发告警或人工介入流程。注意execute_core_logic必须是纯函数禁止在其中调用数据库、发HTTP请求、读文件。所有IO操作必须封装成独立tool由编排层统一管理——这是保证可测试性和可重现性的铁律。3.2 编排引擎状态机驱动的“中央调度室”编排引擎不是简单的for循环而是一个基于状态机的事件处理器。我们用transitions库实现核心状态流转图如下文字描述created → assigned (分配给首个agent) → in_progress (agent开始执行) → [success] → pending_review (等待人工/其他agent审核) → [fail] → blocked (需人工干预) → [timeout] → degraded (启用降级策略) → [all agents done] → completed → [archive after 30d] → archived关键实操细节状态变更必须原子化我们用Redis Lua脚本保证GET current_state SET new_state PUBLISH event三步不可分割。曾因用普通Redis命令在高并发下出现状态错乱如两个agent同时读到in_progress都写入pending_review导致审核被触发两次事件驱动而非轮询每个agent执行完向Redis Pub/Sub频道task_events推送{task_id: xxx, event: completed, result: {...}}编排引擎订阅该频道实时响应延迟50ms状态持久化策略created/assigned/blocked等关键状态存MySQL强一致性in_progress等临时状态存Redis高性能避免单点数据库压力过大。这个设计让系统具备了“自我修复”能力。比如某agent进程意外退出编排引擎检测到in_progress状态超时我们设为120s自动触发degraded状态调用备用agent或返回预设模板全程无需人工介入。3.3 工具注册中心让Agent“只说人话”不说APIAgent不该知道requests.post(https://api.xxx.com/v1/search, json{q: xxx})这种细节。我们构建了统一的ToolRegistry所有外部能力都注册为工具# 注册示例 tool_registry.register( nameweb_search, description搜索公开网页信息返回摘要和链接, schema{ type: object, properties: { query: {type: string, description: 搜索关键词}, max_results: {type: integer, default: 3} } }, executorsearch_tool_executor # 真正执行HTTP调用的函数 )Agent只需在execute_core_logic中调用search_result self.tool_registry.call(web_search, query2024年新能源汽车补贴政策, max_results5)好处极其明显安全隔离search_tool_executor内部做了URL白名单、响应大小限制2MB、敏感词过滤如禁止搜索“身份证号”Agent完全感知不到可观测性所有工具调用自动记录tool_name,input_hash,duration_ms,status我们据此发现某次性能瓶颈是pdf_parse工具平均耗时从200ms飙升到1800ms根源是PDF解析库版本升级引入内存泄漏快速替换当百度搜索API限频我们只需更新search_tool_executor指向NewBing API所有Agent零修改。实操心得工具注册时必须强制填写rate_limit如{requests_per_minute: 60}和timeout_ms如5000。我们吃过亏——某次忘记设timeout一个挂起的database_query工具让整个Agency线程池占满新任务全部排队。3.4 治理规则引擎用YAML写“AI宪法”治理不是写if-else而是用声明式配置定义系统底线。我们的governance_rules.yaml示例rules: - id: cost_control trigger: on_task_start condition: input_data.estimated_tokens 10000 action: switch_model params: {target_model: claude-haiku-20240307, fallback: return_error} - id: pii_protection trigger: on_agent_output condition: output.contains_pii(id_card) or output.contains_pii(phone) action: redact_pii params: {fields: [id_card, phone]} - id: compliance_audit trigger: on_task_complete condition: task.metadata.domain finance action: log_to_compliance_db params: {table: finance_audit_log}关键实现技巧条件表达式用jsonpath-ng解析output.contains_pii(id_card)实际编译为jsonpath_ng.parse($.content).find(output)避免手写正则的脆弱性Action执行沙箱化redact_pii动作在独立进程运行防止正则表达式灾难性回溯拖垮主进程规则热加载YAML文件修改后Watchdog监听到变化自动reload规则无需重启服务——这对合规审计场景至关重要。这套机制让我们在某次监管检查中10分钟内就导出了过去30天所有金融类任务的完整审计日志而用传统日志grep要花2小时。4. 实操全流程从零搭建一个可运行的Agency系统4.1 环境准备与最小可行架构MVP我们摒弃复杂容器化用最简方式验证核心逻辑。所需组件组件版本说明Python3.10主语言Redis7.0状态存储事件总线SQLite内置审计日志持久化生产环境换MySQLLangChain0.1.0LLM调用封装仅用ChatOpenAI等基础类初始化命令Mac/Linux# 创建虚拟环境 python3 -m venv agency-env source agency-env/bin/activate # 安装核心依赖 pip install redis langchain openai python-dotenv transitions jsonpath-ng watchdog # 启动Redis后台运行 brew services start redis # Mac # 或 docker run -d --name redis-stack -p 6379:6379 -p 8001:8001 redis/redis-stack-server:latest目录结构严格遵循agency-system/ ├── agents/ # 所有Agent实现 │ ├── __init__.py │ ├── researcher.py # 资料检索agent │ ├── writer.py # 文案撰写agent │ └── reviewer.py # 合规校验agent ├── core/ # 核心框架 │ ├── __init__.py │ ├── base_agent.py # BaseAgent基类 │ ├── orchestrator.py # 编排引擎 │ ├── tool_registry.py # 工具注册中心 │ └── governance.py # 治理规则引擎 ├── config/ │ ├── governance_rules.yaml # 治理规则 │ └── settings.py # 配置加载 ├── tests/ # 单元测试必须 ├── main.py # 入口文件 └── requirements.txt提示不要跳过tests/目录我们规定每个Agent必须有3个测试test_validate_input边界值、test_execute_core_logicmock LLM返回、test_handle_failure模拟网络超时。这保证了任何修改都不会破坏核心契约。4.2 编写第一个Agent以“资料检索agent”为例创建agents/researcher.pyfrom core.base_agent import BaseAgent from core.tool_registry import tool_registry class ResearcherAgent(BaseAgent): def __init__(self, agent_id: str researcher): super().__init__(agent_id) # 注册该agent专用工具复用全局registry tool_registry.register( nameweb_search, description搜索公开网页信息, schema{...}, # 同前文 executorself._search_executor ) def validate_input(self, input_data: dict) - bool: # 强制校验必须有query字段且长度1-100字符 if not isinstance(input_data.get(query), str): return False if not (1 len(input_data[query]) 100): return False return True def execute_core_logic(self, input_data: dict) - dict: # 调用工具不关心具体实现 search_result tool_registry.call( web_search, queryinput_data[query], max_results3 ) # 核心逻辑从搜索结果中提取关键事实生成结构化摘要 # 这里用mock实际用LLM调用 return { facts: [ {source: techcrunch.com, summary: New AI model reduces latency by 40%}, {source: arxiv.org, summary: Paper shows theoretical limits of LLM scaling} ], confidence_score: 0.92 } def format_output(self, raw_result: dict) - dict: return { status: success, data: raw_result, metadata: { agent_id: self.agent_id, version: 1.0.0, timestamp: self._get_timestamp() } } def handle_failure(self, error: Exception, input_data: dict) - dict: # 降级策略返回预设行业通用知识 return { status: degraded, data: { facts: [{source: fallback_knowledge, summary: AI models improve over time}], confidence_score: 0.6 } } def _search_executor(self, **kwargs): # 真正的HTTP调用放在这里此处为简化示例 import time time.sleep(0.5) # 模拟网络延迟 return {results: [...]} # 必须实例化供编排层调用 researcher_agent ResearcherAgent()关键点解析validate_input是第一道防火墙它拒绝所有非法输入避免后续步骤浪费资源。我们曾因漏掉长度校验被恶意构造的超长query拖垮Redis内存execute_core_logic返回纯dict不包含任何对象引用确保可序列化Redis存储需要handle_failure返回结构化降级结果编排层能据此决定是重试、跳过还是告警而不是抛出异常中断流程。4.3 构建编排引擎让Agent们真正协作起来创建core/orchestrator.pyimport redis import json import logging from transitions import Machine from typing import Dict, Any logger logging.getLogger(__name__) class TaskOrchestrator: def __init__(self, redis_client: redis.Redis): self.redis redis_client # 状态机定义 self.machine Machine( modelself, states[created, assigned, in_progress, pending_review, blocked, completed, archived], initialcreated, transitions[ {trigger: assign_to_agent, source: created, dest: assigned}, {trigger: start_execution, source: assigned, dest: in_progress}, {trigger: complete_task, source: in_progress, dest: pending_review}, {trigger: escalate, source: [in_progress, pending_review], dest: blocked}, {trigger: finalize, source: pending_review, dest: completed}, {trigger: archive, source: completed, dest: archived} ] ) def create_task(self, task_id: str, input_data: Dict[str, Any]) - None: 创建新任务存入Redis task_key ftask:{task_id} task_data { task_id: task_id, input_data: input_data, state: created, created_at: self._now_iso(), history: [] } self.redis.setex(task_key, 3600, json.dumps(task_data)) # 1小时过期 logger.info(fCreated task {task_id}) def assign_task(self, task_id: str, agent_id: str) - None: 分配任务给指定agent task_key ftask:{task_id} task_data json.loads(self.redis.get(task_key)) # 状态检查只能从created分配 if task_data[state] ! created: raise ValueError(fCannot assign task {task_id} from state {task_data[state]}) # 更新状态 task_data[state] assigned task_data[assigned_to] agent_id task_data[assigned_at] self._now_iso() task_data[history].append({event: assigned, agent: agent_id, time: self._now_iso()}) self.redis.setex(task_key, 3600, json.dumps(task_data)) # 发布事件通知agent工作 self.redis.publish(task_events, json.dumps({ task_id: task_id, event: assigned, agent_id: agent_id, input_data: task_data[input_data] })) def _now_iso(self) - str: from datetime import datetime return datetime.utcnow().isoformat() Z # 全局实例 orchestrator TaskOrchestrator(redis.Redis(hostlocalhost, port6379, db0))这个编排引擎的核心价值在于它把“协作”变成了可编程的事件流。当researcher_agent完成工作它不直接调用writer_agent.execute()而是向Redis发布事件# 在researcher_agent的format_output后 self.redis.publish(task_events, json.dumps({ task_id: task_id, event: research_completed, result: formatted_output, next_agent: writer }))编排引擎订阅task_events收到后调用orchestrator.assign_task(task_id, writer)整个过程解耦、异步、可追踪。4.4 治理规则实战用3行YAML堵住一个高危漏洞创建config/governance_rules.yamlrules: - id: prevent_pii_leak trigger: on_agent_output condition: output.data.facts[*].summary contains 身份证 or output.data.facts[*].summary contains 手机号 action: redact_sensitive_info params: {pattern: (\\d{17}[\\dXx]|1[3-9]\\d{9})}对应core/governance.py中的redact_sensitive_info动作import re from core.base_agent import BaseAgent def redact_sensitive_info(output: dict, pattern: str) - dict: 用正则脱敏输出中的敏感信息 def _redact_text(text: str) - str: # 用★替换匹配到的字符保持长度一致便于前端显示 return re.sub(pattern, lambda m: ★ * len(m.group()), text) # 递归处理所有字符串字段 def _redact_dict(d: dict) - dict: for k, v in d.items(): if isinstance(v, str): d[k] _redact_text(v) elif isinstance(v, dict): _redact_dict(v) elif isinstance(v, list): for item in v: if isinstance(item, dict): _redact_dict(item) return d return _redact_dict(output)这个规则在某次测试中拦截了真实风险researcher_agent从某论坛爬取的数据中包含用户发帖“我的身份证号是11010119900307235X”规则自动将其替换为“我的身份证号是★★★★★★★★★★★★★★★★★”避免了下游agent无意中将PII写入日志或返回给用户。实操心得治理规则必须经过“红蓝对抗”测试。我们专门安排一名成员扮演攻击者用各种变体如“身份证11010119900307235X”、“shenfenzheng:11010119900307235X”测试规则覆盖率确保没有绕过方式。5. 常见问题与排查技巧实录那些文档里不会写的血泪教训5.1 问题速查表高频故障现象与根因定位现象可能根因排查命令/步骤解决方案任务卡在in_progress状态不更新Agent进程崩溃未上报Redis连接超时状态更新Lua脚本执行失败redis-cli KEYS task:*查看任务key是否存在redis-cli GET task:xxx检查state字段redis-cli MONITOR观察状态更新命令是否发出在Agent主循环中增加try/except捕获所有异常强制调用orchestrator.update_state(task_id, blocked)为Redis连接配置socket_keepaliveTrue多个Agent同时处理同一任务状态检查非原子化RedisGETSET竞态Pub/Sub消息重复消费redis-cli PUBSUB NUMSUB task_events查看订阅数检查update_state方法是否用Lua脚本用redis.eval(lua_script, 1, key)执行原子操作为Pub/Sub消费者添加ack_timeout和retry_countLLM调用成本远超预期Agent未按治理规则降级工具调用未设rate_limitexecute_core_logic中重复调用LLMgrep -r llm.invoke logs/ | wc -l统计调用次数检查tool_registry注册时的rate_limit参数在BaseAgent.execute_core_logic开头添加self._check_cost_budget()超阈值抛出CostExceededError由handle_failure处理中文输出乱码或格式错乱LLM返回的JSON含中文Unicode转义json.dumps()未设ensure_asciiFalse前端未正确设置charsetcurl -v http://localhost:8000/api/task/xxx查看原始响应头和bodypython -c import json; print(json.dumps({测试: 中文}, ensure_asciiFalse))在所有json.dumps()调用中强制添加ensure_asciiFalseAPI响应头添加Content-Type: application/json; charsetutf-8治理规则不生效YAML缩进错误condition表达式语法错误规则文件未热加载python -c import yaml; print(yaml.safe_load(open(config/governance_rules.yaml)))验证YAML检查watchdog日志是否打印Reloaded rules用yamllint检查YAMLcondition表达式用jsonpath-ng的parse().find()单独测试增加规则加载失败告警5.2 独家避坑技巧来自三次线上事故的总结技巧一给每个Agent加“心跳探针”比日志更早发现问题我们要求每个Agent在execute_core_logic开头和结尾各打一次Redis计数器# 开头 self.redis.incr(fagent:{self.agent_id}:heartbeat) # 结尾 self.redis.decr(fagent:{self.agent_id}:heartbeat)然后用Prometheus抓取agent:researcher:heartbeat指标。如果值持续0说明Agent卡在某处未完成如果值长期为0说明Agent进程已退出。这比等日志报警快5分钟。技巧二用“影子模式”灰度新Agent0风险上线上线reviewer_agentv2.0时我们不直接替换而是让v1.0和v2.0同时接收相同输入v2.0输出不参与主流程只存入shadow_results表用脚本对比v1.0和v2.0输出的confidence_score和issues列表差异差异率1%且v2.0准确率更高时才切流。这避免了某次因v2.0误判“合同有效”为“无效”导致的客户投诉。技巧三为LLM调用加“语义指纹”精准定位幻觉源头当LLM输出错误时传统做法是看prompt。但我们发现相同prompt在不同时间可能产出不同结果。于是我们为每次LLM调用生成语义指纹import hashlib def generate_llm_fingerprint(prompt: str, model_name: str, temperature: float) - str: # 将关键变量哈希忽略无关细节 key f{prompt.strip()[:200]}|{model_name}|{temperature:.2f} return hashlib.md5(key.encode()).hexdigest()[:8]当发现错误输出查日志找llm_fingerprintabcd1234再搜所有该指纹的调用记录发现90%错误都发生在temperature0.8时而temperature0.3时稳定——于是我们把治理规则改为“所有合规类任务强制temperature0.3”。5.3 性能调优实录从30秒到1.2秒的任务耗时优化某客户报告任务平均耗时30秒远超承诺的5秒。我们用cProfile分析发现87%时间花在json.loads()和json.dumps()上。优化步骤第一步替换JSON库pip uninstall ujson # 移除旧版 pip install orjson # 更快的JSON库orjson.dumps()比json.dumps()快3倍且默认ensure_asciiFalse。第二步减少序列化次数原逻辑Agent A输出→json.dumps()→Redis→Agent B读取→json.loads()→处理→json.dumps()→Redis…优化后Agent A输出→orjson.dumps()→RedisAgent B读取→orjson.loads()→处理→**直接传递dict对象给