mirror of
https://github.com/FoundationAgents/MetaGPT.git
synced 2026-05-01 20:03:28 +02:00
92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
@Time : 2023/5/12 00:30
|
|
@Author : alexanderwu
|
|
@File : software_company.py
|
|
"""
|
|
import os
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from metagpt.actions import BossRequirement
|
|
from metagpt.config import CONFIG
|
|
from metagpt.environment import Environment
|
|
from metagpt.logs import logger
|
|
from metagpt.roles import Role
|
|
from metagpt.schema import Message
|
|
from metagpt.utils.common import NoMoneyException
|
|
|
|
|
|
class Team(BaseModel):
|
|
"""
|
|
Team: Possesses one or more roles (agents), SOP (Standard Operating Procedures), and a platform for instant messaging,
|
|
dedicated to perform any multi-agent activity, such as collaboratively writing executable code.
|
|
"""
|
|
environment: Environment = Field(default_factory=Environment)
|
|
investment: float = Field(default=10.0)
|
|
idea: str = Field(default="")
|
|
|
|
class Config:
|
|
arbitrary_types_allowed = True
|
|
|
|
def hire(self, roles: list[Role]):
|
|
"""Hire roles to cooperate"""
|
|
self.environment.add_roles(roles)
|
|
|
|
def invest(self, investment: float):
|
|
"""Invest company. raise NoMoneyException when exceed max_budget."""
|
|
self.investment = investment
|
|
CONFIG.max_budget = investment
|
|
logger.info(f'Investment: ${investment}.')
|
|
|
|
def _check_balance(self):
|
|
if CONFIG.total_cost > CONFIG.max_budget:
|
|
raise NoMoneyException(CONFIG.total_cost, f'Insufficient funds: {CONFIG.max_budget}')
|
|
|
|
def start_project(self, idea, send_to: str = ""):
|
|
"""Start a project from publishing boss requirement."""
|
|
self.idea = idea
|
|
self.environment.publish_message(Message(role="Human", content=idea, cause_by=BossRequirement, send_to=send_to))
|
|
|
|
def _save(self):
|
|
logger.info(self.json())
|
|
|
|
async def run(self, n_round=3):
|
|
"""Run company until target round or no money"""
|
|
while n_round > 0:
|
|
# self._save()
|
|
n_round -= 1
|
|
logger.debug(f"{n_round=}")
|
|
self._check_balance()
|
|
await self.environment.run()
|
|
return self.environment.history
|
|
|
|
def save_legacy(self, project_path):
|
|
prd_path = os.path.join(project_path, 'docs/prd.md')
|
|
design_path = os.path.join(project_path, 'docs/system_design.md')
|
|
api_spec_and_tasks_path = os.path.join(project_path, 'docs/api_spec_and_tasks.md')
|
|
code_path = os.path.join(project_path, os.path.basename(project_path))
|
|
with open(prd_path, 'r', encoding='utf-8') as f:
|
|
legacy_prd = f.read()
|
|
with open(design_path, 'r', encoding='utf-8') as f:
|
|
legacy_design = f.read()
|
|
with open(api_spec_and_tasks_path, 'r', encoding='utf-8') as f:
|
|
legacy_project_management = f.read()
|
|
legacy_codes = []
|
|
for root, dirs, files in os.walk(code_path):
|
|
for file in files:
|
|
legacy_code = {}
|
|
if file.endswith('.py'):
|
|
with open(os.path.join(root, file), 'r', encoding='utf-8') as f:
|
|
legacy_code['filename'] = file
|
|
legacy_code['code'] = f.read()
|
|
legacy_codes.append(legacy_code)
|
|
legacy_dict = {
|
|
'legacy_prd': legacy_prd,
|
|
'legacy_design': legacy_design,
|
|
'legacy_project_management': legacy_project_management,
|
|
'legacy_code': legacy_codes
|
|
}
|
|
self.environment.set_legacy(legacy_dict)
|
|
|