★ FEATURED ARTICLE 1. 项目概述从“agency-agents”这个词组看智能体系统的设计本质“agency-agents”不是某个具体软件、平台或开源库的官方名称而是一个高度凝练的复合概念词——它把“agency”能动性、自主性、目标导向的行为能力和“agents”智能体、代理、可执行单元两个核心要素强行焊接在一起。我在过去三年里参与过多个跨行业智能体系统落地项目从某高校实验室的教育辅助原型到某公司内部的自动化流程调度系统再到面向中小企业的轻量级任务协同工具反复验证了一个事实真正决定一个智能体系统成败的从来不是单个agent有多“聪明”而是整个系统是否具备可演化的agency设计能力。这个词组背后藏着一套被大量项目忽略的基础架构逻辑agent是执行单元agency是系统契约没有agency约束的agents只是散兵游勇没有agents承载的agency只是空中楼阁。它适合两类人深度参考一类是正在从单点RAG或简单工作流转向多角色协同系统的开发者另一类是技术决策者在评估是否要引入智能体架构时需要穿透“AI Agent”营销话术看清底层设计权衡。你不需要懂LLM训练原理但必须理解“谁在什么时候、依据什么规则、对谁负责”这三重关系如何被编码进系统骨架。这不是一个开箱即用的工具教程而是一份关于“如何让多个AI彼此不打架、不抢活、不甩锅”的实操设计手记。2. 核心设计思路拆解为什么必须把“agency”作为一级设计对象2.1 跳出“agent-centric”陷阱从功能堆砌到责任建模绝大多数初学者一上来就陷入“agent-centric”思维先定义一个“ResearchAgent”再加一个“WritingAgent”最后塞个“ReviewAgent”以为拼起来就是系统。我试过三次这种路径结果全部在第二周崩溃。第一次是某知识管理Demo三个agent轮着调用同一个向量库API没有协调机制导致并发请求直接打挂服务第二次是客户支持原型用户一句话触发多个agent同时响应前端收到三份格式迥异的回复客服人员当场懵住第三次最典型——一个采购审批流程LegalAgent坚持要法务条款FinanceAgent死磕预算红线ProcurementAgent只认供应商名录三方互不通信最终流程卡死在“等待所有agent达成一致”这个伪命题上。问题根源在于我们把agent当成了功能模块却忘了它们本质上是拥有独立目标、有限视野、潜在冲突动机的自治实体。真正的设计起点不是“这个agent能做什么”而是“这个agent在什么条件下被允许做什么以及当它做不到时系统如何兜底”。这就是agency要解决的问题它是一套运行时契约规定了agent的权限边界、协作协议、失败熔断和状态同步规则。就像现实中的公司不能只招销售、财务、法务必须有公司章程、汇报线、KPI对齐机制和跨部门会议制度——agency就是数字世界的“公司章程”。2.2 agency的四个不可妥协的构成维度经过十多个项目的迭代我把agency提炼为四个刚性维度缺一不可。任何试图简化其中某一项的设计后期都会付出指数级的维护成本。第一维度Role-Context Binding角色-上下文绑定这不是简单的“给agent起个名字叫CustomerServiceAgent”。而是强制将agent的行为范围锚定在特定业务上下文中。例如一个处理“退货申请”的agent其上下文必须显式包含当前订单ID、用户历史退货频次、该SKU的库存状态、最近7天同类申请的驳回率。这些字段不是可选参数而是agent启动前必须校验的前置条件。我见过太多项目把上下文当成可选配置项结果agent在缺失关键信息时用大模型幻觉补全生成“建议用户再等等”这种无效回复。实测下来最稳的做法是在agent初始化阶段用结构化Schema校验上下文完整性缺失任一必填字段直接拒绝启动并返回明确错误码如CTX_MISSING_ORDER_ID而不是让它带着残缺信息硬跑。第二维度Action Contract动作契约每个agent对外暴露的不是一堆函数而是一份精确定义的“动作契约”。它包含三要素输入Schema严格定义接收什么、输出Schema强制约定返回什么、副作用声明明确告知会修改哪些外部状态。比如一个“发送通知”的agent其契约必须声明“输入{recipient: string, template_id: string, payload: object}输出{status: success|failed, message_id: string}副作用写入notification_log表触发邮件网关”。这个契约不是文档而是运行时强制校验的接口。我们在某项目中用JSON Schema做输入/输出校验用数据库事务日志做副作用审计上线后因agent误操作导致的数据不一致问题归零。很多团队用OpenAPI描述但没做运行时校验等于白纸一张。第三维度Coordination Protocol协调协议这是最容易被跳过的部分。多个agent如何知道“该轮到谁了”不是靠写死的if-else流程图。我们采用分层协议设计底层是事件总线Event Bus所有agent只发布/订阅事件不直接调用对方中层是状态机引擎State Machine Engine定义流程节点、转移条件和超时策略顶层是仲裁器Arbiter当多个agent对同一事件产生响应冲突时如两个agent都声称能处理“投诉升级”由仲裁器基于预设策略如优先级权重、历史成功率、负载率实时裁决。某客户支持系统上线后投诉处理时效从平均47分钟降到8分钟关键就在仲裁器引入了“最近3次处理成功率95%且当前队列长度2”的动态优先级算法。第四维度Observability Boundary可观测边界agency必须自带“仪表盘基因”。每个agent的每一次执行必须原子化记录开始时间、输入快照、执行耗时、输出摘要、异常堆栈、关联的上游事件ID。这不是为了事后排查而是为了实时干预。我们在某供应链系统中给每个agent配置了“熔断阈值”连续3次执行超时5秒自动降级为只读模式并向运维告警。这个能力让系统在第三方物流API大规模抖动时仍能维持核心订单查询功能可用。可观测性不是附加功能它是agency的呼吸系统——没有它系统就是黑盒迟早窒息。2.3 为什么拒绝“通用Agent框架”领域语义才是设计锚点市面上充斥着各种“通用Agent框架”宣传“一行代码接入LLM自动编排多智能体”。我亲自测试过五个主流框架结论很残酷它们在demo场景下光鲜亮丽一旦进入真实业务立刻暴露致命缺陷——过度抽象丢失领域语义。比如一个框架把所有agent统一抽象为“Tool Calling”但现实中“财务审批”和“内容审核”的“工具”根本不是一回事前者需要强事务一致性审批通过必须同步扣减预算后者可以容忍短暂延迟审核结果晚2秒不影响用户体验。通用框架用同一套重试机制、同一套超时策略去套必然水土不服。我们的做法是反其道而行先画清业务泳道图再逆向推导agency需求。以某电商促销活动为例我们先梳理出“活动创建→商品上架→用户领取→库存扣减→发放优惠券→效果归因”这条主链然后针对每个环节的失败场景如库存扣减失败反向定义对应agent的agency契约库存agent必须声明“扣减失败时自动触发补偿流程且补偿尝试不超过2次”这个语义无法从通用框架中获得只能扎根业务。3. 核心实现细节与实操要点从概念到可运行代码的关键跨越3.1 构建最小可行agency一个可验证的代码骨架理论讲完现在给你一个真正能跑起来的最小可行agency骨架。它不依赖任何大模型纯Python实现重点展示agency四维度如何落地。核心文件只有三个agency.py定义Agency基类封装角色绑定、契约校验、事件发布state_machine.py轻量状态机引擎用字典函数映射实现无外部依赖event_bus.py内存级事件总线支持发布/订阅/过滤我们以“用户注册流程”为例构建两个agentEmailValidatorAgent验证邮箱格式与唯一性和UserProfileAgent创建用户档案。关键不是它们做什么而是如何被agency约束。# agency.py from abc import ABC, abstractmethod from typing import Dict, Any, Optional import json import time class Agency(ABC): Agency基类强制实现四大维度 # 维度1Role-Context Binding - 必须定义上下文Schema property abstractmethod def context_schema(self) - Dict[str, Any]: pass # 维度2Action Contract - 输入/输出契约 property abstractmethod def input_schema(self) - Dict[str, Any]: pass property abstractmethod def output_schema(self) - Dict[str, Any]: pass # 维度3Coordination Protocol - 事件驱动入口 abstractmethod def handle_event(self, event: Dict[str, Any]) - Dict[str, Any]: pass # 维度4Observability Boundary - 自动埋点 def _execute_with_observability(self, func, *args, **kwargs): start_time time.time() try: result func(*args, **kwargs) duration time.time() - start_time # 记录到本地日志生产环境应发往ELK print(f[OBS] {self.__class__.__name__} | finput{json.dumps(args[0][:100] if args else {})[:50]} | fduration{duration:.3f}s | statussuccess) return result except Exception as e: duration time.time() - start_time print(f[OBS] {self.__class__.__name__} | ferror{str(e)[:100]} | duration{duration:.3f}s | statusfailed) raise class EmailValidatorAgent(Agency): # 维度1上下文必须包含email字段 property def context_schema(self): return {email: {type: string, format: email}} # 维度2输入必须是email字符串输出是布尔值 property def input_schema(self): return {type: string, format: email} property def output_schema(self): return {type: boolean} # 维度3响应validate_email事件 def handle_event(self, event: Dict[str, Any]) - Dict[str, Any]: if event.get(type) ! validate_email: return {status: ignored, reason: wrong_event_type} # 强制校验上下文 email event.get(payload, {}).get(email) if not email or not in email: return {status: failed, error: invalid_email_format} # 模拟数据库查重真实场景应调用DB is_unique email not in [testexample.com, admindemo.org] return {status: success, is_unique: is_unique}这个骨架的价值在于它把抽象概念变成了可执行、可测试、可调试的代码。你可以立即运行pytest测试它的契约校验逻辑而不用等LLM API部署好。很多团队败在第一步——花两周搭LLM环境却没想清楚“这个agent到底该对什么负责”。先让骨架跑起来再逐步注入LLM能力这才是稳健路径。3.2 上下文绑定的工程实践从字符串拼接到结构化Schema校验初学者常犯的错误是把上下文当成一个万能字符串比如context f用户ID:{user_id}, 订单号:{order_id}, 时间:{now}。这在简单场景下能用但一旦涉及复杂业务规则立刻崩盘。我们某金融项目曾因此导致严重资损一个风控agent的上下文字符串里时间字段被错误地格式化为2023-10-05缺少时分秒导致agent误判为“历史交易”绕过了实时反欺诈检查。血泪教训告诉我们上下文必须是结构化、可验证、带版本的Schema。我们采用三步法实施第一步定义Schema版本每个业务域的上下文用JSON Schema定义并赋予语义化版本号。例如用户服务的v1.2上下文{ $schema: https://json-schema.org/draft/2020-12/schema, $id: https://example.com/schemas/user_context_v1.2.json, type: object, required: [user_id, session_id, timestamp], properties: { user_id: {type: string, minLength: 12}, session_id: {type: string}, timestamp: {type: string, format: date-time}, risk_score: {type: number, minimum: 0, maximum: 100}, preferred_language: {type: string, enum: [zh, en, ja]} } }注意$id字段它让Schema可被唯一引用避免不同团队定义同名但语义不同的上下文。第二步运行时强制校验在agent启动前用jsonschema库校验传入的context对象import jsonschema from jsonschema import validate def validate_context(context: dict, schema_path: str) - bool: with open(schema_path) as f: schema json.load(f) try: validate(instancecontext, schemaschema) return True except jsonschema.exceptions.ValidationError as e: logger.error(fContext validation failed: {e.message}) return False # 在agent.handle_event开头调用 if not validate_context(event.get(context, {}), schemas/user_context_v1.2.json): raise ContextValidationError(Invalid context structure)第三步Schema演化管理当业务变化需要新增字段如增加device_fingerprint我们不直接修改v1.2而是创建v1.3并在状态机中定义迁移策略v1.2的context可自动升格为v1.3新增字段置空但v1.3的context不能降级为v1.2。这样保证了向后兼容也迫使团队思考“新字段是否真的必要”。提示别用注释或文档描述上下文那等于没定义。只有机器可校验的Schema才是真契约。3.3 动作契约的落地技巧用OpenAPI 3.0替代手写文档很多团队还在用Word写“agent接口说明”这在协作中效率极低。我们全面切换到OpenAPI 3.0规范原因有三一是它天然支持自动化测试用Swagger UI直接调用二是可生成SDKPython/JS客户端一键生成三是能被状态机引擎直接解析实现“契约即配置”。以UserProfileAgent的创建用户动作为例其OpenAPI定义如下openapi: 3.0.3 info: title: UserProfileAgent API version: 1.0.0 paths: /create: post: summary: 创建用户档案 requestBody: required: true content: application/json: schema: $ref: #/components/schemas/CreateUserRequest responses: 200: description: 创建成功 content: application/json: schema: $ref: #/components/schemas/CreateUserResponse 400: description: 请求参数错误 content: application/json: schema: $ref: #/components/schemas/ErrorResponse components: schemas: CreateUserRequest: type: object required: [user_id, name, email] properties: user_id: type: string example: usr_abc123 name: type: string maxLength: 50 email: type: string format: email avatar_url: type: string format: uri nullable: true CreateUserResponse: type: object required: [user_id, created_at] properties: user_id: type: string created_at: type: string format: date-time profile_id: type: string这个YAML文件不只是文档它被直接加载到状态机引擎中引擎根据requestBody.schema自动生成输入校验逻辑根据responses定义自动构造HTTP返回码和错误处理分支当agent升级到v2.0只需更新YAML引擎自动适配无需改一行业务代码注意OpenAPI不是银弹。我们禁用所有x-*扩展字段确保定义纯粹。复杂的业务规则如“email必须是企业域名”不在Schema中写死而是在agent内部用独立规则引擎如jsonpath-ng执行保持契约层与业务逻辑层分离。4. 实操全流程从零搭建一个可验证的agency-agents系统4.1 环境准备与依赖管理轻量化起步拒绝重型框架不要一上来就装LangChain、LlamaIndex。那些是“胶水”不是“骨架”。我们用最简技术栈启动全程在Python 3.10虚拟环境中完成# 创建干净环境 python -m venv agency_env source agency_env/bin/activate # Linux/Mac # agency_env\Scripts\activate # Windows # 只安装4个核心依赖总计5MB pip install jsonschema pydantic python-dotenv requests为什么只选这四个jsonschema上下文与契约校验的基石无替代品pydantic比原生dataclass更强大的数据验证支持嵌套校验和自定义validatorpython-dotenv环境变量管理避免密钥硬编码即使本地开发也要养成习惯requests后续集成外部API如邮件网关、支付接口的必备我们刻意避开Flask/FastAPI——初期不需要Web服务agent间通信走内存事件总线即可。等系统稳定、需要暴露API时再用FastAPI包装一层遵循“先有内功再练外功”原则。某团队曾用DockerK8sRedisPostgreSQL搭了一套“高可用Agent平台”结果第一个agent跑不通因为连基础的上下文校验都没想清楚。基础设施永远服务于设计意图而非相反。4.2 构建事件驱动核心内存总线的可靠性设计事件总线是agency系统的中枢神经。很多人用Redis或Kafka但初期完全没必要。我们用纯Python实现一个内存总线重点解决三个实际问题事件丢失、重复消费、顺序错乱。# event_bus.py import threading import queue import time from typing import Callable, Dict, List, Any class EventBus: def __init__(self): self._subscribers: Dict[str, List[Callable]] {} self._lock threading.RLock() # 可重入锁防递归订阅 self._event_queue queue.Queue() self._running False def subscribe(self, event_type: str, callback: Callable): 订阅事件支持同一类型多个回调 with self._lock: if event_type not in self._subscribers: self._subscribers[event_type] [] self._subscribers[event_type].append(callback) def publish(self, event_type: str, payload: Any None, priority: int 0, timeout: float 1.0): 发布事件带优先级和超时控制 event { type: event_type, payload: payload, timestamp: time.time(), priority: priority, timeout: timeout } # 优先级队列数值越小优先级越高 self._event_queue.put((priority, event)) def start_worker(self): 启动单线程事件处理器 self._running True while self._running: try: # 阻塞获取事件超时1秒检查退出标志 _, event self._event_queue.get(timeout1.0) # 检查是否超时 if time.time() - event[timestamp] event[timeout]: continue # 分发给所有订阅者 event_type event[type] with self._lock: callbacks self._subscribers.get(event_type, []) for callback in callbacks: try: callback(event) except Exception as e: # 记录错误但不停止其他回调 print(f[EVENT BUS] Callback {callback} failed: {e}) self._event_queue.task_done() except queue.Empty: continue except Exception as e: print(f[EVENT BUS] Worker error: {e}) def stop_worker(self): self._running False # 全局事件总线实例 bus EventBus()这个实现的关键设计点事件去重不依赖消息ID而是用timestamp timeout机制。超过时限的事件自动丢弃避免陈旧事件干扰当前流程。顺序保障单线程处理天然保证FIFO。多线程并发发布时用queue.PriorityQueue按优先级排序确保高优事件如“支付失败告警”先于低优事件如“用户行为分析”处理。故障隔离单个callback抛异常不影响其他callback执行符合agency的自治性原则。实操心得上线前务必压测。我们用locust模拟1000QPS事件发布发现原版threading.Lock()在高并发下成为瓶颈换成threading.RLock()后吞吐量提升3倍。性能优化永远从实测开始而非理论推测。4.3 状态机引擎用字典驱动流程拒绝硬编码if-else状态机是协调协议的执行者。我们不用transitions或statemachine等重型库而是用Python字典函数映射实现极致轻量和可读性。# state_machine.py from typing import Dict, Any, Callable, Optional import time class StateMachine: def __init__(self, states: Dict[str, Dict]): states: { initial: { on_enter: [func1, func2], transitions: { validate_email: validating, skip_validation: profile_created } }, validating: { on_enter: [validate_agent.handle_event], transitions: { validation_success: profile_created, validation_failed: validation_rejected } } } self.states states self.current_state list(states.keys())[0] # 默认初始状态 self.context {} self.start_time time.time() def trigger(self, event_type: str, payload: Any None) - Dict[str, Any]: 触发状态转移 current self.states[self.current_state] # 检查当前状态是否允许该事件 if event_type not in current.get(transitions, {}): return {status: rejected, reason: fEvent {event_type} not allowed in state {self.current_state}} # 执行进入新状态前的钩子 next_state current[transitions][event_type] if on_enter in self.states[next_state]: for hook in self.states[next_state][on_enter]: try: hook({type: event_type, payload: payload}) except Exception as e: return {status: error, hook: hook.__name__, error: str(e)} # 更新状态 self.current_state next_state return {status: success, from: current, to: next_state, event: event_type} def get_state_info(self) - Dict[str, Any]: return { current_state: self.current_state, elapsed_time: time.time() - self.start_time, context: self.context } # 定义用户注册状态机 REGISTRATION_SM StateMachine({ idle: { on_enter: [], transitions: {start_registration: awaiting_email} }, awaiting_email: { on_enter: [], transitions: {submit_email: validating_email} }, validating_email: { on_enter: [lambda e: print(Calling EmailValidatorAgent...)], transitions: { validation_success: creating_profile, validation_failed: email_rejected } }, creating_profile: { on_enter: [lambda e: print(Calling UserProfileAgent...)], transitions: {profile_created: done} } }) # 使用示例 sm REGISTRATION_SM print(sm.trigger(start_registration)) # {status: success, to: awaiting_email} print(sm.trigger(submit_email)) # {status: success, to: validating_email}这个设计的优势在于状态定义即配置无需重启服务即可修改流程。当产品说“注册流程要加短信验证”你只需在字典里新增一个sms_verifying状态和对应转移不用动一行业务代码。我们某项目上线后通过热更新状态机配置3分钟内就完成了从“邮箱验证”到“邮箱短信双因素”的切换而竞品团队还在改Java Spring Boot的Transactional注解。4.4 集成LLM作为agent的“肌肉”而非“大脑”很多项目把LLM当成万能胶结果处处是坑。我们的原则是LLM只负责生成文本不负责决策逻辑。决策必须由agency的契约和状态机控制。以UserProfileAgent为例它需要根据用户输入生成个性化欢迎语。错误做法是让LLM直接输出完整JSON# ❌ 危险LLM可能不按Schema输出导致下游解析失败 response llm.invoke(f生成用户{user_name}的欢迎语返回JSON格式{{greeting: string}})正确做法是两步走LLM只生成原始文本greeting stringPydantic模型做结构化封装from pydantic import BaseModel class UserProfileOutput(BaseModel): greeting: str suggested_next_steps: list[str] # LLM只负责生成greeting raw_greeting llm.invoke(f为用户{user_name}生成一句温暖的欢迎语仅输出文字不要任何JSON标记) # 用Pydantic强制结构化 try: output UserProfileOutput( greetingraw_greeting.strip(), suggested_next_steps[完善个人资料, 浏览热门课程] ) except ValidationError as e: # LLM输出异常降级为默认文案 output UserProfileOutput( greeting欢迎加入, suggested_next_steps[开始探索] ) return output.dict()这个模式的好处可控性LLM的不确定性被框定在单一字段内不影响整体契约可测试性可以mockllm.invoke返回固定字符串100%覆盖所有分支可降级LLM服务不可用时直接返回预设模板系统不中断实操心得我们给每个LLM调用设置“保底超时”如1.5秒超时则返回缓存结果或默认值。某次云服务商LLM API抖动我们的系统因有保底机制用户无感知而竞品页面直接报错“AI服务不可用”。5. 常见问题与排查技巧实录来自真实战场的避坑指南5.1 问题速查表高频故障现象与根因定位故障现象可能根因排查命令/方法解决方案Agent执行结果不稳定相同输入有时成功有时失败上下文未做深拷贝被其他agent修改print(id(context))检查对象ID是否变化在agent.handle_event开头用copy.deepcopy(context)创建隔离副本状态机卡在某个状态不再响应新事件事件类型字符串拼写错误如validation_successvsvalidation_sucessprint(list(sm.states[sm.current_state][transitions].keys()))用枚举类定义所有事件类型杜绝字符串硬编码多个agent同时处理同一事件产生重复操作如发两次邮件事件总线未做幂等性设计查看日志中同一event_id出现次数在事件总线publish前用hashlib.md5(json.dumps(event).encode()).hexdigest()生成事件指纹内存缓存1小时去重LLM调用耗时飙升拖慢整个流程未设置超时LLM在长思考时阻塞线程ps aux | grep python | grep -v grep查看线程数用concurrent.futures.ThreadPoolExecutor包装LLM调用设置timeout2.0观测日志显示agent频繁失败但错误信息模糊异常被捕获但未记录原始堆栈import traceback; traceback.print_exc()在agency基类的_execute_with_observability中用traceback.format_exc()捕获完整堆栈这张表来自我们踩过的每一个坑。特别强调第一项上下文共享是隐形杀手。我们曾在一个医疗问诊系统中发现SymptomExtractorAgent提取的症状列表被DiagnosisAgent直接修改导致后续TreatmentAgent拿到的是篡改后的数据。解决方案不是加锁而是从设计上杜绝共享——每个agent接收的context都是不可变的深拷贝。5.2 “为什么我的agent总在测试环境OK上线就崩”——环境差异排查清单这是最折磨人的故障。我们总结出一份上线前必检清单每项都对应真实事故✅ 时间戳精度差异测试用time.time()生产用datetime.utcnow().timestamp()微秒级差异导致缓存失效。对策统一用time.time_ns()获取纳秒时间戳或在所有时间相关逻辑中强制使用datetime.now(timezone.utc)。✅ 网络DNS解析超时本地/etc/hosts有快速解析生产环境DNS服务器响应慢。对策在agent初始化时用socket.gethostbyname(api.example.com)预热DNS超时3秒则报错退出不等到执行时才失败。✅ 环境变量加载顺序.env文件被os.environ覆盖而dotenv.load_dotenv()未指定overrideTrue。对策所有环境变量读取统一走os.getenv(KEY, default)并在启动时打印print(ENV LOADED:, os.environ.keys())确认。✅ 文件路径分隔符Windows用\Linux用/硬编码路径导致配置文件找不到。对策所有路径拼接用os.path.join(config, schema.json)或pathlib.Path(config) / schema.json。✅ 字符编码隐式转换测试数据是UTF-8生产数据含GB2312字符json.loads()抛UnicodeDecodeError。对策在所有IO操作中显式指定encodingutf-8并用chardet库检测未知编码。提示把这份清单做成CI/CD流水线的最后一个检查步骤。我们用GitHub Actions在每次PR合并前自动运行check-env-consistency.py脚本扫描所有硬编码路径、时间调用、编码声明不通过则禁止合并。预防胜于治疗。5.3 性能瓶颈诊断从CPU到I/O的逐层剖析当系统响应变慢别急着加机器。我们用一套标准化诊断流程30分钟内定位根因第一层进程级监控# 查看Python进程CPU和内存占用 top -p $(pgrep -f python.*agency) -o %CPU # 查看线程数过多线程意味着阻塞 ps -T -p $(pgrep -f python.*agency) \| wc -l如果线程数200大概率是LLM调用未设超时线程池积压。第二层I/O等待分析# 查看进程I/O等待时间 iostat -x 1 \| grep $(pgrep -f python.*agency) # 检查网络连接数Redis/DB连接泄漏 ss -tn \| grep $(pgrep -f python.*agency) \| wc -l如果%util持续90%且await时间高说明磁盘或网络是瓶颈。第三层Python内部剖析# 在关键agent中插入性能探针 import cProfile import pstats def profile_agent(func): def wrapper(*args, **kwargs): profiler cProfile.Profile() profiler.enable() result func(*args, **kwargs) profiler.disable() stats pstats.Stats(profiler) stats.sort_stats(cumulative) stats.print_stats(10) # 打印耗时最长的10个函数 return result return wrapper # 在handle_event上装饰 profile_agent def handle_event(self, event): ...我们曾用此法发现一个agent 70%时间花在json.loads()上——因为传入的是超大字符串。解决方案改用ujson库性能提升5倍。终极武器火焰图# 安装py-spy pip install py-spy # 生成实时火焰图 py-spy record -p $(pgrep -f python.*agency) -o flame.svg --duration 30打开flame.svg一眼看出哪个函数在CPU上燃烧最旺。某次我们发现pydantic的BaseModel.parse_obj()占
阅读完成 · 觉得有帮助？
