# 常见架构模式参考

本文档提供各种架构模式的识别方法和 Mermaid 模板。

## 目录

- [1. Multi-Agent 架构](#1-multi-agent-架构)
- [2. 微服务架构](#2-微服务架构)
- [3. 事件驱动架构](#3-事件驱动架构)
- [4. 数据管道（ETL）](#4-数据管道etl)
- [5. 分层架构（MVC）](#5-分层架构mvc)
- [6. CQRS 架构](#6-cqrs-架构)

---

## 1. Multi-Agent 架构

### 模式 1.1: LangGraph 编排

**代码特征**：
```python
from langgraph.graph import StateGraph, END

# 定义状态
class AgentState(TypedDict):
    query: str
    data: List[str]
    result: str

# 创建图
graph = StateGraph(AgentState)

# 添加节点（Agent）
graph.add_node("researcher", research_agent)
graph.add_node("writer", writing_agent)

# 添加边（流转规则）
graph.add_edge("researcher", "writer")
graph.add_conditional_edges(
    "writer",
    should_continue,
    {
        "continue": "researcher",
        "end": END
    }
)

# 设置入口
graph.set_entry_point("researcher")
```

**识别方法**：
1. 搜索 `from langgraph` 或 `StateGraph`
2. 提取所有 `add_node()` 调用 → Agent 清单
3. 提取所有 `add_edge()` 和 `add_conditional_edges()` → 流转规则
4. 读取每个 Agent 函数的实现

**Mermaid 模板**：
```mermaid
stateDiagram-v2
    [*] --> Researcher: 用户查询
    Researcher --> Writer: 数据收集完成
    Writer --> QualityChecker: 生成草稿

    QualityChecker --> Researcher: 质量不足<br/>评分 < 7
    QualityChecker --> [*]: 质量通过

    note right of Researcher
        文件: agents/researcher.py:45
        职责: 调用搜索 API 收集资料
        输入: ResearchTask
        输出: List[Document]
    end note

    note right of Writer
        文件: agents/writer.py:78
        职责: 基于资料生成报告
        输入: List[Document]
        输出: Report
    end note
```

**关键信息提取**：
- Agent 名称
- 文件位置（文件:行号）
- 职责描述
- 输入/输出类型
- 条件分支逻辑

---

### 模式 1.2: CrewAI 编排

**代码特征**：
```python
from crewai import Agent, Task, Crew

# 定义 Agent
researcher = Agent(
    role="Researcher",
    goal="Collect comprehensive information",
    tools=[search_tool]
)

writer = Agent(
    role="Writer",
    goal="Write engaging articles",
    tools=[writing_tool]
)

# 定义任务
research_task = Task(
    description="Research AI trends",
    agent=researcher
)

writing_task = Task(
    description="Write article based on research",
    agent=writer
)

# 创建 Crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential
)
```

**识别方法**：
1. 搜索 `from crewai` 或 `Agent(`
2. 提取所有 `Agent()` 定义
3. 提取所有 `Task()` 定义
4. 分析 `Crew()` 的 process 类型（sequential/hierarchical）

**Mermaid 模板**：
```mermaid
flowchart TD
    Start([开始]) --> Task1[Research Task<br/>Agent: Researcher]
    Task1 --> Task2[Writing Task<br/>Agent: Writer]
    Task2 --> End([结束])

    style Task1 fill:#e1f5ff
    style Task2 fill:#fff4e1
```

---

### 模式 1.3: AutoGen 多智能体

**代码特征**：
```python
from autogen import AssistantAgent, UserProxyAgent

# 定义 Agent
assistant = AssistantAgent(
    name="assistant",
    llm_config={"config_list": config_list}
)

user_proxy = UserProxyAgent(
    name="user_proxy",
    code_execution_config={"work_dir": "coding"}
)

# 发起对话
user_proxy.initiate_chat(
    assistant,
    message="Analyze this dataset"
)
```

**Mermaid 模板**：
```mermaid
sequenceDiagram
    participant User as UserProxy
    participant Asst as AssistantAgent

    User->>Asst: 发起任务
    Asst->>Asst: 生成代码
    Asst->>User: 返回代码
    User->>User: 执行代码
    User->>Asst: 反馈执行结果
    Asst->>User: 最终答案
```

---

## 2. 微服务架构

### 模式 2.1: Docker Compose 微服务

**代码特征**：
```yaml
# docker-compose.yml
services:
  api-gateway:
    build: ./api-gateway
    ports:
      - "8000:8000"

  user-service:
    build: ./services/user-service
    environment:
      - DATABASE_URL=postgres://...

  order-service:
    build: ./services/order-service

  notification-service:
    build: ./services/notification-service
```

```python
# api-gateway/routes.py
@app.post("/orders")
def create_order(order_data):
    # 调用 user-service
    user = requests.get(f"http://user-service:8001/users/{user_id}")

    # 调用 order-service
    order = requests.post("http://order-service:8002/orders", json=order_data)

    # 调用 notification-service
    requests.post("http://notification-service:8003/notify", json={
        "user_id": user_id,
        "order_id": order["id"]
    })

    return order
```

**识别方法**：
1. 读取 `docker-compose.yml`，提取所有 services
2. 在每个服务的代码中搜索 `requests.get|post`, `httpx.`
3. 提取目标 URL，识别服务间调用关系

**Mermaid 模板**：
```mermaid
graph TB
    Client[客户端] --> Gateway[API Gateway<br/>:8000]

    Gateway -->|GET /users/:id| UserSvc[User Service<br/>:8001]
    Gateway -->|POST /orders| OrderSvc[Order Service<br/>:8002]
    Gateway -->|POST /notify| NotifySvc[Notification Service<br/>:8003]

    UserSvc --> UserDB[(User DB<br/>PostgreSQL)]
    OrderSvc --> OrderDB[(Order DB<br/>PostgreSQL)]

    style Gateway fill:#ff6b6b
    style UserSvc fill:#4ecdc4
    style OrderSvc fill:#45b7d1
    style NotifySvc fill:#f9ca24
```

**关键信息提取**：
- 服务名称
- 端口号
- 服务间调用关系（from → to）
- 数据库连接

---

### 模式 2.2: Kubernetes 微服务

**代码特征**：
```yaml
# k8s/user-service.yaml
apiVersion: v1
kind: Service
metadata:
  name: user-service
spec:
  selector:
    app: user-service
  ports:
    - port: 8001
```

**识别方法**：
1. 查找 `k8s/`, `kubernetes/`, `.yaml` 文件
2. 提取所有 `kind: Service` 定义

---

## 3. 事件驱动架构

### 模式 3.1: 事件总线（Event Bus）

**代码特征**：
```python
# event_bus.py
class EventBus:
    def __init__(self):
        self._listeners = {}

    def on(self, event_name, handler):
        self._listeners.setdefault(event_name, []).append(handler)

    def emit(self, event_name, data):
        for handler in self._listeners.get(event_name, []):
            handler(data)

# handlers.py
@event_bus.on("order_created")
def send_confirmation_email(order):
    email.send(order.customer_email, ...)

@event_bus.on("order_created")
def update_inventory(order):
    inventory.decrease(order.items)

@event_bus.on("order_created")
def log_order(order):
    logger.info(f"Order {order.id} created")

# service.py
def create_order(order_data):
    order = Order(**order_data)
    db.save(order)
    event_bus.emit("order_created", order)  # 发布事件
```

**识别方法**：
1. 搜索 `emit(`, `publish(`, `dispatch(`
2. 搜索 `@on(`, `@subscribe(`, `.on(`
3. 提取事件名称和对应的处理器

**Mermaid 模板**：
```mermaid
graph TB
    OrderService[Order Service] -->|emit| Event[order_created 事件]

    Event --> Handler1[send_confirmation_email<br/>handlers.py:15]
    Event --> Handler2[update_inventory<br/>handlers.py:28]
    Event --> Handler3[log_order<br/>handlers.py:42]

    Handler1 --> EmailSvc[Email Service]
    Handler2 --> InventoryDB[(Inventory DB)]
    Handler3 --> Logger[Logger]

    style Event fill:#ffd93d,stroke:#333,stroke-width:3px
```

---

### 模式 3.2: 消息队列（RabbitMQ/Kafka）

**代码特征**：
```python
# producer.py
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='tasks')

def publish_task(task_data):
    channel.basic_publish(
        exchange='',
        routing_key='tasks',
        body=json.dumps(task_data)
    )

# consumer.py
def callback(ch, method, properties, body):
    task = json.loads(body)
    process_task(task)

channel.basic_consume(
    queue='tasks',
    on_message_callback=callback,
    auto_ack=True
)
```

**Mermaid 模板**：
```mermaid
graph LR
    Producer[任务生产者] -->|发布| Queue[(RabbitMQ<br/>tasks 队列)]
    Queue -->|消费| Consumer1[Worker 1]
    Queue -->|消费| Consumer2[Worker 2]
    Queue -->|消费| Consumer3[Worker 3]

    style Queue fill:#ff6b6b
```

---

## 4. 数据管道（ETL）

### 模式 4.1: 顺序管道

**代码特征**：
```python
def run_pipeline(source_file):
    # Step 1: Extract
    raw_data = extract_from_csv(source_file)

    # Step 2: Transform
    cleaned_data = remove_duplicates(raw_data)
    enriched_data = add_metadata(cleaned_data)
    validated_data = validate_schema(enriched_data)

    # Step 3: Load
    load_to_database(validated_data)
    load_to_s3(validated_data)
```

**Mermaid 模板**：
```mermaid
flowchart LR
    CSV[CSV 文件] --> Extract[Extract<br/>提取数据]
    Extract --> Clean[Clean<br/>去重]
    Clean --> Enrich[Enrich<br/>添加元数据]
    Enrich --> Validate[Validate<br/>校验]
    Validate --> DB[(PostgreSQL)]
    Validate --> S3[(S3 存储)]

    style Extract fill:#e1f5ff
    style Clean fill:#fff4e1
    style Enrich fill:#e8f5e8
    style Validate fill:#ffe1e1
```

---

### 模式 4.2: Airflow DAG

**代码特征**：
```python
from airflow import DAG
from airflow.operators.python import PythonOperator

dag = DAG('data_pipeline', schedule_interval='@daily')

extract_task = PythonOperator(
    task_id='extract',
    python_callable=extract_data,
    dag=dag
)

transform_task = PythonOperator(
    task_id='transform',
    python_callable=transform_data,
    dag=dag
)

load_task = PythonOperator(
    task_id='load',
    python_callable=load_data,
    dag=dag
)

extract_task >> transform_task >> load_task
```

**Mermaid 模板**：
```mermaid
graph LR
    Start([每日触发]) --> Extract[Extract Task]
    Extract --> Transform[Transform Task]
    Transform --> Load[Load Task]
    Load --> End([完成])
```

---

## 5. 分层架构（MVC）

### 模式 5.1: 三层架构

**代码特征**：
```python
# routes.py (Controller 层)
@app.route("/users", methods=["POST"])
def create_user():
    data = request.json
    user = UserService.create(data)  # 调用 Service 层
    return jsonify(user.to_dict())

# services/user_service.py (Service 层)
class UserService:
    @staticmethod
    def create(data):
        # 业务逻辑
        if not validate_email(data["email"]):
            raise ValueError("Invalid email")

        user = User(**data)  # 使用 Model 层
        db.session.add(user)
        db.session.commit()
        return user

# models/user.py (Model 层)
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), unique=True)
    name = db.Column(db.String(80))
```

**Mermaid 模板**：
```mermaid
graph TD
    Client[HTTP 客户端] -->|POST /users| Controller[Controller<br/>routes.py]

    Controller --> Service[Service 层<br/>user_service.py]
    Service --> Model[Model 层<br/>user.py]
    Model --> DB[(Database)]

    Service --> Validator[验证器<br/>validate_email]

    style Controller fill:#ff6b6b
    style Service fill:#4ecdc4
    style Model fill:#45b7d1
```

---

## 6. CQRS 架构

### 模式 6.1: 命令查询分离

**代码特征**：
```python
# commands/create_order.py (写操作)
class CreateOrderCommand:
    def execute(self, order_data):
        order = Order(**order_data)
        write_db.save(order)
        event_bus.emit("order_created", order)

# queries/get_order.py (读操作)
class GetOrderQuery:
    def execute(self, order_id):
        return read_db.query(Order).filter_by(id=order_id).first()
```

**Mermaid 模板**：
```mermaid
graph TB
    Client[客户端]

    Client -->|写操作| CommandHandler[Command Handler]
    CommandHandler --> WriteDB[(Write DB<br/>PostgreSQL)]
    CommandHandler -->|发布事件| EventBus[Event Bus]

    Client -->|读操作| QueryHandler[Query Handler]
    QueryHandler --> ReadDB[(Read DB<br/>Redis/Elasticsearch)]

    EventBus -->|同步| ReadDB

    style CommandHandler fill:#ff6b6b
    style QueryHandler fill:#4ecdc4
```

---

## 7. 混合架构识别

### 示例：微服务 + 事件驱动 + CQRS

**特征**：
- 多个独立服务（微服务）
- 服务间通过事件通信（事件驱动）
- 每个服务内部 CQRS（命令查询分离）

**Mermaid 模板**：
```mermaid
graph TB
    subgraph "Order Service"
        OrderCmd[Command Handler]
        OrderQuery[Query Handler]
        OrderWriteDB[(Write DB)]
        OrderReadDB[(Read DB)]
    end

    subgraph "Inventory Service"
        InvCmd[Command Handler]
        InvQuery[Query Handler]
        InvWriteDB[(Write DB)]
        InvReadDB[(Read DB)]
    end

    EventBus[Event Bus<br/>RabbitMQ]

    OrderCmd --> OrderWriteDB
    OrderCmd --> EventBus
    OrderQuery --> OrderReadDB

    EventBus -->|order_created| InvCmd
    InvCmd --> InvWriteDB
    InvQuery --> InvReadDB

    style EventBus fill:#ffd93d
```

---

## 8. 模式识别决策树

```
1. 检测 docker-compose.yml 或 k8s/ 目录
   → 是: 可能是微服务架构
   → 否: 继续

2. 检测 agents/ 目录 OR langgraph/crewai 导入
   → 是: Multi-Agent 架构
   → 否: 继续

3. 检测 emit/publish/subscribe
   → 是: 事件驱动架构
   → 否: 继续

4. 检测 extract/transform/load 函数
   → 是: 数据管道架构
   → 否: 继续

5. 检测 routes/ + services/ + models/
   → 是: 分层架构（MVC）
   → 否: 继续

6. 默认: 通用应用，分析主要调用链
```

---

## 9. 使用建议

1. **不要生搬硬套** - 并非所有项目都符合标准模式
2. **可以组合** - 一个项目可能同时使用多种模式
3. **关注核心** - 识别最重要的架构特征
4. **标注依据** - 在报告中注明识别依据（代码位置）

---

## 10. 扩展阅读

- **Martin Fowler's Architecture Patterns**: https://martinfowler.com/architecture/
- **Microservices Patterns**: https://microservices.io/patterns/
- **Event Driven Architecture**: https://martinfowler.com/articles/201701-event-driven.html