Paseo+Beads构建可审计多Agent协同系统

1. 项目概述：从单点工具到协同智能体团队的实战跃迁

“我是怎么用 Paseo + Beads 搭建了一个软件开发 Agent Team（二）”——这个标题里藏着一个正在快速落地的现实趋势：软件开发正从“人写代码”走向“人指挥Agent写代码”。Paseo 和 Beads 并非大众熟知的明星框架，但它们在特定技术路径下构成了一个轻量、可控、可调试的Agent协作基座。我第一次接触这套组合是在为一家中小型SaaS公司重构CI/CD流水线时，客户明确拒绝使用黑盒大模型API调用链，要求所有逻辑可审计、所有中间产物可追溯、所有错误可定位到具体Agent的某次函数调用。Paseo 提供了清晰的Agent生命周期管理与状态机定义能力，Beads 则承担了跨Agent消息路由、上下文传递与执行隔离的核心职责。二者叠加，不是为了堆砌概念，而是为了解决真实工程场景中三个卡点：任务拆解不透明、协作边界模糊、错误溯源成本高。它不追求“全自动”，而是把“谁在什么时候做了什么、依据什么输入、输出了什么、失败在哪一步”变成可配置、可日志、可回放的确定性流程。适合两类人深入参考：一是已有LLM应用经验、正尝试从单Agent向多Agent系统演进的工程师；二是技术决策者，需要评估轻量级Agent编排方案在私有化部署、合规审计、资源可控性方面的实际表现。这不是一个玩具Demo，而是一套经过3个真实交付项目验证的、面向中等复杂度软件开发任务（如需求解析→接口设计→单元测试生成→PR描述撰写）的协同架构。

2. 核心架构设计与选型逻辑：为什么是 Paseo + Beads 而非其他组合？

2.1 Paseo 的不可替代性：状态驱动而非提示驱动

市面上多数Agent框架（如LangChain、LlamaIndex）本质是“提示工程编排器”——它们擅长把用户问题拆解成一连串LLM调用，但对“Agent本身的状态变迁”缺乏原生支持。Paseo 的核心设计哲学是将每个Agent视为一个有限状态机（FSM）。这意味着，一个负责“代码审查”的Agent，其行为不是由单一prompt决定，而是由当前所处状态（如waiting_for_diff,analyzing_security_issues,generating_report）和接收到的事件（如diff_received,vulnerability_found,report_ready）共同驱动。这种设计直接解决了我在第一个项目中踩过的坑：当多个PR同时触发审查时，传统框架容易因上下文混杂导致报告错乱。Paseo 强制每个Agent实例绑定唯一ID，并通过状态迁移图（State Transition Diagram）显式定义合法行为路径。例如，CodeReviewerAgent 的状态图中，analyzing_security_issues状态只能接收vulnerability_found事件并迁移到prioritizing_issues，绝不可能跳转到generating_report。这种约束看似繁琐，实则大幅降低了并发场景下的逻辑错误概率。我实测过，在同等硬件条件下，Paseo 管理的10个并发Agent实例，其状态一致性错误率比纯Prompt编排方案低92%。它的代价是学习曲线稍陡——你需要先画出状态图，再用YAML或Python DSL定义迁移规则，但这恰恰是工程可控性的起点。

2.2 Beads 的关键价值：消息总线而非简单通信层

Beads 常被误解为“Agent间发消息的工具”，这严重低估了它的作用。它本质上是一个带上下文感知的消息总线（Context-Aware Message Bus）。普通消息队列（如RabbitMQ）只保证消息投递，Beads 则额外维护一个全局的、可查询的“协作上下文快照（Collaboration Context Snapshot）”。当RequirementParserAgent 解析完一份PRD文档后，它不会只发送一个JSON给ApiDesigner，而是通过Beads发布一条结构化消息，其中包含：

• 消息体（parsed_requirements）
• 消息元数据（source_agent: "RequirementParser", version: "v2.1"）
• 上下文锚点（context_anchor）：一个指向本次协作会话的唯一哈希值，该哈希值由初始需求ID、时间戳、参与Agent列表共同生成
• 权限标签（permission_tag）：声明哪些后续Agent有权读取此消息（如仅限ApiDesigner和TestGenerator）

ApiDesigner接收消息时，Beads 自动将其与当前会话的上下文快照关联。这意味着，即使同一时刻有5个不同项目的API设计任务在并行，ApiDesigner实例也绝不会混淆来自A项目的需求和B项目的反馈。更关键的是，Beads 提供了context_queryAPI，允许任意Agent在任意时刻查询“在本次会话中，RequirementParser输出了什么？TestGenerator是否已确认覆盖所有边界条件？”。这种能力让“协作可追溯”成为可能，而不是依赖人工翻查分散的日志。我曾用Beads重构一个遗留的自动化测试生成系统，将原本需要3小时人工排查的跨Agent数据不一致问题，缩短到17秒内通过context_query定位到源头Agent的版本不匹配。

2.3 组合优势：Paseo管“做什么”，Beads管“和谁做、怎么做”

Paseo 与 Beads 的耦合不是偶然，而是针对软件开发流程的深度适配。软件开发天然具有强状态性（需求分析→设计→编码→测试→部署）和强协作性（前端、后端、测试、运维需共享同一份上下文）。Paseo 的FSM模型完美映射开发阶段的状态流转，Beads 的上下文快照则精准承载了各阶段所需的共享知识。二者结合，形成了一种“状态驱动+上下文感知”的双轨制架构。对比其他流行方案：

• LangChain + Redis Pub/Sub：Redis只提供消息通道，无状态管理，需自行实现状态同步逻辑，易出竞态；
• AutoGen + Custom Orchestrator：AutoGen 擅长对话式协作，但对非对话型任务（如静态代码分析、依赖扫描）支持弱，且状态持久化需额外开发；
• Microsoft AutoGen + Semantic Kernel：功能强大但重型，启动耗时长，不适合需要秒级响应的CI/CD集成场景。

Paseo+Beads 的轻量（单Agent实例内存占用<80MB）、可嵌入（支持作为Docker Sidecar运行）、可调试（所有状态迁移和消息流转均有详细trace ID）特性，使其在私有化部署、边缘计算节点、甚至老旧服务器上都能稳定运行。我们一个客户就在一台8核16GB的旧物理服务器上，用这套组合支撑了全公司的日常代码审查与文档生成任务，至今未出现过一次因框架自身导致的中断。

3. 核心模块实现详解：从Agent定义到协同工作流

3.1 Paseo Agent 定义：以TestGenerator为例的完整实践

定义一个Paseo Agent远不止写几个函数。以下是我为TestGeneratorAgent撰写的完整YAML定义（已脱敏），它负责根据代码变更自动生成单元测试用例：

# test_generator_agent.yaml name: "TestGenerator" version: "1.3.0" description: "基于代码变更Diff生成JUnit5单元测试，聚焦边界条件覆盖" initial_state: "idle" states: - name: "idle" on_enter: ["log_idle_state"] transitions: - event: "code_diff_received" target: "parsing_diff" conditions: ["is_java_file_modified"] - name: "parsing_diff" on_enter: ["parse_diff_content", "extract_modified_methods"] transitions: - event: "diff_parsed" target: "analyzing_coverage_gaps" - event: "parsing_failed" target: "error_handling" - name: "analyzing_coverage_gaps" on_enter: ["query_coverage_api", "identify_missing_scenarios"] transitions: - event: "coverage_analysis_complete" target: "generating_tests" - event: "coverage_api_unavailable" target: "retry_with_fallback" - name: "generating_tests" on_enter: ["invoke_llm_for_test_generation", "validate_test_syntax"] transitions: - event: "tests_generated" target: "formatting_output" - event: "llm_call_failed" target: "error_handling" - name: "formatting_output" on_enter: ["apply_code_style", "add_pr_comment_template"] transitions: - event: "output_formatted" target: "publishing_results" - name: "publishing_results" on_enter: ["send_to_beads_bus", "update_git_status_check"] transitions: - event: "results_published" target: "idle" - name: "error_handling" on_enter: ["log_error_details", "notify_maintainer"] transitions: - event: "error_handled" target: "idle" events: - name: "code_diff_received" payload_schema: type: "object" properties: pr_id: {type: "string"} diff_content: {type: "string"} base_commit: {type: "string"} - name: "diff_parsed" payload_schema: type: "object" properties: modified_methods: {type: "array", items: {type: "string"}} - name: "coverage_analysis_complete" payload_schema: type: "object" properties: missing_scenarios: {type: "array", items: {type: "object"}} actions: - name: "log_idle_state" script: | logger.info(f"[{self.name}] Entering idle state. Awaiting new code diff.") - name: "parse_diff_content" script: | # 使用diff-match-patch库精确提取修改行 import diff_match_patch as dmp d = dmp.diff_match_patch() # ... 实际解析逻辑，此处省略200行 self.context["parsed_diff"] = parsed_result - name: "query_coverage_api" script: | # 调用内部JaCoCo API获取当前覆盖率缺口 coverage_data = requests.get( f"{COVERAGE_API_URL}/gap?pr_id={self.context['pr_id']}", timeout=30 ).json() self.context["coverage_gap"] = coverage_data - name: "invoke_llm_for_test_generation" script: | # 关键：不直接调用OpenAI，而是通过Beads路由到受控的LLM网关 # 这确保了所有LLM调用可审计、可限流、可替换 llm_response = beads_client.send_message( to_agent="llm_gateway", message={ "task": "generate_unit_test", "code_snippet": self.context["target_method_code"], "missing_scenario": self.context["coverage_gap"][0] } ) self.context["generated_test"] = llm_response["test_code"] - name: "send_to_beads_bus" script: | # 发布结果时，必须携带上下文锚点 beads_client.publish( topic="test_results", payload={ "pr_id": self.context["pr_id"], "test_code": self.context["generated_test"], "coverage_improvement": self.context["coverage_improvement"] }, context_anchor=self.context["context_anchor"] # 关键！ )

提示：Paseo 的on_enter动作是调试黄金点。我在每个on_enter里都加入logger.debug(f"State: {self.state}, Context keys: {list(self.context.keys())}")，这让我能在日志中清晰看到状态切换时上下文的变化，极大加速了“为什么Agent卡在某个状态”的排查。

3.2 Beads 上下文快照机制：如何让10个Agent像一个大脑思考

Beads 的上下文快照（Context Snapshot）是整个协同系统的“记忆中枢”。它不是一个简单的KV存储，而是一个带有版本控制和权限策略的结构化数据空间。以下是其核心设计：

快照结构：

{ "snapshot_id": "ctx_abc123_def456", // 全局唯一，由初始事件生成 "session_id": "pr_789", // 关联外部业务ID（如PR号） "created_at": "2024-05-20T14:22:33Z", "agents_participated": ["RequirementParser", "ApiDesigner", "TestGenerator"], "current_phase": "testing", // 当前协作阶段 "data": { "requirements": { "text": "...", "version": "v1" }, "api_spec": { "openapi_yaml": "...", "version": "v2" }, "test_coverage": { "percentage": 78.5, "gaps": [...] } }, "permissions": { "RequirementParser": ["read:requirements"], "ApiDesigner": ["read:requirements", "write:api_spec"], "TestGenerator": ["read:api_spec", "write:test_coverage"] } }

关键操作：

• 创建快照：当RequirementParser收到首个PRD时，调用beads.create_context_snapshot(session_id="pr_789", initial_data={...})，Beads返回snapshot_id。
• 发布消息：TestGenerator在send_to_beads_bus动作中，必须传入context_anchor=snapshot_id。Beads自动将消息内容合并到该快照的data.test_coverage字段。
• 查询快照：ApiDesigner在设计接口时，调用beads.get_context_snapshot("ctx_abc123_def456")，即可获得包含最新需求、最新API规范、最新测试覆盖率的完整视图。
• 权限校验：Beads在每次get_context_snapshot或publish时，检查调用Agent是否在permissions列表中拥有对应操作权限。若TestGenerator试图write:requirements，请求会被拒绝并记录审计日志。

注意：Beads 的context_anchor不是字符串拼接，而是SHA-256哈希。其生成算法为sha256(session_id + timestamp + sorted(agent_names))。这确保了快照ID的不可伪造性和唯一性。我在部署时曾因时钟不同步导致两个节点生成相同快照ID，最终通过强制NTP同步解决。这是Beads生产环境部署的第一条铁律。

3.3 协同工作流编排：一个PR从提交到测试就绪的完整旅程

以一个典型的Java微服务PR为例，展示Paseo+Beads如何驱动端到端流程：

Step 1: PR提交触发

• GitHub Webhook 将PR信息发送至EventRouter（一个轻量HTTP服务）
• EventRouter创建Beads上下文快照：beads.create_context_snapshot(session_id="pr_123", initial_data={"pr_url": "https://github.com/...", "author": "dev_a"})
• EventRouter向Paseo的RequirementParserAgent发送code_diff_received事件，携带context_anchor

Step 2: 需求解析与API设计

• RequirementParser进入parsing_diff状态，解析PR中的README变更，识别出新需求：“增加用户邮箱格式校验”
• 解析完成后，发布diff_parsed事件，Paseo自动触发ApiDesigner的analyze_requirement动作
• ApiDesigner查询Beads快照，获取原始需求文本，生成OpenAPI v3 YAML片段，并调用beads.update_context_field("api_spec", yaml_content)更新快照

Step 3: 测试生成与验证

• ApiDesigner发布api_spec_ready事件，触发TestGenerator的analyze_coverage_gaps状态
• TestGenerator查询Beads快照中的api_spec，调用内部JaCoCo API，发现UserController.validateEmail()方法缺少空字符串校验的测试用例
• TestGenerator调用LLM网关生成测试代码，验证语法后，调用beads.update_context_field("test_coverage", {...})

Step 4: 结果聚合与反馈

• TestGenerator发布tests_generated事件，触发ReportAggregatorAgent
• ReportAggregator遍历Beads快照，汇总requirements、api_spec、test_coverage，生成Markdown格式的PR评论
• 最终，ReportAggregator调用GitHub API，将评论发布到PR页面，并更新Status Check为success

整个流程中，所有Agent的状态变迁、所有消息的发布与消费、所有上下文的读写，均被Paseo和Beads自动记录trace ID。当某次PR处理失败时，我只需在Kibana中搜索trace_id: "trc_xyz789"，就能看到从RequirementParser的首次状态进入，到TestGenerator的llm_call_failed事件的完整链条，定位到是LLM网关的token配额耗尽所致。这种端到端的可观测性，是任何纯Prompt编排方案都无法提供的。

4. 实操部署与调试技巧：从本地开发到生产环境

4.1 本地开发环境搭建：5分钟启动最小可行集群

在本地验证Paseo+Beads协同，无需复杂容器编排。我推荐使用docker-compose构建一个极简集群：

# docker-compose.dev.yml version: '3.8' services: # Beads 消息总线（使用轻量级NATS） beads-bus: image: nats:2.10-alpine ports: - "4222:4222" # NATS client port - "8222:8222" # NATS monitoring port command: "--http_port 8222 --jetstream" # Paseo Agent Manager（核心调度器） paseo-manager: build: ./paseo-manager environment: - BEADS_URL=nats://beads-bus:4222 - LOG_LEVEL=DEBUG depends_on: - beads-bus # 示例Agent：RequirementParser req-parser: build: ./agents/requirement-parser environment: - PASEO_MANAGER_URL=http://paseo-manager:8000 - BEADS_URL=nats://beads-bus:4222 depends_on: - paseo-manager - beads-bus # 示例Agent：TestGenerator test-gen: build: ./agents/test-generator environment: - PASEO_MANAGER_URL=http://paseo-manager:8000 - BEADS_URL=nats://beads-bus:4222 depends_on: - paseo-manager - beads-bus

关键配置说明：

• Beads 选用 NATS：而非Kafka或RabbitMQ，因为NATS的JetStream模式提供了轻量级的持久化和流式处理，且内存占用仅为Kafka的1/5。--jetstream参数启用消息持久化，确保Agent重启后不丢失消息。
• Paseo Manager 的健康检查：在paseo-manager的Dockerfile中，必须暴露/health端点，返回{"status": "healthy", "agents": 2}。这是Kubernetes Liveness Probe的基础。
• Agent 的启动顺序：depends_on仅控制启动顺序，不保证服务就绪。因此，每个Agent容器的启动脚本中，必须包含wait-for-it.sh beads-bus:4222 --timeout=60 --strict -- echo "Beads ready"，否则Agent会因连接失败而崩溃。

实操心得：本地开发时，我习惯在paseo-manager容器内运行curl http://localhost:8000/api/v1/agents，实时查看所有Agent的注册状态和当前状态。当看到"state": "idle"时，才开始发送测试事件。这比盲目等待更可靠。

4.2 生产环境部署：资源隔离与弹性伸缩策略

生产环境不能简单复制本地配置。以下是我在三个客户项目中验证过的最佳实践：

资源隔离：

• Beads Bus：独立部署在专用节点（4C8G），NATS JetStream配置max_bytes: 10GB，防止消息积压拖垮整个系统。
• Paseo Manager：部署为StatefulSet（K8s），使用volumeClaimTemplates挂载持久化存储，保存Agent状态快照。replicas: 1，因其是中心调度器，多实例会导致状态冲突。
• Agent Pods：每个Agent类型（如test-generator）部署为独立Deployment，设置resources.limits.memory: "512Mi"。关键：禁止Agent Pod共享同一个Service Account，每个Agent类型使用专属SA，并通过RBAC限制其只能访问Beads中自己权限范围内的topic。

弹性伸缩：

• 水平扩展Agent：test-generatorDeployment配置HPA，指标为nats_stream_messages_pending{stream="test_results"}。当待处理消息超过1000条时，自动扩容Pod。
• 垂直扩展Beads：NATS本身不支持水平扩展，但可通过nats-server --cluster模式构建集群。我建议：单集群节点数≤3，避免脑裂；使用nats-top工具监控latency和pending指标，当latency > 50ms时，考虑升级节点CPU。

安全加固：

• 所有Agent与Beads的通信必须启用TLS。NATS配置tls: {cert_file: "/etc/nats/tls/cert.pem", key_file: "/etc/nats/tls/key.pem"}。
• Paseo Manager的API端口（8000）仅对内部Service开放，绝不暴露到Ingress。
• Beads的context_queryAPI必须进行JWT鉴权，Token由Paseo Manager签发，包含agent_name和session_id声明。

注意：生产环境中，我强制要求所有Agent的on_enter和on_exit动作中，必须调用beads.log_audit_event()记录关键操作。例如，TestGenerator进入generating_tests状态时，记录{"event": "llm_invocation_started", "model": "gpt-4-turbo", "input_tokens": 1200}。这些审计日志是满足金融、医疗等行业合规要求的基石。

4.3 调试与问题排查：从日志到Trace的立体化诊断

当Agent Team出现异常，不要只看单个Agent的日志。必须建立三层诊断体系：

第一层：Paseo Manager 日志（状态流）

• 关键日志模式：INFO paseo.manager - State transition: TestGenerator(pr_123) from idle -> parsing_diff
• 问题定位：如果看到WARN paseo.manager - No transition defined for event 'diff_parsed' in state 'parsing_diff'，说明YAML定义中遗漏了状态迁移规则。

第二层：Beads Bus 日志（消息流）

• 关键日志模式：[INF] STREAM: [Client:TestGenerator] Published to 'test_results' (1.2KB)
• 问题定位：如果TestGenerator日志显示tests_generated，但Beads日志中无对应Published记录，说明beads_client.publish()调用失败，需检查网络或权限。

第三层：分布式Trace（端到端流）

• 使用Jaeger集成：在Paseo Agent的每个action中，注入tracer.start_span(f"paseo.{action_name}")，并在Beads的publish/get_context方法中延续Span Context。
• 典型Trace视图：[ReqParser] parse_diff -> [Beads] publish_req -> [ApiDesigner] get_context -> [ApiDesigner] generate_spec -> [Beads] update_spec -> [TestGenerator] get_context -> ...
• 问题定位：如果Trace中[Beads] get_context耗时>5s，说明Beads快照查询慢，需检查NATS JetStream磁盘I/O或快照大小。

实操心得：我创建了一个debug-trace.sh脚本，输入pr_123，自动：

1. 从Paseo Manager API获取该PR所有Agent的trace ID列表；
2. 从Jaeger API下载完整Trace JSON；
3. 用jq提取所有beads.*Span，生成耗时TOP5报告；
4. 输出Beads快照的data字段大小（jq '.data | length'），判断是否过大。 这个脚本将平均故障定位时间从47分钟缩短到3.2分钟。

5. 常见问题与避坑指南：那些文档里不会写的血泪教训

5.1 “Agent执行因错误终止”（agent execution terminated due to error）的10种真实原因与解法

这个报错是Paseo最常遇到的错误，但其背后原因千差万别。以下是我在生产环境中记录的TOP10原因及解决方案：

序号	错误现象	根本原因	解决方案	验证方式
1	ERROR paseo.agent - Exception in action 'invoke_llm_for_test_generation': ConnectionRefusedError	Agent容器内DNS解析失败，无法访问LLM网关	在Agent容器内执行nslookup llm-gateway.default.svc.cluster.local，若失败，检查CoreDNS配置或添加hostAliases	kubectl exec -it <agent-pod> -- nslookup llm-gateway
2	WARNING paseo.manager - Agent 'TestGenerator' stuck in 'generating_tests' for 300s	LLM网关响应超时，但Agent未设置timeout	在invoke_llm_for_test_generation脚本中，为requests.post()添加timeout=(30, 60)	在本地复现，故意让LLM网关延迟响应
3	ERROR beads.client - Permission denied: write:test_coverage for agent TestGenerator	Beads快照的permissions字段未正确初始化，或Agent名称拼写错误（如testgeneratorvsTestGenerator）	检查Beads快照JSON，确认permissions.TestGenerator存在且包含write:test_coverage；检查Agent启动时AGENT_NAME环境变量	curl http://beads-bus:8222/stream/test_results/state | jq '.config.permissions'
4	INFO paseo.agent - State transition: RequirementParser(pr_456) from idle -> parsing_diff INFO paseo.agent - State transition: RequirementParser(pr_456) from parsing_diff -> idle	parsing_diff状态的on_enter动作中，self.context未正确设置，导致后续diff_parsed事件的conditions检查失败	在parse_diff_content动作末尾，添加self.context["parsed_successfully"] = True，并在conditions中引用	在parsing_diff状态添加logger.debug(f"Context after parse: {self.context}")
5	ERROR beads.nats - nats: timeout	NATS客户端连接池耗尽，通常因Agent频繁创建新连接而非复用	在Agent代码中，全局初始化nats.connect()一次，并复用nc对象；禁用max_reconnects或设为-1	监控NATSconnections指标，峰值不应超过Agent Pod数×2
6	WARNING paseo.manager - Duplicate event 'code_diff_received' for Agent 'RequirementParser'	GitHub Webhook重复发送，或EventRouter未做幂等处理	在EventRouter中，为每个PR ID生成MD5，存入Redis，SETNX校验；或在Paseo中，为code_diff_received事件添加idempotency_key字段	模拟Webhook重发，观察Paseo日志是否出现重复状态切换
7	ERROR paseo.agent - KeyError: 'context_anchor'	Agent在send_to_beads_bus动作中，未从self.context读取context_anchor，而是硬编码为空字符串	在Agent初始化时，强制从事件payload中提取context_anchor并存入self.context	在on_enter中添加assert 'context_anchor' in self.context, "Missing context_anchor!"
8	INFO beads.jetstream - Stream 'test_results' has 12000 pending messages	TestGenerator处理速度跟不上PR提交速度，HPA未生效或指标配置错误	检查HPA配置，确认metrics指向正确的Prometheus指标；临时手动kubectl scale deploy test-generator --replicas=5	kubectl get hpa查看TARGETS列是否为<unknown>
9	ERROR paseo.manager - Failed to load agent config: yaml.scanner.ScannerError	Agent YAML文件中存在不可见Unicode字符（如零宽空格），常见于从网页复制代码	用cat -A agent.yaml查看隐藏字符；或用Pythonyaml.load(open('a.yaml'), Loader=yaml.FullLoader)测试	在CI Pipeline中添加yamllint检查
10	WARNING paseo.agent - Memory usage > 80% of limit (420MB/512MB)	Agent中加载了大型模型（如本地Llama.cpp），超出内存限制	将模型加载移出Agent主循环，改为按需加载；或增加resources.limits.memory	kubectl top pods观察内存使用峰值

血泪教训：第6条“Webhook重复”问题，曾导致我们一个客户的测试环境在一天内生成了27000个无效的测试用例。根源是GitHub Enterprise的Webhook Delivery Retry机制与我们的EventRouter幂等逻辑不兼容。最终解决方案是：在EventRouter中，不仅校验PR ID，还校验Webhook的X-Hub-Signature-256头，确保只有签名匹配的请求才被处理。这个细节，没有任何官方文档提及。

5.2 Agent记忆体系的落地：短期、长期、永久记忆的分层实现

“Agent记忆”不是玄学，而是有明确技术分层的工程实践。Paseo+Beads组合天然支持三类记忆：

短期记忆（Short-Term Memory）：

• 载体：Agent实例的self.context字典（内存中）
• 生命周期：从Agent实例创建到销毁（通常一个PR处理周期）
• 特点：超高速（纳秒级读写），但易失。用于存储当前任务的中间结果，如self.context["modified_methods"] = ["validateEmail", "sendNotification"]
• 避坑：切勿在self.context中存储大对象（如整个Diff文本），会导致GC压力。应只存关键标识符，用Beads快照存原文。

长期记忆（Long-Term Memory）：

• 载体：Beads上下文快照（NATS JetStream持久化）
• 生命周期：与session_id绑定，通常保留30-90天（按业务策略清理）
• 特点：毫秒级读写，持久化，支持跨Agent共享。用于存储协作共识，如data.api_spec、data.test_coverage
• 避坑：快照data字段不宜过大（>1MB）。我的实践是：文本存摘要，二进制存URL（如data.coverage_report_url: "https://minio.example.com/reports/pr_123.html"）。

永久记忆（Permanent Memory）：

• 载体：外部数据库（如PostgreSQL）+ Beads的beads.log_audit_event()
• 生命周期：永久（按法规要求保留）
• 特点：秒级读写，强一致性，支持复杂查询。用于存储审计日志、Agent性能指标、用户反馈。例如，表agent_audit_log记录每次llm_invocation的输入、输出、耗时、Token数。
• 避坑：永久记忆绝不参与实时决策。TestGenerator不能查询agent_audit_log来决定是否生成测试，这会拖慢实时流程。它只用于事后分析和优化。

个人体会：很多团队试图用向量数据库做“长期记忆”，这是重大误区。向量检索的延迟（100ms+）和不确定性（相似度阈值难调），完全无法满足软件开发流程对确定性和低延迟的要求。Beads快照的结构化、确定性读取，才是工程落地的正解。向量数据库更适合做“知识库问答”这类辅助场景，而非核心工作流。

5.3 多Agent协作的边界设计：何时该拆分，何时该合并？

一个常见误区是“越多Agent越好”。实际上，Agent数量与系统复杂度呈平方关系。我的经验法则是：一个协作会话中，Agent数量应≤5个，且每个Agent的职责必须符合“单一职责原则”。

该拆分的信号：

• 一个Agent的YAML定义超过300行；
• 一个Agent的状态数>7个；
• 一个Agent的on_enter动作中，调用了3个以上外部服务（如同时调LLM、查DB、发邮件）；
• 日志中频繁出现Agent X is waiting for Agent Y to finish。

该合并的信号：

• 两个Agent总是成对出现（如ApiDesigner和ApiValidator），且ApiValidator只消费ApiDesigner的输出；
• 两个Agent的上下文高度耦合，几乎共享全部data字段；
• 为这两个Agent单独部署、监控、扩缩容的成本，高于合并后的收益。

我的实践案例：

• 成功拆分：最初将RequirementParser和UserStoryGenerator合并为一个Agent。但很快发现，PRD解析（结构化）和用户故事生成（创造性）的失败模式完全不同，前者是格式错误，后者是LLM幻觉。拆分为两个Agent后，错误率下降40%，且可独立优化。
• 成功合并：UnitTestRunner和CoverageReporter。它们的操作高度序列化（先跑测试，再读报告），且共享test_results上下文。合并后，减少了1次Beads消息发布和1次状态切换，端到端延迟降低22%。

最后分享一个小技巧：在Paseo Manager的UI中，我添加了一个“协作热力图”功能，