def generate_repo(
    idea,
    investment=3.0,
    n_round=5,
    code_review=True,
    run_tests=False,
    implement=True,
    project_name="",
    inc=False,
    project_path="",
    reqa_file="",
    max_auto_summarize_code=0,
    recover_path=None,
):
    """核心启动逻辑"""
    # 1. 配置初始化
    config.update_via_cli(project_path, project_name, inc, reqa_file, max_auto_summarize_code)
    ctx = Context(config=config)
    
    # 2. 团队组建
    company = Team(context=ctx)
    company.hire([
        TeamLeader(),
        ProductManager(),
        Architect(),
        Engineer2(),
        DataAnalyst(),
    ])
    
    # 3. 投资设置与异步执行
    company.invest(investment)
    asyncio.run(company.run(n_round=n_round, idea=idea))

@serialize_decorator
async def run(self, n_round=3, idea="", send_to="", auto_archive=True):
    """运行团队直到目标轮数或资金耗尽"""
    if idea:
        self.run_project(idea=idea, send_to=send_to)

    while n_round > 0:
        if self.env.is_idle:  # 所有角色都空闲
            logger.debug("All roles are idle.")
            break
        n_round -= 1
        self._check_balance()  # 检查预算
        await self.env.run()   # 环境运行一轮
        logger.debug(f"max {n_round=} left.")
    
    self.env.archive(auto_archive)  # 归档项目
    return self.env.history

def hire(self, roles: list[Role]):
    """雇佣角色进行协作"""
    self.env.add_roles(roles)

def invest(self, investment: float):
    """投资公司，设置预算上限"""
    self.investment = investment
    self.cost_manager.max_budget = investment

async def run(self, k=1):
    """处理一次所有角色的运行"""
    for _ in range(k):
        futures = []
        for role in self.roles.values():
            if role.is_idle:
                continue
            future = role.run()
            futures.append(future)

        if futures:
            await asyncio.gather(*futures)  # 并发执行所有角色
        logger.debug(f"is idle: {self.is_idle}")

def publish_message(self, message: Message, peekable: bool = True) -> bool:
    """分发消息给接收者"""
    logger.debug(f"publish_message: {message.dump()}")
    found = False
    # 根据 RFC 113 的路由特性计划
    for role, addrs in self.member_addrs.items():
        if is_send_to(message, addrs):
            role.put_message(message)  # 放入角色私有消息缓冲区
            found = True
    if not found:
        logger.warning(f"Message no recipients: {message.dump()}")
    self.history.add(message)  # For debug
    return True

@role_raise_decorator
async def run(self, with_message=None) -> Message | None:
    """观察，思考并根据观察结果行动"""
    if with_message:
        msg = self._process_input_message(with_message)
        self.put_message(msg)
    
    if not await self._observe():
        # 如果没有新信息，暂停等待
        logger.debug(f"{self._setting}: no news. waiting.")
        return

    rsp = await self.react()  # 反应（思考+行动）

    # 重置下一个要执行的动作
    self.set_todo(None)
    # 发送响应消息到环境
    self.publish_message(rsp)
    return rsp

async def _observe(self) -> int:
    """从消息缓冲区准备新消息进行处理"""
    # 读取未处理的消息
    news = []
    if self.recovered and self.latest_observed_msg:
        news = self.rc.memory.find_news(observed=[self.latest_observed_msg], k=10)
    if not news:
        news = self.rc.msg_buffer.pop_all()
    
    # 存储到自己的记忆中防止重复处理
    old_messages = [] if not self.enable_memory else self.rc.memory.get()
    
    # 过滤感兴趣的消息
    self.rc.news = [
        n for n in news 
        if (n.cause_by in self.rc.watch or self.name in n.send_to) 
        and n not in old_messages
    ]
    
    if self.observe_all_msg_from_buffer:
        self.rc.memory.add_batch(news)
    else:
        self.rc.memory.add_batch(self.rc.news)
    
    self.latest_observed_msg = self.rc.news[-1] if self.rc.news else None
    return len(self.rc.news)

async def _think(self) -> bool:
    """考虑做什么并决定下一步行动"""
    if len(self.actions) == 1:
        # 如果只有一个动作，则只能执行这个
        self._set_state(0)
        return True

    if self.recovered and self.rc.state >= 0:
        self._set_state(self.rc.state)  # 从恢复状态运行的动作
        self.recovered = False
        return True

    if self.rc.react_mode == RoleReactMode.BY_ORDER:
        if self.rc.max_react_loop != len(self.actions):
            self.rc.max_react_loop = len(self.actions)
        self._set_state(self.rc.state + 1)
        return self.rc.state >= 0 and self.rc.state < len(self.actions)

    # 使用LLM动态选择动作
    prompt = self._get_prefix() + STATE_TEMPLATE.format(
        history=self.rc.history,
        states="\n".join(self.states),
        n_states=len(self.states) - 1,
        previous_state=self.rc.state,
    )

    next_state = await self.llm.aask(prompt)
    next_state = extract_state_value_from_output(next_state)
    
    if (not next_state.isdigit() and next_state != "-1") or int(next_state) not in range(-1, len(self.states)):
        logger.warning(f"Invalid answer of state, {next_state=}, will be set to -1")
        next_state = -1
    else:
        next_state = int(next_state)
        if next_state == -1:
            logger.info(f"End actions with {next_state=}")
    
    self._set_state(next_state)
    return True

class RoleReactMode(str, Enum):
    REACT = "react"           # 动态选择动作
    BY_ORDER = "by_order"     # 按顺序执行
    PLAN_AND_ACT = "plan_and_act"  # 先计划后执行

async def _act(self) -> Message:
    logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
    response = await self.rc.todo.run(self.rc.history)
    
    if isinstance(response, (ActionOutput, ActionNode)):
        msg = AIMessage(
            content=response.content,
            instruct_content=response.instruct_content,
            cause_by=self.rc.todo,
            sent_from=self,
        )
    elif isinstance(response, Message):
        msg = response
    else:
        msg = AIMessage(content=response or "", cause_by=self.rc.todo, sent_from=self)
    
    self.rc.memory.add(msg)
    return msg

ProductManager → Architect → Engineer → QaEngineer
    (需求)    →   (设计)   →  (编码)  →   (测试)

Engineer → CodeReview → Engineer
 (编码)  →   (评审)   →  (修改)

TeamLeader 统筹 → 各角色并行工作 → 结果汇总

async def run(self, *args, **kwargs):
    """运行动作"""
    if self.node:
        return await self._run_action_node(*args, **kwargs)
    raise NotImplementedError("The run method should be implemented in a subclass.")

async def _run_action_node(self, *args, **kwargs):
    """运行动作节点"""
    msgs = args[0]
    context = "## History Messages\n"
    context += "\n".join([f"{idx}: {i}" for idx, i in enumerate(reversed(msgs))])
    return await self.node.fill(req=context, llm=self.llm)

@exp_cache(serializer=ActionNodeSerializer())
async def fill(
    self,
    *,
    req,
    llm,
    schema="json",
    mode="auto",
    strgy="simple",
    images: Optional[Union[str, list[str]]] = None,
    timeout=USE_CONFIG_TIMEOUT,
    exclude=[],
    function_name: str = None,
):
    """填充节点内容"""
    self.set_llm(llm)
    self.set_context(req)
    if self.schema:
        schema = self.schema

    if mode == FillMode.CODE_FILL.value:
        result = await self.code_fill(context, function_name, timeout)
        self.instruct_content = self.create_class()(**result)
        return self

    elif mode == FillMode.XML_FILL.value:
        context = self.xml_compile(context=self.context)
        result = await self.xml_fill(context, images=images)
        self.instruct_content = self.create_class()(**result)
        return self

    elif mode == FillMode.SINGLE_FILL.value:
        result = await self.single_fill(context, images=images)
        self.instruct_content = self.create_class()(**result)
        return self

    if strgy == "simple":
        return await self.simple_fill(schema=schema, mode=mode, images=images, timeout=timeout, exclude=exclude)
    elif strgy == "complex":
        # 复杂策略：逐个处理子节点
        tmp = {}
        for _, i in self.children.items():
            if exclude and i.key in exclude:
                continue
            child = await i.simple_fill(schema=schema, mode=mode, images=images, timeout=timeout, exclude=exclude)
            tmp.update(child.instruct_content.model_dump())
        cls = self._create_children_class()
        self.instruct_content = cls(**tmp)
        return self

def compile(self, context, schema="json", mode="children", template=SIMPLE_TEMPLATE, exclude=[]) -> str:
    """
    编译提示模板
    mode: all/root/children
    schema: raw/json/markdown
    """
    if schema == "raw":
        return f"{context}\n\n## Actions\n{LANGUAGE_CONSTRAINT}\n{self.instruction}"

    instruction = self.compile_instruction(schema="markdown", mode=mode, exclude=exclude)
    example = self.compile_example(schema=schema, tag=TAG, mode=mode, exclude=exclude)
    constraints = [LANGUAGE_CONSTRAINT, FORMAT_CONSTRAINT]
    constraint = "\n".join(constraints)

    prompt = template.format(
        context=context,
        example=example,
        instruction=instruction,
        constraint=constraint,
    )
    return prompt

def add(self, message: Message):
    """添加新消息到存储，同时更新索引"""
    if self.ignore_id:
        message.id = IGNORED_MESSAGE_ID
    if message in self.storage:
        return
    self.storage.append(message)
    if message.cause_by:
        self.index[message.cause_by].append(message)

def get_by_actions(self, actions: Set) -> list[Message]:
    """返回由指定动作触发的所有消息"""
    rsp = []
    indices = any_to_str_set(actions)
    for action in indices:
        if action not in self.index:
            continue
        rsp += self.index[action]
    return rsp

def find_news(self, observed: list[Message], k=0) -> list[Message]:
    """从最近的 k 条记忆中找到新消息（之前未见过的消息）"""
    already_observed = self.get(k)
    news = []
    for i in observed:
        if i in already_observed:
            continue
        news.append(i)
    return news

async def aask(
    self,
    msg: Union[str, list[dict[str, str]]],
    system_msgs: Optional[list[str]] = None,
    format_msgs: Optional[list[dict[str, str]]] = None,
    images: Optional[Union[str, list[str]]] = None,
    timeout=USE_CONFIG_TIMEOUT,
    stream=None,
) -> str:
    if system_msgs:
        message = self._system_msgs(system_msgs)
    else:
        message = [self._default_system_msg()]
    if not self.use_system_prompt:
        message = []
    if format_msgs:
        message.extend(format_msgs)
    if isinstance(msg, str):
        message.append(self._user_msg(msg, images=images))
    else:
        message.extend(msg)
    if stream is None:
        stream = self.config.stream

    # 图像数据替换为占位符以避免长输出
    masked_message = [self.mask_base64_data(m) for m in message]
    logger.debug(masked_message)

    compressed_message = self.compress_messages(message, compress_type=self.config.compress_type)
    rsp = await self.acompletion_text(compressed_message, stream=stream, timeout=self.get_timeout(timeout))
    return rsp

def compress_messages(
    self,
    messages: list[dict],
    compress_type: CompressType = CompressType.NO_COMPRESS,
    max_token: int = 128000,
    threshold: float = 0.8,
) -> list[dict]:
    """压缩消息以适应token限制"""
    if compress_type == CompressType.NO_COMPRESS:
        return messages

    max_token = TOKEN_MAX.get(self.model, max_token)
    keep_token = int(max_token * threshold)
    compressed = []

    # Always keep system messages
    system_msg_val = self._system_msg("")["role"]
    system_msgs = []
    for i, msg in enumerate(messages):
        if msg["role"] == system_msg_val:
            system_msgs.append(msg)
        else:
            user_assistant_msgs = messages[i:]
            break
    
    compressed.extend(system_msgs)
    current_token_count = self.count_tokens(system_msgs)

    if compress_type in [CompressType.POST_CUT_BY_TOKEN, CompressType.POST_CUT_BY_MSG]:
        # 保留最新的消息
        for i, msg in enumerate(reversed(user_assistant_msgs)):
            token_count = self.count_tokens([msg])
            if current_token_count + token_count <= keep_token:
                compressed.insert(len(system_msgs), msg)
                current_token_count += token_count
            else:
                if compress_type == CompressType.POST_CUT_BY_TOKEN:
                    # 截断消息以适应剩余token数量
                    truncated_content = msg["content"][-(keep_token - current_token_count) :]
                    compressed.insert(len(system_msgs), {"role": msg["role"], "content": truncated_content})
                break

    return compressed

class MGXEnv(Environment, SerializationMixin):
    def publish_message(self, message: Message, user_defined_recipient: str = "", publicer: str = ""):
        """让团队领导接管消息发布"""
        message = self.attach_images(message)  # 多模态消息支持
        
        tl = self.get_role(TEAMLEADER_NAME)  # TeamLeader 名为 Mike
        
        if user_defined_recipient:
            # 用户直接与某个角色对话
            self.direct_chat_roles.add(role_name)
            self._publish_message(message)
        elif publicer == tl.profile:
            # 团队领导处理的消息可以发布
            self._publish_message(message)
        else:
            # 每个常规消息都通过团队领导
            message.send_to.add(tl.name)
            self._publish_message(message)

class AndroidEnv(ExtEnv):
    @mark_as_readable
    def get_screenshot(self):
        """获取屏幕截图"""
        
    @mark_as_writeable  
    def tap(self, x: int, y: int):
        """点击屏幕坐标"""

def _watch(self, actions: Iterable[Type[Action]]):
    """观察感兴趣的动作"""
    self.rc.watch = {any_to_str(t) for t in actions}

def is_watch(self, caused_by: str):
    return caused_by in self.rc.watch

def create_llm_instance(config: LLMConfig) -> BaseLLM:
    """根据配置创建 LLM 实例"""
    if config.api_type == LLMType.OPENAI:
        return OpenAILLM(config)
    elif config.api_type == LLMType.ANTHROPIC:
        return AnthropicLLM(config)
    # ...

class Team(BaseModel):
    def serialize(self, stg_path: Path = None):
        """序列化团队状态"""
        serialized_data = self.model_dump()
        serialized_data["context"] = self.env.context.serialize()
        write_json_file(team_info_path, serialized_data)
    
    @classmethod
    def deserialize(cls, stg_path: Path, context: Context = None) -> "Team":
        """从序列化数据恢复团队"""
        team_info = read_json_file(team_info_path)
        ctx = context or Context()
        ctx.deserialize(team_info.pop("context", None))
        return Team(**team_info, context=ctx)

async def run(self, k=1):
    """并发处理所有角色的运行"""
    futures = []
    for role in self.roles.values():
        if not role.is_idle:
            futures.append(role.run())
    
    if futures:
        await asyncio.gather(*futures)

class CostManager:
    def update_cost(self, prompt_tokens: int, completion_tokens: int, model: str):
        """更新每次请求的 token 成本"""
        
def _check_balance(self):
    if self.cost_manager.total_cost >= self.cost_manager.max_budget:
        raise NoMoneyException(self.cost_manager.total_cost)

@retry(
    wait=wait_random_exponential(min=1, max=20),
    stop=stop_after_attempt(6),
    after=general_after_log(logger),
)
async def _aask_v1(self, prompt: str, ...):
    """带重试的 LLM 调用"""

# Team.run / Environment.run / Role.run（节选）
@serialize_decorator
async def run(self, n_round=3, idea="", send_to="", auto_archive=True):
    if idea:
        self.run_project(idea=idea, send_to=send_to)
    while n_round > 0:
        if self.env.is_idle:
            break
        n_round -= 1
        self._check_balance()
        await self.env.run()
    self.env.archive(auto_archive)
    return self.env.history

async def run(self, k=1):
    for _ in range(k):
        futures = []
        for role in self.roles.values():
            if role.is_idle:
                continue
            futures.append(role.run())
        if futures:
            await asyncio.gather(*futures)

@role_raise_decorator
async def run(self, with_message=None) -> Message | None:
    if with_message:
        msg = Message(content=with_message) if isinstance(with_message, str) else (
            with_message if isinstance(with_message, Message) else Message(content="\n".join(with_message))
        )
        if not msg.cause_by:
            msg.cause_by = UserRequirement
        self.put_message(msg)
    if not await self._observe():
        return
    rsp = await self.react()
    self.set_todo(None)
    self.publish_message(rsp)
    return rsp

# Action.run / ActionNode.fill（节选）
async def _run_action_node(self, *args, **kwargs):
    msgs = args[0]
    context = "## History Messages\n" + "\n".join([f"{idx}: {i}" for idx, i in enumerate(reversed(msgs))])
    return await self.node.fill(req=context, llm=self.llm)

@exp_cache(serializer=ActionNodeSerializer())
async def fill(self, *, req, llm, schema="json", mode="auto", strgy="simple", images=None,
               timeout=USE_CONFIG_TIMEOUT, exclude=[], function_name=None):
    self.set_llm(llm)
    self.set_context(req)
    if self.schema:
        schema = self.schema
    if strgy == "simple":
        return await self.simple_fill(schema=schema, mode=mode, images=images, timeout=timeout, exclude=exclude)
    # 复杂模式：聚合子节点
    tmp = {}
    for _, child in self.children.items():
        c = await child.simple_fill(schema=schema, mode=mode, images=images, timeout=timeout, exclude=exclude)
        tmp.update(c.instruct_content.model_dump())
    cls = self._create_children_class()
    self.instruct_content = cls(**tmp)
    return self

generate_repo()
  └─ Team.run()
      ├─ Team._check_balance()
      ├─ Environment.run()
      │   └─ Role.run()
      │       ├─ Role._observe() → Memory/MessageQueue
      │       ├─ Role.react() → Role._react()/Role._plan_and_act()
      │       │   ├─ Role._think() → LLM.aask() [可选]
      │       │   └─ Role._act() → Action.run() → ActionNode.fill() → LLM.aask()
      │       └─ Environment.publish_message()
      └─ Environment.archive()

sequenceDiagram
    participant SC as SoftwareCompany
    participant T as Team
    participant E as Environment
    participant R as Role
    participant A as Action
    participant N as ActionNode
    participant L as LLM

    SC->>T: run(n_round, idea)
    T->>E: publish_message(Message(idea))
    loop n_round
        T->>E: run()
        par 角色并发
            E->>R: run()
            R->>R: _observe/_think/_act
            R->>A: run(history)
            A->>N: fill(req,llm)
            N->>L: aask(prompt)
            L-->>N: text
            N-->>A: instruct_content
            A-->>R: AIMessage
            R->>E: publish_message()
        end
    end