Python AI Assistants Multi Agents MCP Servers

Python AI助手-带MCP服务器的多代理系统

使用多代理架构和模型上下文协议（MCP）服务器的人工智能助手的最小实现。

特性

🤖 代理系统

• 基于角色的代理 具有专业能力
• 存储器系统 -代理可以通过TTL记住和回忆信息
• 任务历史记录 -跟踪所有已完成的任务
• 回调 -挂接到任务完成事件
• 能力标记 -根据他们的技能找到代理人
• 工具访问 -代理可以使用MCP服务器工具
• 优先级 -为任务选择确定代理的优先级
• 国家管理 -跟踪代理状态（空闲/繁忙）
• 自动重试 -失败时自动重试
• 指标 -成功率、持续时间、任务计数
• 学习 -代理从反馈中学习并改进

🔧 MCP服务器

• 工具注册 包含元数据和描述
• 执行日志记录 -跟踪所有工具调用
• 统计 -用平均值监控成功/失败率
• 错误处理 -优雅的故障管理
• 参数模式 -定义工具接口
• 缓存 -缓存工具性能结果
• 速率限制 -防止服务器过载
• 钩子 -执行前/执行后钩子
• 工具元数据 -描述和模式

🎯 多代理编排

• 授权 -将任务分配给特定代理
• 汽车代表团 -根据能力自动选择最佳代理
• 并行执行 -在超时的情况下同时运行多个代理
• 广播 -将任务发送给所有代理
• 共享内存 -基于TTL的跨代理数据共享
• 事件日志记录 -全系统活动跟踪
• 状态监测 -实时系统健康状况
• 中间件 -拦截和修改任务
• 排行榜 -按业绩对代理人进行排名
• 负载平衡 -高效地分配任务

🔄 工作流引擎

• 顺序执行 -有序任务处理
• 并行工作流 -同时执行独立步骤
• 依赖项 -任务等待先决条件
• 上下文传递 -在步骤之间共享结果
• 多代理工作流 -协调不同的代理
• 有条件的步骤 -根据条件执行
• 错误处理程序 -每一步自定义错误处理
• 重试逻辑 -失败时自动重试
• 工作流统计信息 -跟踪执行指标

🤝 协作

• Agent协商 -通过协商选择最佳代理
• 多Agent协作 -代理在任务上协同工作
• 投票 -代理人对决定进行投票
• 合作历史 -追踪团队努力

📡 事件系统

• 事件总线 -组件之间的发布/订阅消息传递
• 事件订阅 -订阅特定活动
• 事件历史 -查询过去的事件
• 事件过滤 -按类型和时间筛选

⏰ 调度

• 延迟任务 -安排任务以备将来执行
• 周期性任务 -周期性任务执行
• 任务取消 -取消计划任务
• 自动执行 -运行挂起的任务

🛡️ 韧性

• 断路器 -防止级联故障
• 负载平衡器 -在代理之间分配负载（循环、最不繁忙、性能最佳）
• 速率限制器 -控制请求速率
• 重试逻辑 -具有回退功能的自动重试

💾 坚持

• 国家管理 -保存/加载系统状态
• 检查点 -创建命名快照
• 指标收集 -综合指标跟踪
• 出口、进口 -在系统之间传递知识

安装

pip install -r requirements.txt

快速开始

from mcp_server import MCPServer
from agent import Agent
from multi_agent_system import MultiAgentSystem

# Create MCP server with rate limiting
file_server = MCPServer("file_ops", "File operations", rate_limit=100)
file_server.register_tool("read", lambda path: f"Reading {path}", cacheable=True)

# Create agent with priority
coder = Agent("coder", "Code writer", [file_server], ["coding"], priority=1)

# Create system with max workers
system = MultiAgentSystem(max_workers=5)
system.add_agent(coder)

# Execute task
result = system.delegate("coder", "Write function")
print(result)

用法示例

基本代理委托

result = system.delegate("researcher", "Find AI trends")

并行执行

tasks = [
    {"agent": "researcher", "task": "Research topic A"},
    {"agent": "coder", "task": "Write module B"}
]
# With timeout
results = system.parallel_execute(tasks, timeout=30)

汽车代表团

# Automatically select best agent by capability
result = system.auto_delegate("Find research papers", "research")
print(f"Selected: {result['agent']}")

代理商排行榜

leaderboard = system.get_agent_leaderboard()
for rank, entry in enumerate(leaderboard, 1):
    print(f"{rank}. {entry['name']}: {entry['metrics']['success_rate']:.2%}")

代理内存

# Basic memory
agent.remember("key", "value")
value = agent.recall("key")

# Memory with TTL (expires after 60 seconds)
agent.remember("temp_key", "temp_value", ttl=60)

# Memory management
agent.forget("key")
agent.clear_memory()

共享内存

# Basic shared memory
system.share_data("project_name", "AI Assistant")
data = system.get_shared_data("project_name")

# Shared memory with TTL
system.share_data("session_id", "abc123", ttl=3600)

广播

results = system.broadcast("Status check")

按能力查找

web_agents = system.find_agent_by_capability("web")

具有依赖关系的工作流

from workflow import Workflow

workflow = Workflow("pipeline", max_retries=3)
workflow.add_step("researcher", "Gather data")
workflow.add_step("analyst", "Process data", depends_on=[0])
workflow.add_step("coder", "Generate report", depends_on=[1])

# Sequential execution
results = workflow.execute(system)

# Parallel execution (by dependency levels)
results = workflow.execute(system, parallel=True)

# Conditional steps
workflow.add_step("notifier", "Send alert", 
                  condition=lambda r: r.get(0, {}).get("status") == "completed")

# Error handlers
workflow.add_error_handler(0, lambda e, step, ctx: {"status": "recovered"})

# Workflow stats
stats = workflow.get_stats()
print(f"Avg duration: {stats['avg_duration']:.3f}s")

回调

def on_complete(result):
    print(f"Task done: {result['task']}")

agent.add_callback(on_complete)

MCP工具使用

result = agent.use_tool("file_ops", "read", {"path": "/data.txt"})

系统状态

status = system.get_system_status()
print(status)

MCP服务器统计信息

stats = server.get_stats()
print(f"Total executions: {stats['total_executions']}")
print(f"Success rate: {stats['success_rate']:.2%}")
print(f"Avg duration: {stats['avg_duration']:.3f}s")
print(f"Cache size: {stats['cache_size']}")

MCP服务器挂钩

def before_hook(tool_name, params):
    print(f"Executing {tool_name}")

def after_hook(log_entry):
    print(f"Completed in {log_entry['duration']:.3f}s")

server.add_hook("before", before_hook)
server.add_hook("after", after_hook)

代理指标

metrics = agent.get_metrics()
print(f"Total tasks: {metrics['total_tasks']}")
print(f"Success rate: {metrics['success_rate']:.2%}")
print(f"Avg duration: {metrics['avg_duration']:.3f}s")

建筑

MultiAgentSystem
├── Agent (researcher)
│   ├── Memory
│   ├── Callbacks
│   └── MCP Servers
│       └── web_operations
├── Agent (coder)
│   └── MCP Servers
│       └── file_operations
└── Agent (analyst)
    └── MCP Servers
        ├── file_operations
        └── web_operations

组件

代理商(agent.py)

具有角色、功能、内存和MCP服务器访问权限的个人AI助手。

MCP服务器(mcp_server.py)

使用日志和统计信息注册和执行函数的工具提供程序。

多 代理 系统multi_agent_system.py)

通过委派、并行执行和共享状态协调多个代理。

工作流程(workflow.py)

通过依赖关系解析管理顺序任务执行。

运行示例

基础示例

python example.py

高级功能

python advanced_example.py

高级用法

Agent学习

from learning import AgentLearning

learning = AgentLearning(agent)
learning.learn_from_feedback("Research task", "Excellent", 5.0)
best_tasks = learning.get_best_task_types()

# Export/import knowledge
knowledge = learning.export_knowledge()
learning.import_knowledge(knowledge)

协作

from collaboration import AgentCollaboration

collab = AgentCollaboration(system)

# Negotiate best agent
best = collab.negotiate(["agent1", "agent2"], "Complex task")

# Collaborate on task
result = collab.collaborate(["agent1", "agent2"], "Team task")

# Voting
choice = collab.vote(["agent1", "agent2"], "Pick option", ["A", "B", "C"])

事件总线

from event_bus import EventBus

bus = EventBus()

# Subscribe to events
bus.subscribe("task_complete", lambda e: print(e))

# Publish events
bus.publish("task_complete", {"task": "done"})

# Query event history
events = bus.get_events(event_type="task_complete", since=time.time()-3600)

调度

from scheduler import TaskScheduler

scheduler = TaskScheduler(system)

# Delayed task (run after 10 seconds)
scheduler.schedule("agent", "Task", delay=10)

# Recurring task (run every 60 seconds)
scheduler.schedule_recurring("agent", "Task", interval=60)

# Execute pending tasks
results = scheduler.run_pending()

# Cancel recurring task
scheduler.cancel_recurring(0)

负载平衡

from resilience import LoadBalancer

balancer = LoadBalancer(system)

# Set strategy: round_robin, least_busy, best_performance
balancer.set_strategy("best_performance")
agent = balancer.select_agent("capability")

断路器

from resilience import CircuitBreaker

breaker = CircuitBreaker(failure_threshold=5, timeout=60)

# Protected call
result = breaker.call(system.delegate, "agent", "task")
print(f"Circuit state: {breaker.state}")  # closed, open, half-open

速率限制器

from resilience import RateLimiter

limiter = RateLimiter(max_requests=10, window=60)

if limiter.allow():
    # Process request
    pass
else:
    wait = limiter.wait_time()
    print(f"Rate limited, wait {wait:.1f}s")

状态管理

from persistence import StateManager

manager = StateManager("state.json")

# Save current state
manager.save_state(system)

# Create checkpoint
manager.checkpoint(system, "backup1")

# Load state
state = manager.load_state()

# Restore from checkpoint
state = manager.restore_checkpoint("backup1")

指标收集

from persistence import MetricsCollector

metrics = MetricsCollector()

# Record task execution
metrics.record("agent", result)

# Get summary
summary = metrics.get_summary()
print(f"Success rate: {summary['success_rate']:.2%}")
print(f"Avg duration: {summary['avg_duration']:.3f}s")

# Reset metrics
metrics.reset()

API 参考

代理

Agent(name, role, mcp_servers=[], capabilities=[], priority=0)
  .remember(key, value, ttl=None)
  .recall(key) -> value
  .forget(key)
  .clear_memory()
  .use_tool(server_name, tool_name, params)
  .process(task, context=None)
  .get_metrics() -> dict
  .get_available_tools() -> dict

MCP服务器

MCPServer(name, description="", rate_limit=None)
  .register_tool(name, func, description="", params_schema=None, cacheable=False)
  .add_hook(hook_type, func)  # "before" or "after"
  .execute(tool_name, params, use_cache=True)
  .clear_cache()
  .get_stats() -> dict
  .list_tools() -> list

MultiAgent 系统

MultiAgentSystem(max_workers=10)
  .add_agent(agent)
  .remove_agent(agent_name)
  .delegate(agent_name, task, context=None, priority=0)
  .auto_delegate(task, capability, context=None)
  .parallel_execute(tasks, timeout=None)
  .broadcast(task)
  .find_agent_by_capability(capability)
  .get_best_agent(capability)
  .share_data(key, value, ttl=None)
  .get_shared_data(key)
  .get_system_status() -> dict
  .get_agent_leaderboard() -> list
  .add_middleware(middleware_func)

工作流程

Workflow(name, max_retries=3)
  .add_step(agent_name, task, depends_on=[], condition=None)
  .add_error_handler(step_id, handler)
  .execute(system, parallel=False)
  .get_stats() -> dict

许可证

麻省理工学院