Metadata-Version: 2.4
Name: test-ai-one
Version: 0.0.1
Summary: 一个用于学习 Python 公共库构建和发布的示例工程
Requires-Python: >=3.14
Requires-Dist: aiofiles==25.1.0
Requires-Dist: anyio==4.14.0
Requires-Dist: certifi==2026.6.17
Requires-Dist: h11==0.16.0
Requires-Dist: httpcore==1.0.9
Requires-Dist: httpx==0.28.1
Requires-Dist: idna==3.18
Requires-Dist: markdown==3.10.2
Requires-Dist: packaging==26.2
Requires-Dist: python-dateutil==2.9.0.post0
Requires-Dist: python-dotenv==1.2.2
Requires-Dist: six==1.17.0
Description-Content-Type: text/markdown

# 多智能体协同调研系统代码阅读指南

这是一个用于演示 Python 异步编程的轻量级调研项目。用户在命令行输入一个调研主题后，程序会调用大模型把主题拆成多个子任务，再并发执行这些子任务，最后合并结果、生成 Markdown 调研报告并保存到本地。

项目代码规模较小，但模块边界比较清晰，适合按“入口 -> 编排 -> 业务节点 -> 底层工具”的顺序学习。

## 一、项目功能概览

项目的核心流程如下：

1. 读取用户输入的调研需求。
2. 调用大模型，将调研需求分解成 3-5 个相互独立的子任务。
3. 使用 `asyncio.gather` 并发执行所有子任务。
4. 将各子任务结果拼接成中间材料。
5. 再次调用大模型，把中间材料整理成完整 Markdown 报告。
6. 使用异步文件 I/O 将报告保存到 `output` 目录。

整体上它是一个“多智能体协同调研”的教学型项目，其中每个子任务可以理解为一个独立的调研助手。

## 二、目录结构

```text
.
├── README.md
├── requirements.txt
├── .env.example
├── output/
└── src/
    ├── __init__.py
    ├── main.py
    ├── orchestrator.py
    ├── decomposer.py
    ├── worker.py
    ├── merger.py
    ├── summarizer.py
    ├── reporter.py
    ├── ai_tools.py
    └── config.py
```

## 三、运行方式

### 1. 安装依赖

```bash
pip install -r requirements.txt
```

项目依赖：

```text
httpx>=0.27.0
python-dotenv>=1.0.0
aiofiles>=24.1.0
```

### 2. 配置环境变量

复制 `.env.example` 为 `.env`，然后填入真实配置：

```bash
cp .env.example .env
```

`.env.example` 中包含以下配置：

```env
API_BASE=https://api.openai.com/v1
MODEL_NAME=gpt-3.5-turbo
API_KEY=your_api_key_here
OUTPUT_DIR=./output
```

配置说明：

| 变量名 | 作用 | 默认值 |
| --- | --- | --- |
| `API_BASE` | OpenAI 兼容接口地址 | `https://api.openai.com/v1` |
| `MODEL_NAME` | 调用的模型名称 | `gpt-3.5-turbo` |
| `API_KEY` | API 密钥，必须配置 | 空字符串 |
| `OUTPUT_DIR` | 报告输出目录 | `./output` |

### 3. 启动程序

```bash
python -m src.main
```

运行后根据提示输入调研需求，例如：

```text
请输入调研需求: 调研 Python 异步编程在 Web 服务中的应用
```

报告会保存到 `OUTPUT_DIR` 指定的目录中，默认是 `./output`。

## 四、模块职责

| 模块 | 主要职责 | 关键函数/变量 |
| --- | --- | --- |
| `src/main.py` | 命令行入口，读取用户输入，启动异步主流程 | `main()` |
| `src/orchestrator.py` | 流程编排层，串联任务分解、并发执行、合并、总结、保存 | `run_pipeline()` |
| `src/decomposer.py` | 调用大模型进行任务分解，并解析 JSON 数组结果 | `decompose()`、`_parse_json_array()` |
| `src/worker.py` | 执行单个调研子任务 | `execute_subtask()` |
| `src/merger.py` | 将子任务、子任务结果合并成 Markdown 中间材料 | `merge_results()` |
| `src/summarizer.py` | 调用大模型生成最终调研报告 | `summarize()` |
| `src/reporter.py` | 创建输出目录，并异步写入 Markdown 报告文件 | `save_report()` |
| `src/ai_tools.py` | OpenAI 兼容 Chat Completions 流式调用封装 | `completions()`、`get_text()` |
| `src/config.py` | 加载 `.env` 配置并暴露项目配置常量 | `API_BASE`、`MODEL_NAME`、`API_KEY`、`OUTPUT_DIR` |

## 五、模块依赖关系

从调用方向看，依赖关系如下：

```text
src/main.py
└── src/orchestrator.py
    ├── src/decomposer.py
    │   └── src/ai_tools.py
    │       └── src/config.py
    ├── src/worker.py
    │   └── src/ai_tools.py
    │       └── src/config.py
    ├── src/merger.py
    ├── src/summarizer.py
    │   └── src/ai_tools.py
    │       └── src/config.py
    └── src/reporter.py
        └── src/config.py
```

更抽象地看，可以分成四层：

```text
入口层
  main.py

编排层
  orchestrator.py

业务节点层
  decomposer.py
  worker.py
  merger.py
  summarizer.py
  reporter.py

基础设施层
  ai_tools.py
  config.py
```

### 外部依赖关系

| 外部库 | 被哪个模块使用 | 用途 |
| --- | --- | --- |
| `httpx` | `src/ai_tools.py` | 异步 HTTP 客户端，请求 Chat Completions 接口 |
| `python-dotenv` | `src/config.py` | 读取 `.env` 文件中的环境变量 |
| `aiofiles` | `src/reporter.py` | 异步写入报告文件 |
| Python 标准库 `asyncio` | `src/main.py`、`src/orchestrator.py` | 启动事件循环、并发执行子任务 |
| Python 标准库 `json` | `src/ai_tools.py`、`src/decomposer.py` | 解析流式响应和任务分解结果 |
| Python 标准库 `re` | `src/decomposer.py` | 从非标准输出中提取 JSON 数组 |
| Python 标准库 `os` | `src/config.py`、`src/reporter.py` | 读取环境变量、创建输出目录 |
| Python 标准库 `datetime` | `src/reporter.py` | 生成报告文件时间戳 |

## 六、主流程时序

一次完整运行的函数调用顺序如下：

```text
python -m src.main
    |
    v
main()
    |
    v
run_pipeline(topic)
    |
    +--> decompose(topic)
    |       |
    |       +--> get_text(messages)
    |               |
    |               +--> completions(messages)
    |
    +--> asyncio.gather(...)
    |       |
    |       +--> execute_subtask(index, total, subtask)
    |               |
    |               +--> get_text(messages)
    |                       |
    |                       +--> completions(messages)
    |
    +--> merge_results(topic, subtasks, results)
    |
    +--> summarize(topic, merged)
    |       |
    |       +--> get_text(messages)
    |               |
    |               +--> completions(messages)
    |
    +--> save_report(topic, report)
            |
            +--> aiofiles.open(...).write(...)
```

## 七、推荐阅读顺序

### 第 1 步：先看入口和总流程

建议先读：

1. `src/main.py`
2. `src/orchestrator.py`

阅读目标：

- 理解程序如何启动。
- 理解 `asyncio.run(main())` 的作用。
- 理解 `run_pipeline()` 如何把整个调研流程拆成 4 个步骤。
- 重点关注 `asyncio.gather(..., return_exceptions=True)`，这是项目异步并发的核心。

`orchestrator.py` 是全项目最适合先读的文件，因为它展示了所有模块之间的协作方式。

### 第 2 步：看任务分解模块

继续读：

1. `src/decomposer.py`

阅读目标：

- 理解系统提示词 `_DECOMPOSE_SYSTEM_PROMPT` 如何约束大模型输出。
- 理解 `decompose()` 如何调用 `get_text()`。
- 理解 `_parse_json_array()` 为什么要兼容代码块、完整 JSON、夹杂文本的 JSON 以及普通行文本。

这个模块能帮助你理解：实际调用大模型时，不能完全假设输出一定是严格 JSON，所以项目做了较宽松的解析兜底。

### 第 3 步：看并发执行子任务

继续读：

1. `src/worker.py`

阅读目标：

- 理解每个子任务如何被包装成 messages。
- 理解它只负责单个任务，不负责并发。
- 理解并发是在 `orchestrator.py` 里用 `asyncio.gather` 触发的。

这里要区分两个概念：

- `worker.py` 负责“怎么执行一个任务”。
- `orchestrator.py` 负责“同时执行多个任务”。

### 第 4 步：看中间结果合并和最终总结

继续读：

1. `src/merger.py`
2. `src/summarizer.py`

阅读目标：

- 理解 `merge_results()` 只是本地字符串拼接，不调用模型。
- 理解 `summarize()` 会把合并后的中间材料再次交给大模型，让模型生成最终报告。

`merger.py` 和 `summarizer.py` 是两个阶段：

- `merger.py`：整理原始材料。
- `summarizer.py`：生成面向用户的最终报告。

### 第 5 步：看报告保存

继续读：

1. `src/reporter.py`

阅读目标：

- 理解 `OUTPUT_DIR` 的来源。
- 理解如何生成安全文件名。
- 理解 `aiofiles.open` 的异步文件写入方式。

这个模块的职责很单一：把最终 Markdown 内容保存成文件。

### 第 6 步：最后看底层 AI 调用和配置

最后读：

1. `src/config.py`
2. `src/ai_tools.py`

阅读目标：

- 理解 `.env` 如何加载。
- 理解 `API_BASE`、`MODEL_NAME`、`API_KEY` 如何被读取。
- 理解 `ai_tools.py` 如何调用 OpenAI 兼容的 `/chat/completions` 接口。
- 理解流式响应中 `data: ...` 和 `[DONE]` 的处理逻辑。

`ai_tools.py` 是底层基础设施模块。刚开始读项目时可以先知道它“负责调用模型”即可，等理解业务流程后再回头细看。

## 八、异步编程重点

这个项目里最值得学习的异步点有三个：

### 1. 事件循环入口

```python
if __name__ == "__main__":
    asyncio.run(main())
```

`asyncio.run()` 会创建事件循环，执行异步入口函数 `main()`，并在结束后关闭事件循环。

### 2. 并发执行多个子任务

```python
raw_results = await asyncio.gather(
    *[execute_subtask(i + 1, len(subtasks), task) for i, task in enumerate(subtasks)],
    return_exceptions=True,
)
```

这里会把多个子任务同时发起，而不是一个执行完再执行下一个。由于每个子任务主要在等待网络请求结果，所以非常适合用异步并发提升整体速度。

`return_exceptions=True` 的作用是：某个子任务失败时，不会让整个 `gather` 直接抛异常中断，而是把异常作为结果返回，后续统一处理。

### 3. 异步网络和异步文件 I/O

`src/ai_tools.py` 中使用：

```python
httpx.AsyncClient(...)
```

`src/reporter.py` 中使用：

```python
aiofiles.open(...)
```

这两个模块分别展示了异步网络请求和异步文件写入。

## 九、数据流说明

核心数据从用户输入到最终文件，大致经历下面的变化：

```text
topic: str
  |
  v
subtasks: list[str]
  |
  v
results: list[str]
  |
  v
merged: str
  |
  v
report: str
  |
  v
filepath: str
```

对应模块：

| 数据 | 产生位置 | 说明 |
| --- | --- | --- |
| `topic` | `main.py` | 用户输入的原始调研需求 |
| `subtasks` | `decomposer.py` | 大模型拆解后的子任务列表 |
| `results` | `worker.py` | 每个子任务的调研结果 |
| `merged` | `merger.py` | 合并后的 Markdown 中间材料 |
| `report` | `summarizer.py` | 最终 Markdown 报告 |
| `filepath` | `reporter.py` | 保存后的文件绝对路径 |

## 十、值得注意的实现细节

### 1. `ai_tools.py` 在导入时检查 API_KEY

`src/ai_tools.py` 中有如下逻辑：

```python
if not API_KEY:
    print("错误: 请在 .env 文件中配置 API_KEY")
    sys.exit(1)
```

这意味着只要导入了 `src.ai_tools`，如果没有配置 `API_KEY`，程序就会直接退出。由于 `decomposer.py`、`worker.py`、`summarizer.py` 都会导入 `get_text`，所以这些模块也间接受这个检查影响。

### 2. `decomposer.py` 对模型输出做了容错

模型可能输出：

- 标准 JSON 数组。
- 包在 Markdown 代码块中的 JSON。
- 文本中夹带 JSON 数组。
- 多行文本列表。

`_parse_json_array()` 按上述顺序尝试解析，最后才退化成普通行文本列表。

### 3. `merger.py` 中有一个轻微冗余判断

`orchestrator.py` 已经把异常转换成了字符串：

```python
if isinstance(r, BaseException):
    results.append(f"[执行失败] {r}")
```

因此 `merge_results()` 里的这段判断通常不会再命中：

```python
if isinstance(result, Exception):
    lines.append(f"该子任务执行失败: {result}\n")
```

这不影响运行，只是从代码整洁度上看略有重复。

### 4. 任务分解数量依赖提示词约束

项目通过 prompt 要求模型将任务拆成 3-5 个子任务，但代码本身没有强制限制数量。如果模型返回更多或更少子任务，程序仍会照常执行。

### 5. 目前没有重试机制

如果模型接口偶发失败，`execute_subtask()` 的异常会被 `asyncio.gather(..., return_exceptions=True)` 收集并写入结果。但 `decompose()` 和 `summarize()` 失败时会直接中断流程，因为它们不在类似的异常兜底里。

## 十一、适合练习和扩展的方向

如果你想通过这个项目继续练习，可以按下面顺序扩展：

1. 给 `ai_tools.py` 增加请求失败重试机制。
2. 给 `decompose()` 增加子任务数量上限和空结果兜底。
3. 为 `get_text()` 增加非流式调用模式。
4. 把执行过程中的 token、耗时、失败原因记录到日志。
5. 增加命令行参数，例如 `--topic`、`--output-dir`、`--model`。
6. 增加单元测试，优先测试 `_parse_json_array()` 和 `merge_results()`。
7. 将多个 worker 的并发数量限制为固定值，避免一次性创建过多请求。

## 十二、最短学习路线

如果时间有限，建议按这个顺序快速阅读：

```text
1. src/main.py
2. src/orchestrator.py
3. src/decomposer.py
4. src/worker.py
5. src/merger.py
6. src/summarizer.py
7. src/reporter.py
8. src/config.py
9. src/ai_tools.py
```

其中最关键的三个文件是：

```text
src/orchestrator.py
src/decomposer.py
src/ai_tools.py
```

读懂这三个文件后，就基本掌握了项目的主流程、任务分解逻辑和底层模型调用方式。
