Langgraph Multiagent Boilerplate

LangGraph多试剂煮沸板

一个强大的样板，用于使用LangGraph高效构建人工智能代理集群，具有监管架构、模型上下文协议（MCP）集成和全面的API。

🌟 特性

• 多代理架构：在主管协调下构建人工智能代理集群
• LangGraph集成：利用LangGraph强大的状态管理实现代理工作流
• MCP支持：通过模型上下文协议服务器集成工具
• 流式API：交互式对话的实时流式响应
• 数据库持久性：将对话、代理状态和活动日志存储在PostgreSQL中
• 云存储：使用Cloudflare R2进行文件管理
• 综合API：带有FastAPI的RESTful端点，包括Swagger文档
• 安全：身份验证中间件、错误处理和安全最佳实践

🚀 入门指南

先决条件

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

安装

1. 克隆存储库

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

1. 设置Python虚拟环境

python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

1. 安装依赖项

pip install -r requirements.txt

1. 配置环境变量

cp .env.example .env
# Edit .env with your settings (database, API keys, etc.)

1. 设置数据库

# Create a PostgreSQL database
# Then run migrations (once implemented)

1. 运行服务器

uvicorn app.main:app --reload

1. 访问API文档

• Swagger用户界面：http://localhost:8000/api/docs
• 重新记录：http://localhost:8000/api/redoc

📋 项目结构

langgraph-multiagent-boilerplate/
├── app/
│   ├── api/
│   │   ├── exceptions.py       # Error handling
│   │   ├── middleware/         # Security & auth middleware
│   │   └── routes/             # API endpoints
│   ├── core/
│   │   ├── config.py           # Configuration management
│   │   └── langgraph/          # LangGraph components
│   ├── db/
│   │   └── base.py             # Database setup
│   ├── models/                 # SQLAlchemy models
│   ├── schemas/                # Pydantic schemas
│   ├── services/               # Business logic
│   └── main.py                 # Application entry point
├── tests/                      # Test suite
├── .env.example                # Environment template
├── pyproject.toml             # Python project metadata
├── requirements.txt           # Dependencies
├── README.md                  # This file
├── PROJECT_OVERVIEW.md        # Detailed project documentation
└── IMPLEMENTATION_TASKS.md    # Development roadmap

🧠 运作原理

多智能体系统架构

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

示例流程

1. 用户向机组人员发送消息
2. 主管代理通过API调用接收输入
3. 主管分析输入和机组人员的能力
4. 主管要么直接回答，要么制定详细的计划
5. 如果需要，主管会将任务分配给专业代理
6. 代理使用附带的MCP工具执行任务
7. 主管收集结果，分析结果，并制定回应
8. 响应被流式传输回用户

🔌 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

# Create a new crew
crew_data = {
    "name": "Research Crew",
    "description": "A crew specialized in research tasks",
    "metadata": {"specialization": "research"}
}

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

# Create a supervisor agent
supervisor_data = {
    "crew_id": crew_id,
    "name": "Research Supervisor",
    "description": "Supervises research operations",
    "system_prompt": "You are a research supervisor responsible for coordinating research efforts.",
    "model": "google/gemini-2.5-flash",
    "is_supervisor": True,
    "metadata": {}
}

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

# Create specialized agents
web_researcher_data = {
    "crew_id": crew_id,
    "name": "Web Researcher",
    "description": "Specializes in web research",
    "system_prompt": "You are a web researcher that finds accurate information online.",
    "model": "claude-3-sonnet",
    "is_supervisor": False,
    "metadata": {"specialty": "web_search"}
}

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

开始对话

# Create a conversation with a crew
conversation_data = {
    "user_id": "user123",
    "crew_id": crew_id,
    "title": "Research on AI trends"
}

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

# Send a message to the crew
message_data = {
    "message": "What are the latest trends in multi-agent AI systems?",
    "metadata": {}
}

# For non-streaming response
response = httpx.post(
    f"http://localhost:8000/api/conversations/{conversation_id}/chat", 
    json=message_data
)
print(response.json()["content"])

# For streaming response
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工具

1. 在数据库中注册新的MCP服务器
2. 从服务器发现和注册工具
3. 为代理分配工具

创建自定义代理类型

1. 使用专门的系统提示创建新代理
2. 将相关MCP工具分配给代理
3. 将代理添加到团队中

实施自定义工作流

1. 修改中的主管逻辑 app/core/langgraph/supervisor.py
2. 调整状态图以实现您的自定义工作流

🤝 贡献

欢迎投稿！请随时提交拉取请求。

📄 许可证

此项目根据MIT许可证获得许可-有关详细信息，请参阅许可证文件。

📚 资源

• Langgraph文档
• 模型上下文协议
• FastAPI文档
• SQLAlchemy文档