概述

本文档详细记录了如何在 MetaGPT 框架中实现一个具有动态 Action 创建能力的 Agent。通过本教程，你将深入理解 MetaGPT 的 React 机制（run → react → think → act）以及如何在运行时动态切换 Action 序列。

作业目标：

• 创建一个 Agent，初始化时拥有三个动作：Print1, Print2, Print3
• 顺序执行这三个动作
• 执行完毕后，动态生成新的动作：Print4, Print5, Print6
• 继续顺序执行新动作

学习重点：

• MetaGPT 的 React 循环机制
• 状态管理（state, todo, actions）
• 动态 Action 创建与替换
• Python 对象引用与多态

核心概念

1. MetaGPT 的 React 机制

┌─────────────────────────────────────┐
│            run(message)             │
│  - 接收消息并存入 memory            │
│  - 调用 react()                     │
└──────────────┬──────────────────────┘
               ↓
┌─────────────────────────────────────┐
│            react()                  │
│  - while True 循环                  │
│    - think() → 决定下一个动作       │
│    - 检查 todo 是否为 None          │
│    - act() → 执行当前动作           │
└──────────────┬──────────────────────┘
               ↓
┌──────────────┴──────────────────────┐
│                                     │
│  ┌─────────────┐  ┌──────────────┐ │
│  │   think()   │  │    act()     │ │
│  │ 分配下一个  │  │  执行当前    │ │
│  │   动作      │  │    动作      │ │
│  └─────────────┘  └──────────────┘ │
│                                     │
└─────────────────────────────────────┘

2. 关键属性

self.actions      # 动作列表 [Action1, Action2, ...]
self.states       # 状态列表 ['0. Action1', '1. Action2', ...]
self.rc.state     # 当前状态索引 (int)
self.rc.todo      # 当前要执行的动作 (Action 对象或 None)

3. 状态转换

# _set_state() 方法的作用
def _set_state(self, state: int):
    self.rc.state = state
    self.rc.todo = self.actions[state] if state >= 0 else None

状态值含义：

• state = -1 → 无任务，todo = None
• state = 0 → 执行第一个动作，todo = actions[0]
• state = 1 → 执行第二个动作，todo = actions[1]
• 以此类推...

完整实现

Step 1: 定义 Action 类

from metagpt.actions import Action
from metagpt.logs import logger

class PrintAction(Action):
    """简单的打印动作
    
    Args:
        name: 动作名称
        content: 要打印的内容
    """
    
    name: str = "PrintAction"
    content: str = ""
    
    async def run(self, *args, **kwargs) -> str:
        """执行打印操作"""
        logger.info(f"执行 {self.name}: {self.content}")
        return self.content

关键点：

• 继承自 Action 基类
• run() 方法必须是 async（因为可能调用 LLM）
• 返回执行结果（字符串）

Step 2: 定义 Agent 类框架

from metagpt.roles.role import Role, RoleReactMode
from metagpt.schema import Message

class SimpleAgent(Role):
    """简单的顺序执行 Agent
    
    Args:
        name: 角色名称
        profile: 角色描述
    """
    
    name: str = "SimpleAgent"
    profile: str = "Simple Sequential Agent"
    
    phase: int = 1  # 当前阶段（1 或 2）
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        
        # 初始化第一阶段的三个动作
        action1 = PrintAction(content="1")
        action2 = PrintAction(content="2")
        action3 = PrintAction(content="3")
        
        self._init_actions([action1, action2, action3])
        self._set_react_mode(react_mode=RoleReactMode.REACT.value)

关键点：

• phase 属性用于追踪当前阶段
• _init_actions() 将动作列表存入 self.actions
• _set_react_mode() 设置为 REACT 模式（自定义 react 逻辑）

Step 3: 实现 _think() 方法

async def _think(self) -> None:
    """决定下一个要执行的动作"""
    logger.info(f"{self._setting}: thinking about the next action to take")
    
    # 情况 1: 刚开始或刚切换阶段，从第 0 个动作开始
    if self.rc.todo is None:
        self._set_state(0)
        return
    
    # 情况 2: 还有下一个动作，移动到下一个
    if self.rc.state + 1 < len(self.states):
        self._set_state(self.rc.state + 1)
    
    # 情况 3: 所有动作执行完毕，设为 None（停止信号）
    else:
        self.rc.todo = None

逻辑流程：

┌─────────────────────────┐
│  todo is None?          │
│  (刚开始或刚切换阶段)   │
└──────┬──────────────────┘
       │ Yes
       ↓
┌─────────────────────────┐
│  _set_state(0)          │
│  state=0, todo=actions[0]│
└─────────────────────────┘
       │ No
       ↓
┌─────────────────────────┐
│  state+1 < len(states)? │
│  (还有下一个动作吗)     │
└──────┬──────────────────┘
       │ Yes
       ↓
┌─────────────────────────┐
│  _set_state(state+1)    │
│  移动到下一个动作       │
└─────────────────────────┘
       │ No
       ↓
┌─────────────────────────┐
│  todo = None            │
│  (所有动作完成)         │
└─────────────────────────┘

Step 4: 实现 _act() 方法

async def _act(self) -> Message:
    """执行当前动作"""
    todo = self.rc.todo
    logger.info(f"{self._setting}: executing {todo.name}")
    
    # 执行当前动作
    result = await todo.run()
    
    # 检测阶段切换条件
    if self.phase == 1 and self.rc.state == 2:
        logger.info("========== 第一阶段完成，准备进入第二阶段 ==========")
        self._switch_to_phase_2()
    
    return Message(content=result, role=self.name)

关键点：

• todo = self.rc.todo 获取当前动作对象
• await todo.run() 执行动作（多态调用）
• 执行完后检测是否需要切换阶段
• 返回 Message 对象包装结果

为什么 todo.run() 会调用 PrintAction.run()？

# 对象引用追踪：
action1 = PrintAction(content="1")  # 创建 PrintAction 实例
  ↓
self.actions = [action1, ...]  # 存入列表（存的是引用）
  ↓
self.rc.todo = self.actions[0]  # todo 指向 action1
  ↓
todo = self.rc.todo  # todo 指向同一个 PrintAction 对象
  ↓
todo.run()  # Python 根据对象类型调用 PrintAction.run()

Step 5: 实现阶段切换方法

def _switch_to_phase_2(self) -> None:
    """切换到第二阶段"""
    logger.info("创建第二阶段的动作: Print4, Print5, Print6")
    
    # 更新阶段标志
    self.phase = 2
    
    # 创建新的动作
    action4 = PrintAction(content="4")
    action5 = PrintAction(content="5")
    action6 = PrintAction(content="6")
    
    # 替换动作列表
    self._init_actions([action4, action5, action6])
    
    # 重置 todo，让下次 _think() 重新开始
    self.rc.todo = None
    
    logger.info("第二阶段动作已准备就绪")

关键点：

1. 为什么用 _init_actions() 而不是 append()？

   • _init_actions() 会替换整个 self.actions 列表
   • 这样第二阶段就只有 Print4, 5, 6，而不是 1-6 全部

2. 为什么要设置 self.rc.todo = None？

   让我们对比两种方案：

   ❌ 错误方案：使用 _set_state(0)

   def _switch_to_phase_2(self):
       self._init_actions([...])
       self._set_state(0)  # state=0, todo=actions[0]
       
   # 执行流程：
   # 1. 返回 _react() 循环
   # 2. 下次 _think() 被调用
   # 3. if self.rc.todo is None: → False（todo 已经是 actions[0]）
   # 4. if self.rc.state + 1 < len(self.states): → True (0+1 < 3)
   # 5. self._set_state(1) → 跳过了 Print4，直接执行 Print5！

   ✓ 正确方案：使用 self.rc.todo = None

   def _switch_to_phase_2(self):
       self._init_actions([...])
       self.rc.todo = None  # 重置为 None
       
   # 执行流程：
   # 1. 返回 _react() 循环
   # 2. 下次 _think() 被调用
   # 3. if self.rc.todo is None: → True
   # 4. self._set_state(0) → state=0, todo=actions[0] (Print4)
   # 5. 正确从 Print4 开始执行！

Step 6: 实现 _react() 方法

async def _react(self) -> Message:
    """循环执行 think 和 act"""
    msg = None
    
    while True:
        # 思考下一步
        await self._think()
        
        # 检查是否所有任务完成
        if self.rc.todo is None:
            break
        
        # 执行当前任务
        msg = await self._act()
    
    return msg

执行流程：

开始
  ↓
┌───────────────┐
│ msg = None    │
└───────┬───────┘
        ↓
    ┌───────────────┐
    │ while True:   │
    └───┬───────────┘
        ↓
    ┌───────────────┐
    │ _think()      │  ← 决定下一个动作
    └───┬───────────┘
        ↓
    ┌───────────────┐
    │ todo is None? │
    └───┬───────────┘
        │ Yes → break
        │ No
        ↓
    ┌───────────────┐
    │ msg = _act()  │  ← 执行动作
    └───┬───────────┘
        │
        └──→ 循环
        
返回 msg

Step 7: 测试代码

import asyncio

async def main():
    """测试函数"""
    logger.info("========== 开始测试 ==========")
    
    # 创建 Agent
    agent = SimpleAgent()
    
    # 运行 Agent
    result = await agent.run("开始执行")
    
    logger.info(f"========== 全部完成，最终结果: {result} ==========")

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

执行流程详解

完整执行时间线

时刻 T1: 初始化
├─ actions = [Print1, Print2, Print3]
├─ states = ['0. PrintAction', '1. PrintAction', '2. PrintAction']
├─ state = -1
├─ todo = None
└─ phase = 1

─────────────────────────────────────────

时刻 T2: run("开始执行")
└─ 调用 react()

─────────────────────────────────────────

时刻 T3: 第 1 次循环
├─ _think()
│  └─ todo is None → _set_state(0)
│     ├─ state = 0
│     └─ todo = actions[0] (Print1)
├─ _act()
│  └─ 执行 PrintAction: 1

─────────────────────────────────────────

时刻 T4: 第 2 次循环
├─ _think()
│  └─ state+1 < 3 → _set_state(1)
│     ├─ state = 1
│     └─ todo = actions[1] (Print2)
├─ _act()
│  └─ 执行 PrintAction: 2

─────────────────────────────────────────

时刻 T5: 第 3 次循环
├─ _think()
│  └─ state+1 < 3 → _set_state(2)
│     ├─ state = 2
│     └─ todo = actions[2] (Print3)
├─ _act()
│  ├─ 执行 PrintAction: 3
│  └─ 检测到 phase==1 and state==2
│     └─ _switch_to_phase_2()
│        ├─ phase = 2
│        ├─ actions = [Print4, Print5, Print6]
│        └─ todo = None

─────────────────────────────────────────

时刻 T6: 第 4 次循环
├─ _think()
│  └─ todo is None → _set_state(0)
│     ├─ state = 0
│     └─ todo = actions[0] (Print4)
├─ _act()
│  └─ 执行 PrintAction: 4

─────────────────────────────────────────

时刻 T7: 第 5 次循环
├─ _think()
│  └─ state+1 < 3 → _set_state(1)
│     ├─ state = 1
│     └─ todo = actions[1] (Print5)
├─ _act()
│  └─ 执行 PrintAction: 5

─────────────────────────────────────────

时刻 T8: 第 6 次循环
├─ _think()
│  └─ state+1 < 3 → _set_state(2)
│     ├─ state = 2
│     └─ todo = actions[2] (Print6)
├─ _act()
│  └─ 执行 PrintAction: 6

─────────────────────────────────────────

时刻 T9: 第 7 次循环
├─ _think()
│  └─ state+1 < 3 → False
│     └─ todo = None
└─ todo is None → break

─────────────────────────────────────────

返回结果: SimpleAgent: 6

关键技术点

1. Python 对象引用

在 Python 中，变量是"标签"而不是"盒子"：

# 创建对象
action1 = PrintAction(content="1")

# 多个变量可以指向同一个对象
self.actions[0] = action1    # 指向同一对象
self.rc.todo = action1        # 指向同一对象
todo = action1                # 指向同一对象

# 验证（id 相同说明是同一对象）
id(self.actions[0]) == id(self.rc.todo) == id(todo)  # True

2. 多态（Polymorphism）

# _act() 方法不需要知道 todo 的具体类型
async def _act(self):
    todo = self.rc.todo  # 可能是任何 Action 子类
    result = await todo.run()  # Python 会自动找到对应类的 run() 方法
    
# 当 todo 是 PrintAction 时 → 调用 PrintAction.run()
# 当 todo 是 WriteDirectory 时 → 调用 WriteDirectory.run()
# 这就是"面向接口编程"

3. async/await 使用规则

# ✓ 需要 async（函数内有 await）
async def _act(self):
    result = await todo.run()  # ← 有 await
    return result

# ✓ 不需要 async（函数内无 await）
def _switch_to_phase_2(self):
    self.phase = 2  # ← 没有 await
    self._init_actions([...])

记忆法则：

• 函数内有 await → 必须用 async def
• 调用 async def 函数 → 必须用 await

4. 状态重置的重要性

# 在动态创建 actions 后，必须重置 todo
self._init_actions([新动作...])
self.rc.todo = None  # ← 关键：让 _think() 重新评估

# 为什么不用 _set_state(0)？
# 因为会导致 _think() 判断错误，跳过第一个动作

完整代码

from metagpt.actions import Action
from metagpt.logs import logger
from metagpt.roles.role import Role, RoleReactMode
from metagpt.schema import Message
import asyncio


class PrintAction(Action):
    """简单的打印动作"""
    name: str = "PrintAction"
    content: str = ""
    
    async def run(self, *args, **kwargs) -> str:
        logger.info(f"执行 {self.name}: {self.content}")
        return self.content


class SimpleAgent(Role):
    """简单的顺序执行 Agent"""
    name: str = "SimpleAgent"
    profile: str = "Simple Sequential Agent"
    phase: int = 1
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        action1 = PrintAction(content="1")
        action2 = PrintAction(content="2")
        action3 = PrintAction(content="3")
        self._init_actions([action1, action2, action3])
        self._set_react_mode(react_mode=RoleReactMode.REACT.value)
    
    async def _think(self) -> None:
        """决定下一个要执行的动作"""
        logger.info(f"{self._setting}: thinking about the next action to take")
        
        if self.rc.todo is None:
            self._set_state(0)
            return
        
        if self.rc.state + 1 < len(self.states):
            self._set_state(self.rc.state + 1)
        else:
            self.rc.todo = None
    
    async def _act(self) -> Message:
        """执行当前动作"""
        todo = self.rc.todo
        logger.info(f"{self._setting}: executing {todo.name}")
        result = await todo.run()
        
        if self.phase == 1 and self.rc.state == 2:
            logger.info("========== 第一阶段完成，准备进入第二阶段 ==========")
            self._switch_to_phase_2()
        
        return Message(content=result, role=self.name)
    
    def _switch_to_phase_2(self) -> None:
        """切换到第二阶段"""
        logger.info("创建第二阶段的动作: Print4, Print5, Print6")
        self.phase = 2
        action4 = PrintAction(content="4")
        action5 = PrintAction(content="5")
        action6 = PrintAction(content="6")
        self._init_actions([action4, action5, action6])
        self.rc.todo = None
        logger.info("第二阶段动作已准备就绪")
    
    async def _react(self) -> Message:
        """循环执行 think 和 act"""
        msg = None
        while True:
            await self._think()
            if self.rc.todo is None:
                break
            msg = await self._act()
        return msg


async def main():
    logger.info("========== 开始测试 ==========")
    agent = SimpleAgent()
    result = await agent.run("开始执行")
    logger.info(f"========== 全部完成，最终结果: {result} ==========")


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

常见问题

Q1: 为什么要用 _react() 循环而不是递归？

答：

• 循环更高效（避免栈溢出）
• 更容易控制（可以随时 break）
• 符合 MetaGPT 的设计模式

Q2: 可以在 __init__ 中就初始化所有 6 个动作吗？

答：
可以，但失去了动态性：

# 静态方案（不推荐）
self._init_actions([Print1, Print2, Print3, Print4, Print5, Print6])

# 动态方案（推荐）
# 初始只有 Print1-3，运行时根据条件创建 Print4-6

动态方案的优势：

• 更灵活（可以根据第一阶段的结果决定第二阶段）
• 模拟真实场景（如 TutorialAssistant 根据大纲动态创建章节）

Q3: _set_state() 和直接赋值 self.rc.state = n 有什么区别？

答：

# ✓ 推荐：使用方法
self._set_state(0)  # 同时更新 state 和 todo

# ❌ 不推荐：直接赋值
self.rc.state = 0  # 只更新 state，todo 还是旧的

_set_state() 确保 state 和 todo 同步更新。

Q4: 如何添加第三阶段？

答：
在 _act() 中添加新的检测条件：

async def _act(self) -> Message:
    todo = self.rc.todo
    result = await todo.run()
    
    # 第一阶段 → 第二阶段
    if self.phase == 1 and self.rc.state == 2:
        self._switch_to_phase_2()
    
    # 第二阶段 → 第三阶段
    elif self.phase == 2 and self.rc.state == 2:
        self._switch_to_phase_3()
    
    return Message(content=result, role=self.name)

def _switch_to_phase_3(self) -> None:
    self.phase = 3
    action7 = PrintAction(content="7")
    action8 = PrintAction(content="8")
    action9 = PrintAction(content="9")
    self._init_actions([action7, action8, action9])
    self.rc.todo = None

进阶扩展

扩展 1: 使用 LLM 决定下一阶段

async def _act(self) -> Message:
    todo = self.rc.todo
    result = await todo.run()
    
    if self.phase == 1 and self.rc.state == 2:
        # 询问 LLM 下一步要做什么
        prompt = "第一阶段完成了，请决定第二阶段要执行哪些动作，返回 JSON 格式"
        response = await self._aask(prompt)
        # 解析 LLM 返回的 JSON
        next_actions = self._parse_llm_response(response)
        self._switch_to_dynamic_phase(next_actions)
    
    return Message(content=result, role=self.name)

扩展 2: 条件分支

async def _act(self) -> Message:
    todo = self.rc.todo
    result = await todo.run()
    
    if self.phase == 1 and self.rc.state == 2:
        # 根据第一阶段的结果选择分支
        if int(result) % 2 == 0:
            self._switch_to_phase_2A()  # 偶数分支
        else:
            self._switch_to_phase_2B()  # 奇数分支
    
    return Message(content=result, role=self.name)

扩展 3: 循环执行

def _switch_to_phase_2(self) -> None:
    self.phase = 2
    
    # 创建更多动作
    actions = []
    for i in range(4, 10):  # Print4 到 Print9
        actions.append(PrintAction(content=str(i)))
    
    self._init_actions(actions)
    self.rc.todo = None

总结

通过本教程，你已经掌握了：

1. MetaGPT 的 React 循环机制

   • run → react → think → act 的完整流程
   • 状态管理（state, todo, actions）

2. 动态 Action 创建

   • 在运行时根据条件创建新 actions
   • 使用 _init_actions() 替换动作列表
   • 通过 self.rc.todo = None 重置状态

3. Python 核心概念

   • 对象引用与多态
   • async/await 正确使用
   • 状态机设计模式

4. 实践技巧

   • 如何调试 Agent 执行流程
   • 如何避免常见错误
   • 如何扩展功能

核心要点：

• _think() 负责决定下一个动作（状态转换）
• _act() 负责执行动作并处理副作用（如阶段切换）
• _react() 负责循环驱动整个流程
• 动态创建后必须重置 todo 为 None

参考资源

• MetaGPT 官方文档
• MetaGPT GitHub
• Python asyncio 文档

坐在工位前加班的时候突然想起去年的一些旅行旧事。
去年的十二月底，搭乘全日空从东京经由札幌飞往北海道最北端的小城稚内。天气很不好，在新千岁航站楼的广播里一直在放送欠航预警。
延迟了半小时后最终还是冒着风险起飞了，落地稚内的时候已经是下午五点，外面飘着暴雪。
我从来没有见过这么厚的雪。即使是在北极，斯瓦尔巴群岛上的极夜里，让人意识到这是在北国之冬的更多也是来自于无言的冰川而非呼啸着的暴风雪。
我拖着行李箱在几十厘米深的积雪中艰难地行走，总算到了前几日提前预定的青旅，里面亮着温暖的灯。

“チェックインをお願いします”，我推开青旅厚重的铁门，希望能办理入住。
虽然看不见前台的脸，但是里面已经传来了回应声。“はい！少々お待ちください。”
我走过转角朝前台走去，出乎意料的是，看到的并非是一个黄种人的脸。
是美国人。
我愣了一下，前台的小哥约莫和我差不多年纪，棕黑色的脸，看起来像是墨西哥裔和白人的混血。难道刚刚是他用日文回应的我吗？
“すみません。チェックインしたいのですが？” 我又重复了一遍。
他似乎看出了我心中的疑惑，友好地笑了笑，继续用他那无比流利的日文回应我：“はい。ご予約のお名前をいただけますか？”

办完入住他切成英文和我对话。原来他确实是美国人，来自德克萨斯。大学毕业之后出于对日本的向往学了几年日语来到这里工作。他的本职工作是在稚内做一名英语老师，在其他的时间里，也来这间青旅帮忙做前台。
他的日文和英文都很流利，所以不论是外国还是日本客人都能很顺利地接待。
我赞叹于他的勇气，但还是有些许疑惑。稚内只不过是北海道一个被人遗忘的
小镇，如果不是因为这里有着日本最北端的界碑宗谷岬，大概也不会有太多游人来此。
“Well,” 他做了个鬼脸，“I was just assigned here — it was random.”
“But the people here are all very friendly, even though life isn’t as bustling as in Tokyo,” 他补充道， “I'm really happy here.”

他问起我是如何学日文的。
我摇了摇头，说我的日文远没有他那么好。这也就是我们现在正在用英文聊天的原因：再多说两分钟日文我就要露馅了。
他有些惊讶，说不会日文是如何在日本生存的。“我刚来的时候我啥都看不懂。” 他很腼腆地回忆着当时的不堪，像一个犯了错的牛仔。
这反应实在太过有趣，以至于让我不由得笑出了声。“可能是因为汉字文化圈的缘故吧。大部分东西我们都看得懂，只不过不会念。”我和他解释道，“而且还能笔谈。“
“Really？”他很吃惊地看着我，“That’s crazy bro. How does it work?”
我抓过放在桌子上的便签纸，“打个比方”，我在纸上写下 “東京” 两个字，“你知道为什么東京是東京吗？”
他很迷惑地看着我。“不知道。我的汉字水平其实不高，speaking 还可以，看东西依然有点困难。”
“‘東’是 East 的意思，而‘京’的意思就是 Captial，连在一起的意思就是‘东边的都城’。你看 Kyoto 的汉字里面也有个 ‘京’。”我和他解释道。
“中国还有个 Beijing 呢，写成汉字是 ‘北京’，也是同一个意思。”
“北方的都城？”他拿起笔在纸上画了个问号。
“是的，你已经掌握了造词法。”
“汉字太难了。我只当他们像画画。”他坦白。

不一会儿青旅里又来了个人。瘦瘦高高的，典型的理工男长相，戴着一个没有边框的长方形眼镜。实不相瞒，要不是没有蓄胡子，我还以为是希特勒。
是东德人，果然。但出乎我意料的是，他的英文水平很差。
他试图加入我们的对话。可由于说话磕磕巴巴的，我俩都没有太听懂他在说什么。
“I… don’t , don’t un…understand.” 他举起手机，”Why she gh…ghost me”
我和前台的德州牛仔花了将近五分钟才弄懂究竟是怎么一回事——原来这位德国佬在 ig 上约了个日本女孩，两人聊的很开心，于是东德哥希望约她出来吃饭。没想到这位日本女孩很礼貌地回了一个她需要回旭川打工之后就再也不回消息了。
“I think we.. we are good”，东德哥磕磕绊绊地说，“she is good.. too”
德州牛仔和我对了个眼神，然后哈哈大笑，伸出手准备拍了拍他的肩膀。
“Welcome to East Asia.”
“我也是在这住了四年才慢慢理解远东，”他说，“这边的姑娘们说话都弯弯绕绕的。从来不会说不，但是意思就是没戏。”
德国佬明显并没有太理解他的话，“But she said we could like have a dinner.. if possible”
“Yah if possible bro —— which means impossible.”
“他们不会在明面上拒绝任何事的。”这位德州牛仔说，“至少我感觉日本是这样。”
“欢迎来到远东，”我重复了一遍他的话，“我猜中日韩都差不多。”
德国佬显得有些泄气，摇了摇头，转身走出了青旅的门，一头钻进稚内的雪夜里。

我看了眼前台的牛仔，“其实远东整体上都很封闭。远没有表面上看起来那么 friendly。”
他难得地沉默。过了许久，他才接上了我的话。
“是的。虽然他们都很友好，但是我在这呆了四年，我才发现似乎永远成不了‘one of them’。”
“你还准备回德州吗？”我问道。
“会的。我只是每年在这里呆大半年，然后我还会回去。”他说，“可能再干一段时间我也会结束在日本的日子了。我已经在这里呆了四年。这里是个很适合旅游的地方。”
他顿了顿。随即又补充道，“也仅仅是个很适合旅游的地方。”
他准备下班了，离开的时候指着青旅墙上的地图和我告别。
“宗谷岬很值得去一趟。每天早上有巴士可以过去。”他又递给我一张小地图，“另外街区南边拐角那个拉面店很好吃，请一定要去试试。”
我和他挥手，说下次有机会来稚内还会来找他。虽然我们都知道很难有下一次了。
但那家拉面店的确很好吃。

数日后我在旭川的火车站再次见到那位东德人。
准确的说，是他远远地看到了我，跑过来和我打招呼。
“フレッド！” 他喊出我的名字。这口音实在诡异，以至于直到现在我也不知道他说的究竟是德文、英文还是日文。
“你怎么也在旭川？”我有点惊讶。“你在旭川找到那位姑娘了吗？”
他摇晃着脑袋，脸上露出几分遗憾又几分释然的表情。“我要继续往南边走了。十分钟后的车票。”
“Auf Wiedersehen!” 进站前，他用德文和我告别。
“Tschüss.” 我也用德文回他。

概述

本文记录了在部署和使用 SWE-smith（一个用于生成软件工程任务的工具）过程中遇到的各种技术问题及其解决方案。SWE-smith 是一个复杂的系统，涉及多个组件：bug生成、验证、收集、issue生成等。

遇到的问题与解决方案

1. Git推送权限问题

问题描述：
在执行 python -m swesmith.harness.gather 命令时，遇到以下错误：

subprocess.CalledProcessError: Command 'git push origin catchorg__Catch2.9b3f508a.func_pm_ctrl_invert_if__7p0kyikq' returned non-zero exit status 128.
ERROR: Permission to swesmith/catchorg__Catch2.9b3f508a.git denied to fredsun02.

根本原因：

• 原始代码配置将镜像仓库创建在 swesmith 组织下
• 当前用户的GitHub Token没有推送到该组织的权限
• 用户 fredsun02 不在 swesmith 组织中

解决方案：
修改 swesmith/constants.py 文件中的组织配置：

# 原来
ORG_NAME_GH = "swesmith"

# 修改为
ORG_NAME_GH = "fredsun02"

这样所有镜像仓库都会创建在用户的个人GitHub账户下，避免了组织权限问题。

2. 内存不足问题

问题描述：
在执行issue生成时，系统内存不足导致进程被杀死：

exit code 137  # 通常表示内存不足

根本原因：

• 系统只有1.9GB内存
• 代码尝试加载大型数据集（SWE-bench_Verified、SWE-bench/SWE-smith）
• 本地模型加载需要大量内存

解决方案：

2.1 使用API替代本地模型

创建简化的配置文件 configs/issue_gen/ig_api.yaml：

model: anthropic/claude-3-5-sonnet-20241022
system: |-
  You are a software engineer helping to create a realistic dataset of synthetic GitHub issues.
  # ... 系统提示词
demonstration: ""  # 不使用演示数据

2.2 优化代码逻辑

修改 swesmith/issue_gen/generate.py：

• 延迟加载 SWE-bench_Verified 数据集
• 只在需要时才加载大型数据集
• 对于本地数据集，跳过不必要的过滤

3. 包版本兼容性问题

问题描述：

ImportError: cannot import name 'ResponseTextConfig' from 'openai.types.responses.response'

根本原因：
litellm 和 openai 包版本不兼容

解决方案：
升级相关包到兼容版本：

pip install --upgrade openai litellm

4. 代码导入错误

问题描述：
在运行静态issue生成时遇到导入错误：

ImportError: cannot import name 'PM_TECHNIQUES_CLASSES' from 'swesmith.bug_gen.procedural.generate'

根本原因：
代码中引用了不存在的常量

解决方案：
修改 swesmith/issue_gen/get_static.py：

# 原来
from swesmith.bug_gen.procedural.generate import (
    PM_TECHNIQUES_CLASSES,
    PM_TECHNIQUES_FUNCS,
)

# 修改为
from swesmith.bug_gen.procedural import MAP_EXT_TO_MODIFIERS

5. 函数参数错误

问题描述：
在运行F2P方法时遇到函数调用错误：

TypeError: run_command_in_container() missing 1 required positional argument: 'rp'

解决方案：
修复 swesmith/issue_gen/get_from_tests.py 中的函数调用：

# 原来
test_output = run_command_in_container(instance, cmd)

# 修改为
test_output = run_command_in_container(instance, cmd, rp)

最终工作流程

经过修复后，完整的SWE-smith工作流程如下：

1. Bug生成 ✅

   python -m swesmith.bug_gen.procedural.generate --repo catchorg/Catch2 --n_instances 10

2. 验证 ✅

   python -m swesmith.harness.validate --dataset_path logs/bug_gen/catchorg__Catch2.9b3f508a_all_patches.json

3. 收集补丁 ✅

   python -m swesmith.harness.gather logs/run_validation/catchorg__Catch2.9b3f508a/ --debug_subprocess

4. Issue生成 ✅

   • API方法：python -m swesmith.issue_gen.generate -d logs/task_insts/catchorg__Catch2.9b3f508a.json -c configs/issue_gen/ig_api.yaml -w 1
   • 静态方法：python -m swesmith.issue_gen.get_static logs/task_insts/catchorg__Catch2.9b3f508a.json
   • F2P方法：python -m swesmith.issue_gen.get_from_tests logs/task_insts/catchorg__Catch2.9b3f508a.json

技术要点总结

1. 权限管理

• 在多人协作的项目中，确保GitHub Token有足够的权限
• 考虑使用个人账户而非组织账户来避免权限复杂性

2. 资源优化

• 对于内存受限的环境，优先使用API而非本地模型
• 实现延迟加载和按需加载来减少内存使用

3. 版本兼容性

• 定期更新依赖包以保持兼容性
• 在部署前测试所有依赖的版本组合

4. 代码维护

• 及时修复过时的导入和函数调用
• 保持文档与代码的同步

成本分析

使用API方法的成本：

• 处理9个实例
• 总成本：约$0.152
• 平均每个实例：约$0.017

结论

通过系统性的问题诊断和解决，我们成功部署了SWE-smith系统。主要挑战集中在权限管理、资源优化和代码兼容性方面。这些解决方案为类似项目的部署提供了有价值的参考。

最终，我们成功生成了：

• 10个bug分支并推送到GitHub
• 9个高质量的GitHub issue
• 完整的任务实例数据集

这个经验表明，在部署复杂的AI系统时，需要综合考虑技术、权限、资源和成本等多个方面。