智能体系统工程：从Demo到生产级Agent的落地方法论

1. 这不是AI工程师的升级路线图而是一套可落地的“智能体系统工程”方法论“Agentic Engineering Roadmap”这个标题乍看像又一个技术 buzzword 堆砌——毕竟这两年“agent”“agentic”“autonomous agent”被用得太多从大厂技术白皮书到创业公司BP几乎成了标配词汇。但真正沉下心拆解过几十个生产级智能体项目后我越来越确信当前行业最缺的不是新模型、不是更长上下文而是能把“智能体”从Demo变成产品、从单点能力变成可维护系统的能力。这正是“Agentic Engineering”要解决的核心问题——它不是教你怎么调用LLM API写个自动订咖啡的bot而是告诉你当你要构建一个能跨12个内部系统调度资源、在SLA约束下自主决策、出错时可追溯可回滚、上线后能持续演化的智能体集群时该从哪下手、每一步踩什么坑、哪些设计今天省事明天要重写。我带团队做过三个典型场景金融风控中的实时反欺诈策略协同体需对接核心交易系统规则引擎人工审核流、制造业设备预测性维护调度体联动IoT平台工单系统备件库存API维修知识库、生物医药临床试验患者随访体集成EMR结构化数据非结构化医患对话合规审计日志。这些项目共同暴露出一个事实90%以上的失败不源于模型能力不足而源于工程链路断裂——比如没有定义清晰的Agent边界导致状态混乱比如缺乏可观测性设计让一次超时错误排查耗时三天比如把所有逻辑硬编码进prompt导致策略变更必须全量回归测试。这份Roadmap就是我们用三年、十二个项目、上百次推倒重来沉淀下来的“智能体系统工程”实操手册。它不讲大道理只列具体动作什么时候该建State Schema为什么Action必须带幂等ID如何设计能让业务方直接修改的决策规则DSL以及最关键的——怎么让一个智能体在不重启服务的前提下安全地切换底层模型或更新工具集。如果你正卡在“demo很炫、上线就崩”的阶段或者团队里算法和工程同学还在为“谁该负责Agent的稳定性”扯皮那接下来的内容就是你真正需要的施工图纸。2. 内容整体设计与思路拆解为什么必须抛弃“LLM应用开发”思维2.1 从“应用层封装”到“系统层架构”的范式迁移传统LLM应用开发如RAG问答、摘要生成本质是单向函数调用输入Query → LLM处理 → 输出Response。整个链路是线性的、无状态的、可预测的。而Agentic Engineering面对的是闭环反馈系统Agent接收Observation → 执行Action → 改变Environment状态 → 触发新Observation → 进入下一轮决策。这个差异决定了工程设计的根本转向状态管理不可回避单次调用无需关心状态但Agent必须持续维护自身状态当前目标、已执行步骤、缓存中间结果、失败重试计数。我们曾在一个供应链调度项目中因未持久化Agent的“已锁定库存量”状态导致并发请求重复扣减同一库存引发资损。最终方案是引入轻量级状态机基于Redis Hash每个Agent实例绑定唯一Session ID所有状态变更通过原子操作更新。环境交互必须契约化传统API调用只需关注输入输出格式而Agent的Action可能触发真实世界变更如创建工单、发送邮件、调整服务器配置。这就要求每个Tool必须明确定义前置条件Precondition和后置效应Postcondition。例如“创建Jira工单”Tool的Precondition是“用户有对应项目权限”Postcondition是“返回工单URL且状态为Open”。我们在医疗项目中强制所有Tool实现validate_preconditions()和describe_postcondition()方法并在Agent执行前自动校验避免因权限缺失导致流程中断却无明确报错。可观测性是第一性需求LLM应用的日志通常是“输入/输出对”而Agent系统需要追踪决策链Decision Trace为什么选这个Action依据哪些Observation拒绝了哪些候选Action耗时分布在哪我们在金融项目中设计了三层日志L1基础日志时间戳Agent IDStep ID、L2决策日志包含prompt模板、选择的tool name、confidence score、L3调试日志完整prompt内容、所有tool call参数、原始API响应。关键不是记录多少而是确保任意一次异常都能在5分钟内定位到具体决策节点。提示不要试图用现有APM工具如Datadog直接监控Agent。它们擅长追踪HTTP请求链路但无法理解“Agent选择Tool A而非B”的决策逻辑。必须自建决策追踪层这是Agentic Engineering的基础设施。2.2 Roadmap的四阶段演进逻辑从单体Agent到自治系统这份Roadmap不是线性学习路径而是按系统复杂度划分的四个能力域每个域解决特定维度的工程挑战Stage 1Controlled Agent受控智能体核心目标让单个Agent在预设边界内稳定运行。重点解决确定性执行问题——给定相同输入和环境状态必须产生可复现的行为。关键技术点Prompt工程标准化使用JSON Schema约束输出、Tool调用强类型校验、超时熔断机制。这是我们所有项目的起点也是验证业务逻辑正确性的基石。Stage 2Composable Agent可组合智能体核心目标将多个Agent按职责拆分并安全协作。重点解决边界隔离问题——避免Agent间状态污染、循环调用、责任模糊。关键技术点定义清晰的Agent接口Input Schema / Output Schema / SLA承诺、引入Message Broker如NATS解耦通信、设计跨Agent事务协调器如Saga模式处理分布式状态变更。Stage 3Resilient Agent弹性智能体核心目标在环境波动API抖动、模型降级、网络分区下维持核心功能。重点解决故障自愈问题——不是简单重试而是动态降级、策略切换、状态补偿。关键技术点多模型路由根据query复杂度/延迟要求选择GPT-4o或Claude-3-haiku、Fallback Tool链当主Tool失败时自动启用备用方案、状态快照与回滚定期保存关键决策点异常时恢复至最近一致状态。Stage 4Evolvable Agent可演化智能体核心目标支持业务策略、工具集、模型能力的热更新无需停机。重点解决在线演进问题——让系统具备“生长”能力。关键技术点规则引擎嵌入将业务策略从prompt中剥离存于独立规则库、Tool Registry动态注册/注销、模型版本灰度发布按流量比例分配不同模型实例。这四个阶段不是严格先后关系而是能力矩阵。一个成熟系统往往同时具备Stage 2的组合能力和Stage 3的弹性能力。Roadmap的价值在于帮你诊断当前瓶颈如果团队总在争论“这个Agent该不该调用那个API”说明还卡在Stage 1如果每次新增一个Agent就要重构整个通信协议那就是Stage 2的缺失。2.3 为什么拒绝“通用Agent框架”我们的技术选型哲学市面上已有LangChain、LlamaIndex、Semantic Kernel等成熟框架但我们坚持“最小框架原则”只封装重复出现三次以上的模式其余全部手写。原因很现实性能损耗不可接受某次压测发现LangChain的CallbackHandler在高并发下增加平均延迟120ms而我们的核心风控Agent SLA要求端到端800ms。最终我们用原生asyncio重写了事件回调延迟降至17ms。调试成本指数级上升当Agent行为异常时框架抽象层会掩盖真实问题。我们曾为排查一个“Tool调用参数为空”的bug在LangChain源码中追踪了7层装饰器耗时两天。而手写代码中问题直接暴露在tool_call_params self._build_params(observation)这一行。定制化需求无法满足医疗项目要求所有决策必须附带合规依据如“根据《XX指南》第3.2条”这需要深度修改prompt生成逻辑。框架的插件机制要么不支持要么改造后失去可维护性。因此我们的技术栈是“乐高式组合”核心运行时Python 3.11 asyncio极致控制协程生命周期状态管理Redis高性能 SQLite本地开发轻量替代消息通信NATS低延迟、内置流式处理比Kafka更适合Agent间短消息可观测性自研DecisionTrace SDK结构化日志决策图谱可视化部署Docker Kubernetes StatefulSet保障Agent Session亲和性这不是标新立异而是每个选择都来自血泪教训。当你在深夜收到告警说“Agent集群CPU飙升至95%”你会感激自己没用那些“开箱即用”的黑盒框架。3. 核心细节解析与实操要点从概念到代码的关键跨越3.1 Agent状态Schema设计别让JSON成为你的技术债很多团队把Agent状态简单定义为{goal: str, history: List[Dict]}这在Demo阶段可行但上线后必然崩溃。真正的状态Schema必须回答三个问题状态是否可序列化是否可并发安全是否可增量更新我们采用三段式状态设计# 示例设备维护调度Agent的状态Schema class MaintenanceAgentState(BaseModel): # 【必需】唯一标识与元信息用于路由和审计 session_id: str Field(..., description全局唯一会话ID由外部系统生成) created_at: datetime Field(default_factorydatetime.utcnow) last_updated: datetime Field(default_factorydatetime.utcnow) # 【核心】业务状态必须可序列化且无引用环 current_goal: str Field(..., description当前主目标如完成A320发动机更换) target_equipment: EquipmentRef Field(..., description目标设备引用) scheduled_maintenance: Optional[MaintenancePlan] None # 【关键】执行上下文避免重复计算但必须可控 context_cache: Dict[str, Any] Field( default_factorydict, description临时缓存如最近获取的备件库存快照需设置TTL ) # 【防御】失败管理显式记录而非隐式重试 failure_history: List[FailureRecord] Field(default_factorylist) max_retries: int Field(default3, description全局最大重试次数) # 【演进】策略版本支持热更新 policy_version: str Field(defaultv1.0, description当前生效的业务策略版本)为什么这样设计session_id强制外部系统生成避免Agent自建ID导致分布式冲突。我们要求前端调用时必须传入trace_id后端直接复用保证全链路可追溯。context_cache显式声明TTL通过Redis key过期实现而不是用lru_cache——后者在多进程部署下完全失效且无法控制缓存大小。failure_history不仅记录错误还包含recovery_action字段如已切换至备用供应商API这是Stage 3弹性的基础。policy_version是Stage 4演化的钥匙。当业务方更新策略时只需修改数据库中该session_id对应的versionAgent下次决策时自动加载新规则。注意永远不要在状态中存函数对象、数据库连接、文件句柄等不可序列化对象。我们曾因在state中存了一个requests.Session实例导致Redis序列化失败整个集群雪崩。现在所有状态类都强制继承BaseModel并开启arbitrary_types_allowedFalse校验。3.2 Tool接口契约让Agent学会“先思考再行动”Tool不是API包装器而是Agent的“手脚”。它的设计质量直接决定系统鲁棒性。我们定义Tool必须实现的四大契约契约类型强制要求实操示例为什么重要输入校验必须提供validate_input()方法返回ValidationResultvalidate_input({part_id: ABC123, quantity: 5})→ValidorInvalid(quantity must be 0)避免无效请求打到下游系统造成脏数据或资损前置条件必须声明preconditions: List[str]描述执行前必须满足的环境状态[inventory_service_available, user_has_permission_to_order]Agent可在执行前快速检查避免调用后才发现权限不足幂等性所有写操作必须支持idempotency_key参数且相同key多次调用返回相同结果create_work_order(idempotency_keysess_abc123_step4)→ 第二次调用返回首次创建的工单ID网络超时重试时防止重复创建工单、重复扣款可观测性必须返回ExecutionReport包含耗时、状态码、关键指标{duration_ms: 245, http_status: 201, items_created: 1}为决策优化提供数据支撑如某Tool平均耗时1s则降级一个典型Tool实现class CreateWorkOrderTool(Tool): def __init__(self, inventory_client: InventoryClient): self.inventory_client inventory_client property def name(self) - str: return create_work_order property def description(self) - str: return Create a maintenance work order in the CMMS system property def preconditions(self) - List[str]: return [cmms_system_healthy, user_authenticated] def validate_input(self, input_dict: Dict) - ValidationResult: if not isinstance(input_dict.get(equipment_id), str): return ValidationResult.invalid(equipment_id must be string) if not (1 len(input_dict.get(description, )) 500): return ValidationResult.invalid(description length must be 1-500 chars) return ValidationResult.valid() async def execute(self, input_dict: Dict, idempotency_key: str) - ExecutionReport: # 关键所有写操作必须包裹在idempotency_key检查中 existing await self._check_idempotency(idempotency_key) if existing: return ExecutionReport.success(existing, duration_ms5) # 执行真实业务逻辑 start time.time() order await self.cmms_client.create_order( equipment_idinput_dict[equipment_id], descriptioninput_dict[description], priorityinput_dict.get(priority, medium) ) duration (time.time() - start) * 1000 # 记录idempotency key await self._record_idempotency(idempotency_key, order.id) return ExecutionReport.success( {order_id: order.id, url: order.url}, duration_msround(duration), http_status201 )实操心得别急着写execute逻辑先花30分钟写完validate_input和preconditions。我们统计过72%的线上故障源于输入校验缺失。某次因未校验equipment_id长度导致CMMS系统SQL注入整个维护队列瘫痪4小时。3.3 决策链Decision Trace设计让黑盒决策变得可解释LLM的“幻觉”无法根除但我们可以让幻觉发生的位置变得透明。Decision Trace不是日志而是决策过程的结构化快照。我们要求每个Agent Step必须生成以下字段字段类型示例值用途step_idUUIDd8e6f2a1-...全局唯一用于跨系统追踪agent_idstrmaintenance-scheduler-v2标识Agent类型与版本input_observationJSON{current_inventory: {...}, pending_orders: [...]}决策依据的原始数据prompt_template_hashstrsha256:abc123...精确关联到使用的prompt版本selected_actiondict{tool: create_work_order, params: {...}}最终执行的动作candidate_actionsList[dict][{tool: create_work_order, score: 0.92}, ...]所有候选动作及置信度reasoning_tracestrInventory low (12 threshold 20), need urgent orderLLM生成的决策理由用于人工审计execution_resultdict{status: success, output: {...}}Action执行结果关键设计点Prompt Template Hash不是记录prompt全文太长而是对模板字符串做hash。当业务方修改prompt时hash变化系统自动告警“检测到prompt变更建议回归测试”。Candidate Actions强制Agent生成Top-3候选动作及分数。这不仅是调试用更是Stage 3弹性的基础——当首选动作失败时可立即尝试第二候选。Reasoning Trace要求LLM在prompt中明确指令“用中文一句话说明选择此Action的核心原因”并提取该句。这比完整response更易读且能快速定位逻辑偏差如“因库存充足所以不订货” vs 实际库存已告急。我们用NATS JetStream持久化所有Decision Trace保留30天。运维同学用一条SQL就能查出“过去1小时所有因‘库存服务不可用’失败的决策”然后精准定位到故障模块。4. 实操过程与核心环节实现从零搭建一个可上线的Maintenance Agent4.1 环境准备与依赖安装避开Python生态的深坑别跳过这一步。Python的异步生态看似简单实则暗礁密布。我们的标准环境配置# 使用pyenv管理Python版本避免系统Python污染 pyenv install 3.11.9 pyenv local 3.11.9 # 创建隔离环境virtualenv比conda更轻量适合容器部署 python -m venv .venv source .venv/bin/activate # 安装核心依赖注意版本锁死 pip install --upgrade pip pip install \ redis4.6.0 \ # Redis客户端4.6.0修复了async连接池bug \ nats-py2.10.0 \ # NATS客户端2.10.0支持JetStream流式消费 \ pydantic2.6.4 \ # 数据验证2.6.4修复了BaseModel deepcopy性能问题 \ httpx0.24.1 \ # HTTP客户端比aiohttp更简洁0.24.1修复了SSL证书验证 \ tenacity8.2.3 \ # 重试库8.2.3支持异步retry_async \ orjson3.9.14 \ # 超快JSON序列化比json快5倍内存占用少40% \ prometheus-client0.17.1 # 指标采集0.17.1支持async metrics为什么锁死版本某次升级nats-py到2.11.0其内部连接池逻辑变更导致Agent在高并发下随机断连排查耗时3天。现在所有项目都用requirements.txt精确锁定CI流水线强制校验。4.2 构建第一个Controlled Agent设备巡检报告生成器目标接收设备传感器数据生成结构化巡检报告并识别异常项。Step 1定义输入Observation Schemaclass SensorObservation(BaseModel): equipment_id: str timestamp: datetime temperature: float Field(..., ge-50, le150, description摄氏度) vibration_rms: float Field(..., ge0, le10, descriptionmm/s RMS) pressure: float Field(..., ge0, le1000, descriptionbar) status_code: Literal[OK, WARNING, CRITICAL] field_validator(temperature, vibration_rms, pressure) def validate_numeric_range(cls, v, info): # 自定义校验逻辑 if info.field_name temperature and v -40: raise ValueError(temperature too low for sensor calibration) return vStep 2编写Prompt TemplateJSON Schema约束输出PROMPT_TEMPLATE 你是一个专业的设备巡检工程师。请根据以下传感器数据生成一份结构化巡检报告。 严格按JSON Schema输出不要任何额外文本 传感器数据 {observation_json} JSON Schema: {schema_json} 特别注意 - anomalies数组中每个元素必须包含metric指标名、value实际值、threshold阈值、severity严重等级LOW/MEDIUM/HIGH - recommendations必须具体可执行如清洁散热片而非检查设备 - 如果所有指标正常anomalies为空数组recommendations写无 Step 3实现Agent核心逻辑class InspectionAgent: def __init__(self, llm_client: AsyncLLMClient, schema: Type[BaseModel]): self.llm_client llm_client self.schema schema self.schema_json json.dumps(schema.model_json_schema(), indent2) async def run(self, observation: SensorObservation) - Dict: # 1. 构建prompt observation_json observation.model_dump_json(indent2) prompt PROMPT_TEMPLATE.format( observation_jsonobservation_json, schema_jsonself.schema_json ) # 2. 调用LLM带超时和重试 try: response await tenacity.AsyncRetrying( stoptenacity.stop_after_attempt(3), waittenacity.wait_exponential(multiplier1, min1, max10), reraiseTrue ).astart(self._call_llm_with_timeout, prompt) # 3. 解析JSON用orjson比json快3倍 parsed orjson.loads(response) # 4. 用Pydantic校验结构关键 validated self.schema.model_validate(parsed) return validated.model_dump() except orjson.JSONDecodeError as e: raise RuntimeError(fLLM output not valid JSON: {e}) except ValidationError as e: raise RuntimeError(fLLM output violates schema: {e}) async def _call_llm_with_timeout(self, prompt: str) - str: # 实际调用OpenAI/Gemini等API timeout aiohttp.ClientTimeout(total15) async with aiohttp.ClientSession(timeouttimeout) as session: async with session.post( https://api.openai.com/v1/chat/completions, headers{Authorization: fBearer {self.api_key}}, json{ model: gpt-4o-mini, messages: [{role: user, content: prompt}], response_format: {type: json_object} } ) as resp: if resp.status ! 200: raise RuntimeError(fLLM API error: {resp.status}) data await resp.json() return data[choices][0][message][content]Step 4添加Decision Trace记录async def run_with_trace(self, observation: SensorObservation) - Dict: step_id str(uuid4()) start_time time.time() # 记录输入 trace_input { step_id: step_id, agent_id: inspection-v1, input_observation: observation.model_dump(), prompt_template_hash: hashlib.sha256(PROMPT_TEMPLATE.encode()).hexdigest()[:8] } try: result await self.run(observation) # 记录成功结果 trace_output { status: success, output: result, duration_ms: round((time.time() - start_time) * 1000), step_id: step_id } # 发送到NATS流 await self.nats_jetstream.publish(decision.trace, orjson.dumps(trace_output)) return result except Exception as e: # 记录失败 trace_error { status: error, error_type: type(e).__name__, error_message: str(e), duration_ms: round((time.time() - start_time) * 1000), step_id: step_id } await self.nats_jetstream.publish(decision.trace, orjson.dumps(trace_error)) raise实测效果单次推理平均耗时320ms含网络延迟Schema校验失败率0.02%主要因LLM模型波动决策Trace完整率100%NATS JetStream持久化保障这个Controlled Agent已稳定运行在产线每天处理2.3万次巡检。它证明了最简单的Agent只要工程扎实就是最可靠的Agent。4.3 进阶构建Composable Agent集群——巡检调度双Agent协同当单Agent无法覆盖全场景就需要组合。以“发现异常→生成工单→调度维修”为例架构设计InspectionAgent专注分析输出{anomalies: [...], urgency: HIGH}SchedulerAgent专注调度输入{equipment_id, anomaly, urgency}输出{assigned_to: tech_001, scheduled_at: 2024-06-15T14:00Z}通信方式InspectionAgent处理完向NATS主题maintenance.request发布消息SchedulerAgent订阅该主题处理后向maintenance.scheduled发布结果。关键实现消息Schema定义class MaintenanceRequest(BaseModel): request_id: str Field(default_factorylambda: str(uuid4())) equipment_id: str anomaly: str # 如vibration_rms_exceed_threshold urgency: Literal[LOW, MEDIUM, HIGH] inspection_step_id: str # 关联原始决策用于溯源 created_at: datetime Field(default_factorydatetime.utcnow) class MaintenanceScheduled(BaseModel): request_id: str assigned_to: str scheduled_at: datetime estimated_duration_hours: float scheduler_step_id: str # 关联原始request形成完整链路 inspection_step_id: strSchedulerAgent的防错设计去重收到request_id已存在直接返回缓存结果用Redis Set记录已处理ID超时NATS消费消息时设置ack_wait30s超时未ack则重发避免单点故障丢失死信连续3次处理失败的消息自动转入maintenance.dead_letter主题供人工干预协同效果端到端SLA从传感器数据上报到维修工单生成平均耗时90秒故障率0.17%主要因CMMS系统临时不可用触发Fallback到邮件通知可观测性通过inspection_step_id和scheduler_step_id可在Kibana中一键查看完整决策链这验证了Stage 2的核心价值组合不是功能叠加而是通过契约化接口让复杂系统保持可理解、可测试、可替换。5. 常见问题与排查技巧实录那些文档里不会写的坑5.1 “Agent突然不工作了但日志全是200”——决策链断裂的隐形杀手现象Agent服务健康检查通过API返回200但业务方反馈“工单没生成”。查看日志只有INFO: Agent started无错误。排查路径先查Decision Trace在NATS JetStream中搜索agent_id:scheduler-v2 AND status:success发现最近1小时无记录 → 问题不在执行而在触发。查上游消息搜索subject:maintenance.request发现消息积压 → InspectionAgent没发消息。查InspectionAgent日志发现大量WARNING: Skipping report generation - no anomalies detected→ 原来传感器阈值被运维误调高所有数据都“正常”了。根本原因Agent的“成功”定义被窄化。我们只监控HTTP状态码却忽略了业务语义成功如“生成工单”。解决方案在Decision Trace中强制添加business_outcome字段{type: no_anomaly_found, details: all_metrics_within_normal_range}Prometheus指标新增agent_business_outcome_total{outcomeno_anomaly_found}设置告警“连续5分钟no_anomaly_found 0”运维大盘增加“业务决策分布图”一眼看出异常模式实操心得永远不要相信“HTTP 200 业务成功”。我们后来在所有Agent的run()方法末尾加了一行self.metrics.inc(business_outcome, outcome_type)哪怕outcome是no_action_taken。5.2 “为什么同样的输入两次结果不一样”——LLM非确定性的工程应对现象测试时输入相同传感器数据第一次输出{anomalies: [...]}第二次输出{anomalies: []}无报错。原因分析LLM的temperature参数默认0.7引入随机性模型服务端负载高时token采样策略可能微调Prompt中未强制要求确定性输出如未加“请严格按JSON Schema输出不要任何额外字符”工程化解方案非调参强制确定性Prompt在所有Prompt末尾添加固定指令请严格按上述JSON Schema输出不要任何额外文本、空格、换行符。如果无法确定请输出空数组。后置校验重试对LLM输出做两次校验第一次orjson.loads()→ 失败则重试最多2次第二次pydantic.validate()→ 失败则触发Fallback如调用规则引擎缓存确定性结果对相同observation_hash缓存首次成功结果后续直接返回Redis TTL1h避免陈旧数据效果非确定性失败率从12%降至0.3%平均重试次数0.08次/请求绝大部分一次成功缓存命中率63%传感器数据变化缓慢5.3 “Agent集群CPU爆了但QPS不高”——异步陷阱与资源泄漏现象K8s监控显示Agent Pod CPU持续95%但http_requests_total仅50 QPS远低于设计容量200 QPS。排查过程kubectl exec进入Pod运行top→python进程占95% CPU用py-spy record -p pid -o profile.svg生成火焰图 → 90%时间在asyncio.base_events._run_once检查代码发现_call_llm_with_timeout中aiohttp.ClientSession未关闭根本问题aiohttp.ClientSession是异步资源必须显式close()否则连接池泄漏event loop被阻塞。修复代码async def _call_llm_with_timeout(self, prompt: str) - str: timeout aiohttp.ClientTimeout(total15) # ✅ 正确用async with确保关闭 async with aiohttp.ClientSession(timeouttimeout) as session: async with session.post(...) as resp: ...延伸问题Redis连接池同样需close()我们用aioredis.from_url(..., max_connections20)并全局复用所有异步资源DB连接、HTTP客户端、消息队列必须统一管理我们创建ResourceManager单例启动时初始化关闭时批量清理经验总结Python异步编程的“坑”不在语法而在资源生命周期管理。我们现在的Code Review Checklist第一条就是“所有async with是否配对所有await resource.close()是否执行”5.4 “业务方说策略变了我要改多少代码”——Stage 4演化的落地实践背景医疗客户要求原策略“血压140/90即触发随访”改为“收缩压140且舒张压90或收缩压160或舒张压100”。旧方案失败策略硬编码在Prompt中If systolic 140 and diastolic 90, trigger follow-up→ 修改需改Prompt、重新测试、全量发布新方案Stage 4策略存入数据库CREATE TABLE patient_rules ( id SERIAL PRIMARY KEY, version VARCHAR(20) NOT NULL, condition TEXT NOT NULL, -- 存储表达式字符串 action VARCHAR(50) NOT NULL, created_at TIMESTAMP DEFAULT NOW() ); INSERT INTO patient_rules (version, condition, action) VALUES (v2.1, systolic 140 and diastolic 90 or systolic 160 or diastolic 100