<task-notification>
<id>root#local_agent#0</id>
<state>completed</state>
<timestamp>2025-04-04T10:30:45Z</timestamp>
<result>
<success>true</success>
<output>
<key>value</key>
</output>
<metrics>
<tokens_used>1250</tokens_used>
<duration_ms>3450</duration_ms>
</metrics>
</result>
<next_tasks>
<task_id>root#local_agent#1</task_id>
<task_id>root#local_agent#2</task_id>
</next_tasks>
</task-notification>from typing import Dict, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime
import json
from threading import RLock
@dataclass
class ScratchpadEntry:
"""Scratchpad中的一个条目"""
key: str
value: Any
created_at: datetime
updated_at: datetime
writer_id: str # 最后的写者ID
version: int = 0
read_only: bool = False
class Scratchpad:
"""跨Worker的共享内存"""
def __init__(self):
self._data: Dict[str, ScratchpadEntry] = {}
self._lock = RLock()
self._access_log: list = []
def put(self, key: str, value: Any, writer_id: str, read_only: bool = False) -> None:
"""写入值"""
with self._lock:
now = datetime.now()
new_version = 0
created_at = now
if key in self._data:
if self._data[key].read_only:
raise ValueError(f"Key {key} is read-only")
new_version = self._data[key].version + 1
created_at = self._data[key].created_at
entry = ScratchpadEntry(
key=key,
value=value,
created_at=created_at,
updated_at=now,
writer_id=writer_id,
version=new_version,
read_only=read_only,
)
self._data[key] = entry
self._access_log.append({
"timestamp": now,
"operation": "put",
"key": key,
"writer_id": writer_id,
})
def get(self, key: str, reader_id: str, default: Optional[Any] = None) -> Any:
"""读取值"""
with self._lock:
if key not in self._data:
return default
entry = self._data[key]
self._access_log.append({
"timestamp": datetime.now(),
"operation": "get",
"key": key,
"reader_id": reader_id,
"version": entry.version,
})
return entry.value
def get_all(self, reader_id: str) -> Dict[str, Any]:
"""获取所有键值对"""
with self._lock:
self._access_log.append({
"timestamp": datetime.now(),
"operation": "get_all",
"reader_id": reader_id,
})
return {entry.key: entry.value for entry in self._data.values()}
def delete(self, key: str, writer_id: str) -> bool:
"""删除键"""
with self._lock:
if key not in self._data:
return False
if self._data[key].read_only:
raise ValueError(f"Key {key} is read-only")
del self._data[key]
self._access_log.append({
"timestamp": datetime.now(),
"operation": "delete",
"key": key,
"writer_id": writer_id,
})
return True
def batch_get(self, keys: list, reader_id: str) -> Dict[str, Any]:
"""批量获取"""
with self._lock:
result = {}
for key in keys:
if key in self._data:
result[key] = self._data[key].value
self._access_log.append({
"timestamp": datetime.now(),
"operation": "batch_get",
"keys": keys,
"reader_id": reader_id,
"found": len(result),
})
return result
def exists(self, key: str) -> bool:
"""检查键是否存在"""
with self._lock:
return key in self._data
def get_version(self, key: str) -> int:
"""获取键的版本号"""
with self._lock:
return self._data[key].version if key in self._data else -1
def watch(self, key: str, callback, reader_id: str) -> None:
"""监视键的变化(简化实现)"""
# 这是一个简化版本,实际实现应该使用观察者模式
pass
def get_access_log(self, limit: int = 100) -> list:
"""获取访问日志"""
with self._lock:
return self._access_log[-limit:]
def clear(self) -> None:
"""清空所有数据"""
with self._lock:
self._data.clear()
self._access_log.clear()
# 使用示例:多个Worker共享数据
class Worker:
"""Worker进程"""
def __init__(self, worker_id: str, scratchpad: Scratchpad):
self.worker_id = worker_id
self.scratchpad = scratchpad
def research_phase(self):
"""Research阶段"""
print(f"[{self.worker_id}] 执行Research阶段")
findings = {
"key_insights": ["洞察1", "洞察2"],
"requirements": ["需求1", "需求2"],
}
self.scratchpad.put("research_findings", findings, self.worker_id)
print(f"[{self.worker_id}] Research完成,写入Scratchpad")
def synthesis_phase(self):
"""Synthesis阶段"""
print(f"[{self.worker_id}] 执行Synthesis阶段")
# 读取Research的输出
findings = self.scratchpad.get("research_findings", self.worker_id)
print(f"[{self.worker_id}] 读取Research结果: {findings}")
plan = {
"steps": ["步骤1", "步骤2", "步骤3"],
"resources": ["资源1", "资源2"],
}
self.scratchpad.put("execution_plan", plan, self.worker_id)
print(f"[{self.worker_id}] Synthesis完成,写入Scratchpad")
def implementation_phase(self):
"""Implementation阶段"""
print(f"[{self.worker_id}] 执行Implementation阶段")
# 读取Synthesis的输出
plan = self.scratchpad.get("execution_plan", self.worker_id)
print(f"[{self.worker_id}] 读取执行计划: {plan}")
# 执行并写入结果
results = {"artifacts": ["工件1", "工件2"]}
self.scratchpad.put("implementation_results", results, self.worker_id)
def run(self):
"""执行所有阶段"""
self.research_phase()
self.synthesis_phase()
self.implementation_phase()
if __name__ == "__main__":
# 创建共享Scratchpad
scratchpad = Scratchpad()
# 创建Workers
worker1 = Worker("Worker-1", scratchpad)
worker2 = Worker("Worker-2", scratchpad)
# 执行(演示单线程执行,实际应为多线程)
worker1.run()
print("\n=== 最终Scratchpad内容 ===")
all_data = scratchpad.get_all("demo")
for key, value in all_data.items():
print(f"{key}: {value}")
print("\n=== 访问日志 ===")
logs = scratchpad.get_access_log()
for log in logs:
print(log)channels:
# HTTP webhook渠道
webhooks:
type: http_post
endpoints:
- url: https://api.example.com/webhook
method: POST
headers:
Authorization: "Bearer token"
retry_policy:
max_retries: 3
backoff: exponential
# 电子邮件渠道
email:
type: email
smtp_config:
host: smtp.example.com
port: 587
use_tls: true
from_address: agent@example.com
# Slack通知
slack:
type: slack
webhook_url: "https://hooks.slack.com/services/..."
channel: "#agent-notifications"
# 数据库存储
database:
type: database
connection_string: "postgresql://..."
table: agent_messages
## 路由规则
routing_rules:
- trigger: workflow_start
channels: [webhooks, slack]
format: json
- trigger: workflow_complete
channels: [email, slack]
format: plain
- trigger: approval_needed
channels: [slack, database]
format: json
- trigger: error_occurred
channels: [email, database]
priority: high
format: jsonfrom typing import Dict, Any, List
from dataclasses import dataclass
from enum import Enum
import json
class MessagePriority(Enum):
"""消息优先级"""
LOW = 0
NORMAL = 1
HIGH = 2
CRITICAL = 3
@dataclass
class Message:
"""统一的消息格式"""
id: str
source_agent_id: str
target_agent_ids: List[str]
message_type: str # e.g., "task_result", "approval_request"
payload: Dict[str, Any]
priority: MessagePriority = MessagePriority.NORMAL
timestamp: str = None
ttl_seconds: int = 3600 # Time-to-live
require_acknowledgment: bool = False
def to_json(self) -> str:
"""转换为JSON"""
return json.dumps({
"id": self.id,
"source_agent_id": self.source_agent_id,
"target_agent_ids": self.target_agent_ids,
"message_type": self.message_type,
"payload": self.payload,
"priority": self.priority.value,
"timestamp": self.timestamp,
})
class ChannelAdapter:
"""渠道适配器"""
def adapt_to_slack(self, msg: Message) -> Dict[str, Any]:
"""适配到Slack格式"""
return {
"text": f"消息ID: {msg.id}",
"blocks": [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*{msg.message_type}*\n\n{json.dumps(msg.payload, indent=2)}",
}
}
],
}
def adapt_to_email(self, msg: Message) -> Dict[str, str]:
"""适配到Email格式"""
return {
"subject": f"[{msg.message_type}] Message {msg.id}",
"body": f"""
消息ID: {msg.id}
来源: {msg.source_agent_id}
类型: {msg.message_type}
优先级: {msg.priority.name}
内容:
{json.dumps(msg.payload, indent=2)}
""",
}
def adapt_to_webhook(self, msg: Message) -> Dict[str, Any]:
"""适配到Webhook格式"""
return {
"event": msg.message_type,
"message_id": msg.id,
"source": msg.source_agent_id,
"data": msg.payload,
"timestamp": msg.timestamp,
}class ReliableMessageQueue:
"""可靠的消息队列"""
def __init__(self):
self.pending_messages = {} # msg_id -> message
self.confirmed_messages = set()
def send(self, message: Message, max_retries: int = 3) -> bool:
"""发送消息,支持重试"""
for attempt in range(max_retries):
try:
# 发送消息
self._transmit(message)
# 等待确认
if self._wait_for_ack(message.id, timeout=5):
self.confirmed_messages.add(message.id)
return True
except Exception as e:
print(f"发送失败(第{attempt+1}次): {e}")
if attempt < max_retries - 1:
time.sleep(2 ** attempt) # 指数退避
return False
def _transmit(self, message: Message):
"""实际发送"""
self.pending_messages[message.id] = message
# 调用实际的传输机制
def _wait_for_ack(self, msg_id: str, timeout: int) -> bool:
"""等待确认"""
# 实现等待逻辑
return True
def acknowledge(self, msg_id: str) -> None:
"""接收方确认消息"""
if msg_id in self.pending_messages:
del self.pending_messages[msg_id]
self.confirmed_messages.add(msg_id)class OrderedMessageQueue:
"""保证顺序的消息队列"""
def __init__(self):
self.queues: Dict[str, list] = {} # key -> messages
def enqueue(self, key: str, message: Message) -> None:
"""入队(保证同一key的消息顺序)"""
if key not in self.queues:
self.queues[key] = []
self.queues[key].append(message)
def dequeue(self, key: str) -> Message:
"""出队(FIFO)"""
if key in self.queues and self.queues[key]:
return self.queues[key].pop(0)
return None
def get_queue_length(self, key: str) -> int:
"""获取队列长度"""
return len(self.queues.get(key, []))class BackpressureController:
"""背压控制"""
def __init__(self, max_queue_size: int = 1000):
self.max_queue_size = max_queue_size
self.queue_sizes: Dict[str, int] = {}
def can_send(self, agent_id: str) -> bool:
"""检查是否可以发送"""
current_size = self.queue_sizes.get(agent_id, 0)
return current_size < self.max_queue_size
def record_send(self, agent_id: str) -> None:
"""记录发送"""
self.queue_sizes[agent_id] = self.queue_sizes.get(agent_id, 0) + 1
def record_delivery(self, agent_id: str) -> None:
"""记录送达"""
if agent_id in self.queue_sizes:
self.queue_sizes[agent_id] -= 1
def get_backpressure_status(self, agent_id: str) -> float:
"""获取背压状态(0-1,1表示队列满)"""
return min(1.0, self.queue_sizes.get(agent_id, 0) / self.max_queue_size)