内容
# mini-crew
一款轻量级的多智能体框架 — **类似 CrewAI 的角色编排 + 事件驱动的流程引擎**，纯 Python 实现。定义智能体，赋予它们工具，并在团队或事件驱动的流程中运行。支持 MCP 协议进行工具链集成。
## 快速开始
```bash
pip install -e .
```
在 `.env` 文件中设置您的 API 密钥：
```
DEEPSEEK_API_KEY=your-key-here
DEEPSEEK_MODEL=deepseek-v4-pro
```
### 5 分钟示例
```python
from mini_crew import Agent, Task, Crew, Process, tool
@tool
def calculator(expression: str) -> str:
"""安全地评估数学表达式。"""
import math
safe = {"abs": abs, "sqrt": math.sqrt, "pi": math.pi}
return str(eval(expression, {"__builtins__": {}}, safe))
agent = Agent(
name="数学助手",
role="数学专家",
goal="准确进行计算",
backstory="您使用计算器工具进行精确数学运算。",
tools=[calculator],
verbose=True,
)
task = Task(
description="计算 (123 + 456) * 789",
expected_output="数值结果",
agent=agent,
)
crew = Crew(agents=[agent], tasks=[task], process=Process.sequential)
result = crew.kickoff()
print(result.results[0]["output"])
```
## 您将获得
四个基本组件。
**智能体** — 一个具有角色、目标、背景故事和一组工具的 LLM 驱动的工作者。它调用 LLM，通过函数调用使用工具，并生成结果。
```python
agent = Agent(
name="研究员",
role="信息收集者",
goal="收集和总结信息",
backstory="经验丰富的研究员，细致入微。",
llm="deepseek-v4-pro",
tools=[web_search, calculator],
verbose=True,
)
result = agent.execute("法国的 GDP 是多少？")
```
**任务** — 与智能体绑定的工作单元。任务可以通过 `context` 依赖于其他任务，您可以将其标记为 `async_execution=True` 以与兄弟任务并行运行。
```python
task = Task(
description="分析季度报告",
expected_output="包含关键指标的 JSON 摘要",
agent=analyst,
tools=[calculator],
context=[gather_task, clean_task],
async_execution=True,
)
```
**团队** — 连接智能体和任务。两种模式：
```python
# 顺序执行：DAG 拓扑排序，分层执行，并行兄弟节点
crew = Crew(agents=[a1, a2, a3], tasks=[t1, t2, t3], process=Process.sequential)
# 分层：管理智能体通过 LLM 委托
crew = Crew(agents=[a1, a2, a3], tasks=[t1, t2, t3], process=Process.hierarchical)
```
**流程** — 事件驱动的管道。装饰函数以连接 DAG。基于 SQLite 的状态持久化，支持检查点和恢复。
```python
from mini_crew import Flow
flow = Flow(name="数据管道")
@flow.start
def gather():
return fetch_raw_data()
@flow.listen(gather)
def clean(raw):
return preprocess(raw)
@flow.router(clean)
def decide(cleaned):
if cleaned["quality"] > 0.8:
return "publish"
return "review"
@flow.listen("publish")
def publish(data):
return upload(data)
results = flow.run()
```
流程装饰器：`@flow.start`（入口），`@flow.listen(source)`（响应上游），`@flow.router(source)`（分支到命名下游），`@flow.checkpoint`（人工审批门控 — 使用 `resume_from_checkpoint()` 恢复）。
## 两个引擎，选择一个
mini-crew 提供两个执行引擎。它们解决不同的问题。
| 引擎 | 适合 | 工作原理 |
|--------|----------|--------------|
| **CrewAI 模式** | 线性管道，固定智能体到任务分配 | 顺序或分层，DAG 拓扑排序，并行兄弟节点执行 |
| **流程模式** | 条件分支，人机协作，事件链 | 事件驱动，状态图，SQLite 持久化，检查点/恢复 |
```python
from mini_crew import Crew, Engine
# 根据任务结构自动检测
crew = Crew(agents=[...], tasks=[...], engine=Engine.AUTO)
# 或强制选择
crew = Crew(agents=[...], tasks=[...], engine=Engine.LANGGRAPH)
```
流程模式在底层使用类似 LangGraph 的状态图 — 条件、循环、检查点节点。Crew 模式更简单：定义智能体，分配任务，选择顺序或分层。
两者共享相同的智能体/任务/工具基元，因此切换成本较低。
## 工具与 MCP
使用装饰器定义本地工具：
```python
@tool
def web_search(query: str) -> str:
"""搜索网页信息。"""
# ... 实现
```
工具在全局 `ToolRegistry` 中自动注册，并转换为 OpenAI 函数调用模式。内置工具：`calculator`、`web_search`、`read_file`、`write_file`、`list_files`、`rag_search`。
### MCP — 远程工具像本地工具一样
通过 MCP 暴露工具，并从任何地方消费：
```python
# 服务器：暴露工具
from mini_crew.mcp import MCPServer
server = MCPServer()
server.register_tool(calculator)
server.register_tool(web_search)
server.serve(port=9000)
# 客户端：发现并使用
from mini_crew.mcp import MCPClient
client = MCPClient("http://localhost:9000")
tools = client.discover_tools()
# -> [{"name": "calculator", "schema": {...}}, ...]
agent = Agent(name="研究员", tools=[local_tool, *tools])
```
MCP 工具与本地工具的工作方式相同 — 智能体不关心工具位于何处。
## 示例
```bash
python examples/travel_planner/main.py # 3 智能体顺序团队
python examples/code_review/main.py # 3 智能体顺序团队
python examples/market_research/main.py # 流程 + LangGraph 路由
```
### 旅行规划器（快速浏览）
```python
planner = Agent(name="规划师", role="旅行规划师", goal="设计行程", tools=[calculator])
reviewer = Agent(name="审查员", role="质量检查员", goal="验证可行性")
writer = Agent(name="撰写者", role="报告撰写者", goal="生成最终文档")
plan_task = Task(description="计划 3 天东京之旅", agent=planner)
review_task = Task(description="审查计划中的问题", agent=reviewer, context=[plan_task])
write_task = Task(description="撰写最终行程", agent=writer, context=[review_task])
crew = Crew(agents=[planner, reviewer, writer], tasks=[plan_task, review_task, write_task])
crew.kickoff()
```
## 项目布局
```
mini-crew/
├── src/mini_crew/
│ ├── agent.py # 智能体执行循环 + 工具调用
│ ├── task.py # 任务 DAG + 拓扑排序 + 异步并行
│ ├── crew.py # 团队编排（顺序，分层）
│ ├── flow.py # 流程引擎：EventBus，SQLite 状态，检查点
│ ├── langgraph_engine.py # LangGraph 状态图引擎
│ ├── llm.py # LLM 适配器（OpenAI SDK → DeepSeek）
│ ├── tool.py # ToolRegistry + @tool 装饰器
│ ├── mcp/ # MCP 服务器和客户端
│ ├── types.py # Pydantic 模型
│ ├── memory.py # 短期/长期记忆
│ └── utils.py
├── tools/ # 内置工具
├── examples/ # 演示项目
├── tests/ # 84 个测试
└── docs/ # 架构文档
```
## API 一览
| 类型/函数 | 描述 |
|----------------|------------|
| `Agent` | LLM 工作者：角色，目标，背景故事，工具，`execute()` |
| `Task` | 工作单元：描述，预期输出，智能体绑定，`async_execution` |
| `Crew` | 编排器：`Process.sequential` 或 `Process.hierarchical`，`kickoff()` |
| `Flow` | 事件管道：`@start`，`@listen`，`@router`，`@checkpoint` |
| `@tool` | 装饰器，在 ToolRegistry 中自动注册 |
| `dag_sort(tasks)` | 拓扑排序，返回并行执行的层 |
| `execute_parallel(tasks)` | 异步并行执行同层任务 |
| `ToolRegistry` | 全局单例：注册，执行，获取模式 |
| `MCPClient(url)` | 发现并调用远程工具 |
| `MCPServer()` | 通过 MCP 暴露工具 |
| `Process` | 枚举：`sequential`，`hierarchical` |
| `Engine` | 枚举：`CREWAI`，`LANGGRAPH`，`AUTO` |
| `AgentResult` | 内容，工具调用，轮次，成功，错误 |
| `TaskResult` | 输出，智能体结果，成功，错误 |
| `CrewResult` | 结果，摘要，成功，总轮次，总工具调用 |
## 运行测试
```bash
pytest tests/ -v
# 84 passed in ~2s
```
## 技术栈
Python 3.10+ · Pydantic v2 · OpenAI SDK · LangGraph · MCP 协议 · asyncio · loguru · SQLite · pytest
---
*郭应学 · 江西农业大学 · 2025–2026*
