语言图谱多智能体模板框架

Name: 语言图谱多智能体模板框架
Author: mrgoonie

作者：mrgoonie3 星标更新：2025-07-05

安装使用来源：spider==github.com

项目介绍

LangGraph 多代理基础模板

一个强大的基础模板，用于高效地使用LangGraph构建具有监督架构的AI代理集群，并集成模型上下文协议（MCP）和全面的API。

🌟 功能

多代理架构：通过监督协调构建AI代理集群
LangGraph 集成：利用LangGraph的强大状态管理进行代理工作流
MCP 支持：通过MCP服务器集成工具
流式API：实时流式响应以支持交互式对话
数据库持久化：在PostgreSQL中存储对话、代理状态和活动日志
云存储：使用Cloudflare R2进行文件管理
全面API：带有Swagger文档的RESTful端点，使用FastAPI
安全性：认证中间件、错误处理和最佳安全实践

🚀 快速开始

先决条件

Python 3.10+
PostgreSQL
Cloudflare R2账户（可选，用于云存储）
OpenRouter AI API密钥（或其他兼容的AI提供商）

安装

克隆仓库

git clone https://github.com/yourusername/langgraph-multiagent-boilerplate.git
cd langgraph-multiagent-boilerplate

设置Python虚拟环境

python -m venv venv
source venv/bin/activate  # 在Windows上：venv\Scripts\activate

安装依赖

pip install -r requirements.txt

配置环境变量

cp .env.example .env
# 编辑.env文件以设置您的设置（数据库、API密钥等）

设置数据库

# 创建一个PostgreSQL数据库
# 然后运行迁移（一旦实现）

运行服务器

uvicorn app.main:app --reload

访问API文档

Swagger UI: http://localhost:8000/api/docs
ReDoc: http://localhost:8000/api/redoc

📋 项目结构

langgraph-multiagent-boilerplate/
├── app/
│   ├── api/
│   │   ├── exceptions.py       # 错误处理
│   │   ├── middleware/         # 安全与认证中间件
│   │   └── routes/             # API端点
│   ├── core/
│   │   ├── config.py           # 配置管理
│   │   └── langgraph/          # LangGraph组件
│   ├── db/
│   │   └── base.py             # 数据库设置
│   ├── models/                 # SQLAlchemy模型
│   ├── schemas/                # Pydantic模式
│   ├── services/               # 业务逻辑
│   └── main.py                 # 应用程序入口点
├── tests/                      # 测试套件
├── .env.example                # 环境模板
├── pyproject.toml             # Python项目元数据
├── requirements.txt           # 依赖项
├── README.md                  # 此文件
├── PROJECT_OVERVIEW.md        # 详细的项目文档
└── IMPLEMENTATION_TASKS.md    # 开发路线图

🧠 工作原理

多代理系统架构

AI团队：每个AI代理集群包含多个团队，每个团队由一个监督代理领导
监督架构：监督代理分析用户输入，创建计划并分配任务给其他代理
工具集成：代理可以通过MCP服务器访问外部工具
流式通信：实时响应事件流
持久性：所有对话、状态和活动都存储在数据库中

示例流程

用户向一个团队发送消息
监督代理通过API调用接收输入
监督代理分析输入和团队的能力
监督代理直接回答或创建详细计划
如果需要，监督代理将任务分配给专门的代理
代理使用连接的MCP工具执行任务
监督代理收集结果，分析它们并制定回复
回复被流式传输回用户

🔌 API参考

核心端点

团队和代理

GET /api/crews - 列出所有团队
POST /api/crews - 创建新团队
GET /api/crews/{crew_id} - 获取团队详情
PUT /api/crews/{crew_id} - 更新团队
DELETE /api/crews/{crew_id} - 删除团队
GET /api/agents - 列出所有代理
POST /api/agents - 创建新代理
GET /api/agents/{agent_id} - 获取代理详情
PUT /api/agents/{agent_id} - 更新代理
DELETE /api/agents/{agent_id} - 删除代理

对话

GET /api/conversations - 列出对话
POST /api/conversations - 创建新对话
GET /api/conversations/{conversation_id} - 获取对话详情
POST /api/conversations/{conversation_id}/chat - 发送消息并获取回复
POST /api/conversations/{conversation_id}/chat/stream - 获取流式回复

请参阅Swagger文档以获取完整的API参考。

📝 使用示例

创建带有代理的团队

import httpx

# 创建新团队
crew_data = {
    "name": "研究团队",
    "description": "专注于研究任务的团队",
    "metadata": {"specialization": "研究"}
}

response = httpx.post("http://localhost:8000/api/crews", json=crew_data)
crew = response.json()
crew_id = crew["id"]

# 创建监督代理
supervisor_data = {
    "crew_id": crew_id,
    "name": "研究监督员",
    "description": "监督研究操作",
    "system_prompt": "您是负责协调研究工作的研究监督员。",
    "model": "google/gemini-2.5-flash",
    "is_supervisor": True,
    "metadata": {}
}

httpx.post("http://localhost:8000/api/agents", json=supervisor_data)

# 创建专门的代理
web_researcher_data = {
    "crew_id": crew_id,
    "name": "网络研究员",
    "description": "专长于网络研究",
    "system_prompt": "您是一位在线查找准确信息的网络研究员。",
    "model": "claude-3-sonnet",
   _ "is_supervisor": False,
    "metadata": {"specialty": "web_search"}
}

httpx.post("http://localhost:8000/api/agents", json=web_researcher_data)

启动对话

# 与团队创建对话
conversation_data = {
    "user_id": "user123",
    "crew_id": crew_id,
    "title": "关于AI趋势的研究"
}

response = httpx.post("http://localhost:8000/api/conversations", json=conversation_data)
conversation = response.json()
conversation_id = conversation["id"]

# 向团队发送消息
message_data = {
    "message": "多代理AI系统的最新趋势是什么？",
    "metadata": {}
}

# 对于非流式回复
response = httpx.post(
    f"http://localhost:8000/api/conversations/{conversation_id}/chat", 
    json=message_data
)
print(response.json()["content"])

# 对于流式回复
with httpx.stream(
    "POST",
    f"http://localhost:8000/api/conversations/{conversation_id}/chat/stream",
    json=message_data,
    timeout=60.0
) as response:
    for chunk in response.iter_lines():
        if chunk.startswith("data: "):
            data = json.loads(chunk[6:])
            if "choices" in data and data["choices"][0]["delta"].get("content"):
                print(data["choices"][0]["delta"]["content"], end="")

🧪 测试

运行测试套件：

pytest

🔧 配置

关键环境变量：

DATABASE_URL：PostgreSQL连接字符串
OPENROUTER_API_KEY：OpenRouter API密钥
MCP_SERVER_URL：MCP服务器的URL
R2_ENDPOINT、R2_BUCKET_NAME等：Cloudflare R2配置
JWT_SECRET_KEY：JWT认证的秘密
DEBUG：启用调试模式

请参阅.env.example以获取完整的配置选项列表。

🧩 扩展基础模板

添加新的MCP工具

在数据库中注册一个新的MCP服务器
发现并从服务器注册工具
将工具分配给代理

创建自定义代理类型

创建具有专业系统提示的新代理
将相关的MCP工具分配给代理
将代理添加到团队

实现自定义工作流

修改app/core/langgraph/supervisor.py中的监督逻辑
调整状态图以实现自定义工作流

🤝 贡献

欢迎贡献！请随时提交Pull Request。

📄 许可证

此项目根据MIT许可证发布 - 请参阅LICENSE文件了解详细信息。