forked from geekan/MetaGPT
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathsoftware_company.py
61 lines (51 loc) · 1.96 KB
/
software_company.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2023/5/12 00:30
@Author : alexanderwu
@File : software_company.py
"""
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 SoftwareCompany(BaseModel):
"""
Software Company: Possesses a team, SOP (Standard Operating Procedures), and a platform for instant messaging,
dedicated to 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):
"""Start a project from publishing boss requirement."""
self.idea = idea
self.environment.publish_message(Message(role="BOSS", content=idea, cause_by=BossRequirement))
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