Merge branch 'main' into feat_repair_llmoutput

README.md

@@ -63,6 +63,7 @@ # If executing, ensure that NPM is installed on your system. Then install mermai
 detail installation please refer to [cli_install](https://docs.deepwisdom.ai/guide/get_started/installation.html#install-stable-version)
 
 ### Docker installation
+> Note: In the Windows, you need to replace "/opt/metagpt" with a directory that Docker has permission to create, such as "D:\Users\x\metagpt"
 
 ```bash
 # Step 1: Download metagpt official image and prepare config.yaml
@@ -123,7 +124,7 @@ ### Contact Information
 
 ## Citation
 
-For now, cite the [Arxiv paper](https://arxiv.org/abs/2308.00352):
+For now, cite the [arXiv paper](https://arxiv.org/abs/2308.00352):
 
 ```bibtex
 @misc{hong2023metagpt,

docs/README_CN.md

@@ -60,6 +60,7 @@ # 如果执行，确保您的系统上安装了 NPM。并使用npm安装mermaid-
 详细的安装请安装 [cli_install](https://docs.deepwisdom.ai/guide/get_started/installation.html#install-stable-version)
 
 ### Docker安装
+> 注意：在Windows中，你需要将 "/opt/metagpt" 替换为Docker具有创建权限的目录，比如"D:\Users\x\metagpt"
 
 ```bash
 # 步骤1: 下载metagpt官方镜像并准备好config.yaml
@@ -120,7 +121,7 @@ ### 联系信息
 
 ## 引用
 
-引用 [Arxiv paper](https://arxiv.org/abs/2308.00352):
+引用 [arXiv paper](https://arxiv.org/abs/2308.00352):
 
 ```bibtex
 @misc{hong2023metagpt,

docs/README_JA.md

@@ -163,6 +163,7 @@ # NPM がシステムにインストールされていることを確認して
     注: この方法は pdf エクスポートに対応していません。
 
 ### Docker によるインストール
+> Windowsでは、"/opt/metagpt"をDockerが作成する権限を持つディレクトリに置き換える必要があります。例えば、"D:\Users\x\metagpt"などです。
 
 ```bash
 # ステップ 1: metagpt 公式イメージをダウンロードし、config.yaml を準備する
@@ -299,7 +300,7 @@ ## クイックスタート
 
 ## 引用
 
-現時点では、[Arxiv 論文](https://arxiv.org/abs/2308.00352)を引用してください:
+現時点では、[arXiv 論文](https://arxiv.org/abs/2308.00352)を引用してください:
 
 ```bibtex
 @misc{hong2023metagpt,

examples/agent_creator.py

@@ -49,6 +49,8 @@ class CreateAgent(Action):
         pattern = r'```python(.*)```'
         match = re.search(pattern, rsp, re.DOTALL)
         code_text = match.group(1) if match else ""
+        if not WORKSPACE_ROOT.exists():
+            WORKSPACE_ROOT.mkdir(parents=True)
         with open(WORKSPACE_ROOT / "agent_created_agent.py", "w") as f:
             f.write(code_text)
         return code_text

examples/search_kb.py

@@ -5,22 +5,40 @@
 """
 import asyncio
 
+from metagpt.actions import Action
 from metagpt.const import DATA_PATH
 from metagpt.document_store import FaissStore
 from metagpt.logs import logger
 from metagpt.roles import Sales
+from metagpt.schema import Message
+
+""" example.json, e.g.
+[
+    {
+        "source": "Which facial cleanser is good for oily skin?",
+        "output": "ABC cleanser is preferred by many with oily skin."
+    },
+    {
+        "source": "Is L'Oreal good to use?",
+        "output": "L'Oreal is a popular brand with many positive reviews."
+    }
+]
+"""
 
 
 async def search():
-    store = FaissStore(DATA_PATH / 'example.json')
+    store = FaissStore(DATA_PATH / "example.json")
     role = Sales(profile="Sales", store=store)
-
-    queries = ["Which facial cleanser is good for oily skin?", "Is L'Oreal good to use?"]
+    role._watch({Action})
+    queries = [
+        Message("Which facial cleanser is good for oily skin?", cause_by=Action),
+        Message("Is L'Oreal good to use?", cause_by=Action),
+    ]
     for query in queries:
         logger.info(f"User: {query}")
         result = await role.run(query)
         logger.info(result)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     asyncio.run(search())

metagpt/actions/action.py

@@ -9,7 +9,7 @@
 from abc import ABC
 from typing import Optional
 
-from tenacity import retry, stop_after_attempt, wait_fixed, after_log, _utils
+from tenacity import retry, stop_after_attempt, wait_random_exponential
 
 from metagpt.actions.action_output import ActionOutput
 from metagpt.llm import LLM
@@ -51,8 +51,8 @@ class Action(ABC):
         return await self.llm.aask(prompt, system_msgs)
 
     @retry(
-        stop=stop_after_attempt(3),
-        wait=wait_fixed(1),
+        wait=wait_random_exponential(min=1, max=60),
+        stop=stop_after_attempt(6),
         after=general_after_log(logger),
     )
     async def _aask_v1(

metagpt/actions/search_and_summarize.py

@@ -140,4 +140,3 @@ class SearchAndSummarize(Action):
         logger.debug(prompt)
         logger.debug(result)
         return result
-    

metagpt/actions/write_code.py

@@ -11,7 +11,7 @@ from metagpt.const import WORKSPACE_ROOT
 from metagpt.logs import logger
 from metagpt.schema import Message
 from metagpt.utils.common import CodeParser
-from tenacity import retry, stop_after_attempt, wait_fixed
+from tenacity import retry, stop_after_attempt, wait_random_exponential
 
 PROMPT_TEMPLATE = """
 NOTICE
@@ -66,7 +66,7 @@ class WriteCode(Action):
         code_path.write_text(code)
         logger.info(f"Saving Code to {code_path}")
 
-    @retry(stop=stop_after_attempt(2), wait=wait_fixed(1))
+    @retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
     async def write_code(self, prompt):
         code_rsp = await self._aask(prompt)
         code = CodeParser.parse_code(block="", text=code_rsp)

metagpt/actions/write_code_review.py

@@ -10,7 +10,7 @@ from metagpt.actions.action import Action
 from metagpt.logs import logger
 from metagpt.schema import Message
 from metagpt.utils.common import CodeParser
-from tenacity import retry, stop_after_attempt, wait_fixed
+from tenacity import retry, stop_after_attempt, wait_random_exponential
 
 PROMPT_TEMPLATE = """
 NOTICE
@@ -65,7 +65,7 @@ class WriteCodeReview(Action):
     def __init__(self, name="WriteCodeReview", context: list[Message] = None, llm=None):
         super().__init__(name, context, llm)
 
-    @retry(stop=stop_after_attempt(2), wait=wait_fixed(1))
+    @retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
     async def write_code(self, prompt):
         code_rsp = await self._aask(prompt)
         code = CodeParser.parse_code(block="", text=code_rsp)

metagpt/document_store/faiss_store.py

@@ -5,6 +5,7 @@
 @Author  : alexanderwu
 @File    : faiss_store.py
 """
+import asyncio
 import pickle
 from pathlib import Path
 from typing import Optional
@@ -20,7 +21,7 @@ from metagpt.logs import logger
 
 
 class FaissStore(LocalStore):
-    def __init__(self, raw_data: Path, cache_dir=None, meta_col='source', content_col='output'):
+    def __init__(self, raw_data: Path, cache_dir=None, meta_col="source", content_col="output"):
         self.meta_col = meta_col
         self.content_col = content_col
         super().__init__(raw_data, cache_dir)
@@ -50,7 +51,7 @@ class FaissStore(LocalStore):
             pickle.dump(store, f)
         store.index = index
 
-    def search(self, query, expand_cols=False, sep='\n', *args, k=5, **kwargs):
+    def search(self, query, expand_cols=False, sep="\n", *args, k=5, **kwargs):
         rsp = self.store.similarity_search(query, k=k, **kwargs)
         logger.debug(rsp)
         if expand_cols:
@@ -58,6 +59,9 @@ class FaissStore(LocalStore):
         else:
             return str(sep.join([f"{x.page_content}" for x in rsp]))
 
+    async def asearch(self, *args, **kwargs):
+        return await asyncio.to_thread(self.search, *args, **kwargs)
+
     def write(self):
         """Initialize the index and library based on the Document (JSON / XLSX, etc.) file provided by the user."""
         if not self.raw_data.exists():
@@ -78,8 +82,8 @@ class FaissStore(LocalStore):
         raise NotImplementedError
 
 
-if __name__ == '__main__':
-    faiss_store = FaissStore(DATA_PATH / 'qcs/qcs_4w.json')
-    logger.info(faiss_store.search('Oily Skin Facial Cleanser'))
-    faiss_store.add([f'Oily Skin Facial Cleanser-{i}' for i in range(3)])
-    logger.info(faiss_store.search('Oily Skin Facial Cleanser'))
+if __name__ == "__main__":
+    faiss_store = FaissStore(DATA_PATH / "qcs/qcs_4w.json")
+    logger.info(faiss_store.search("Oily Skin Facial Cleanser"))
+    faiss_store.add([f"Oily Skin Facial Cleanser-{i}" for i in range(3)])
+    logger.info(faiss_store.search("Oily Skin Facial Cleanser"))

metagpt/provider/openai_api.py

@@ -15,7 +15,7 @@ from tenacity import (
     retry,
     retry_if_exception_type,
     stop_after_attempt,
-    wait_fixed,
+    wait_random_exponential,
 )
 
 from metagpt.config import CONFIG
@@ -115,7 +115,7 @@ class CostManager(metaclass=Singleton):
     def get_total_cost(self):
         """
         Get the total cost of API calls.
-
+    
         Returns:
         float: The total cost of API calls.
         """
@@ -229,8 +229,8 @@ class OpenAIGPTAPI(BaseGPTAPI, RateLimiter):
         return await self._achat_completion(messages)
 
     @retry(
-        stop=stop_after_attempt(3),
-        wait=wait_fixed(1),
+        wait=wait_random_exponential(min=1, max=60),
+        stop=stop_after_attempt(6),
         after=after_log(logger, logger.level("WARNING").name),
         retry=retry_if_exception_type(APIConnectionError),
         retry_error_callback=log_and_reraise,

metagpt/roles/__init__.py

@@ -12,7 +12,7 @@ from metagpt.roles.project_manager import ProjectManager
 from metagpt.roles.product_manager import ProductManager
 from metagpt.roles.engineer import Engineer
 from metagpt.roles.qa_engineer import QaEngineer
-from metagpt.roles.seacher import Searcher
+from metagpt.roles.searcher import Searcher
 from metagpt.roles.sales import Sales
 from metagpt.roles.customer_service import CustomerService
 

metagpt/roles/sales.py

@@ -12,24 +12,23 @@ from metagpt.tools import SearchEngineType
 
 class Sales(Role):
     def __init__(
-            self,
-            name="Xiaomei",
-            profile="Retail sales guide",
-            desc="I am a sales guide in retail. My name is Xiaomei. I will answer some customer questions next, and I "
-                 "will answer questions only based on the information in the knowledge base."
-                 "If I feel that you can't get the answer from the reference material, then I will directly reply that"
-                 " I don't know, and I won't tell you that this is from the knowledge base,"
-                 "but pretend to be what I know. Note that each of my replies will be replied in the tone of a "
-                 "professional guide",
-            store=None
+        self,
+        name="Xiaomei",
+        profile="Retail sales guide",
+        desc="I am a sales guide in retail. My name is Xiaomei. I will answer some customer questions next, and I "
+        "will answer questions only based on the information in the knowledge base."
+        "If I feel that you can't get the answer from the reference material, then I will directly reply that"
+        " I don't know, and I won't tell you that this is from the knowledge base,"
+        "but pretend to be what I know. Note that each of my replies will be replied in the tone of a "
+        "professional guide",
+        store=None,
     ):
         super().__init__(name, profile, desc=desc)
         self._set_store(store)
 
     def _set_store(self, store):
         if store:
-            action = SearchAndSummarize("", engine=SearchEngineType.CUSTOM_ENGINE, search_func=store.search)
+            action = SearchAndSummarize("", engine=SearchEngineType.CUSTOM_ENGINE, search_func=store.asearch)
         else:
             action = SearchAndSummarize()
         self._init_actions([action])
-        

metagpt/roles/searcher.py

@@ -3,7 +3,7 @@
 """
 @Time    : 2023/5/23 17:25
 @Author  : alexanderwu
-@File    : seacher.py
+@File    : searcher.py
 """
 from metagpt.actions import ActionOutput, SearchAndSummarize
 from metagpt.logs import logger

metagpt/subscription.py

@@ -0,0 +1,101 @@
+import asyncio
+from typing import AsyncGenerator, Awaitable, Callable
+
+from pydantic import BaseModel, Field
+
+from metagpt.logs import logger
+from metagpt.roles import Role
+from metagpt.schema import Message
+
+
+class SubscriptionRunner(BaseModel):
+    """A simple wrapper to manage subscription tasks for different roles using asyncio.
+
+    Example:
+        >>> import asyncio
+        >>> from metagpt.subscription import SubscriptionRunner
+        >>> from metagpt.roles import Searcher
+        >>> from metagpt.schema import Message
+
+        >>> async def trigger():
+        ...     while True:
+        ...         yield Message("the latest news about OpenAI")
+        ...         await asyncio.sleep(3600 * 24)
+
+        >>> async def callback(msg: Message):
+        ...     print(msg.content)
+
+        >>> async def main():
+        ...     pb = SubscriptionRunner()
+        ...     await pb.subscribe(Searcher(), trigger(), callback)
+        ...     await pb.run()
+
+        >>> asyncio.run(main())
+    """
+
+    tasks: dict[Role, asyncio.Task] = Field(default_factory=dict)
+
+    class Config:
+        arbitrary_types_allowed = True
+
+    async def subscribe(
+        self,
+        role: Role,
+        trigger: AsyncGenerator[Message, None],
+        callback: Callable[
+            [
+                Message,
+            ],
+            Awaitable[None],
+        ],
+    ):
+        """Subscribes a role to a trigger and sets up a callback to be called with the role's response.
+
+        Args:
+            role: The role to subscribe.
+            trigger: An asynchronous generator that yields Messages to be processed by the role.
+            callback: An asynchronous function to be called with the response from the role.
+        """
+        loop = asyncio.get_running_loop()
+
+        async def _start_role():
+            async for msg in trigger:
+                resp = await role.run(msg)
+                await callback(resp)
+
+        self.tasks[role] = loop.create_task(_start_role(), name=f"Subscription-{role}")
+
+    async def unsubscribe(self, role: Role):
+        """Unsubscribes a role from its trigger and cancels the associated task.
+
+        Args:
+            role: The role to unsubscribe.
+        """
+        task = self.tasks.pop(role)
+        task.cancel()
+
+    async def run(self, raise_exception: bool = True):
+        """Runs all subscribed tasks and handles their completion or exception.
+
+        Args:
+            raise_exception: _description_. Defaults to True.
+
+        Raises:
+            task.exception: _description_
+        """
+        while True:
+            for role, task in self.tasks.items():
+                if task.done():
+                    if task.exception():
+                        if raise_exception:
+                            raise task.exception()
+                        logger.opt(exception=task.exception()).error(f"Task {task.get_name()} run error")
+                    else:
+                        logger.warning(
+                            f"Task {task.get_name()} has completed. "
+                            "If this is unexpected behavior, please check the trigger function."
+                        )
+                    self.tasks.pop(role)
+                    break
+            else:
+                await asyncio.sleep(1)

metagpt/utils/token_counter.py

@@ -16,11 +16,13 @@ TOKEN_COSTS = {
     "gpt-3.5-turbo-0613": {"prompt": 0.0015, "completion": 0.002},
     "gpt-3.5-turbo-16k": {"prompt": 0.003, "completion": 0.004},
     "gpt-3.5-turbo-16k-0613": {"prompt": 0.003, "completion": 0.004},
+    "gpt-3.5-turbo-1106": {"prompt": 0.001, "completion": 0.002},
     "gpt-4-0314": {"prompt": 0.03, "completion": 0.06},
     "gpt-4": {"prompt": 0.03, "completion": 0.06},
     "gpt-4-32k": {"prompt": 0.06, "completion": 0.12},
     "gpt-4-32k-0314": {"prompt": 0.06, "completion": 0.12},
     "gpt-4-0613": {"prompt": 0.06, "completion": 0.12},
+    "gpt-4-1106-preview": {"prompt": 0.01, "completion": 0.03},
     "text-embedding-ada-002": {"prompt": 0.0004, "completion": 0.0},
     "chatglm_turbo": {"prompt": 0.0, "completion": 0.00069}  # 32k version, prompt + completion tokens=0.005￥/k-tokens
 }
@@ -32,11 +34,13 @@ TOKEN_MAX = {
     "gpt-3.5-turbo-0613": 4096,
     "gpt-3.5-turbo-16k": 16384,
     "gpt-3.5-turbo-16k-0613": 16384,
+    "gpt-3.5-turbo-1106": 16384,
     "gpt-4-0314": 8192,
     "gpt-4": 8192,
     "gpt-4-32k": 32768,
     "gpt-4-32k-0314": 32768,
     "gpt-4-0613": 8192,
+    "gpt-4-1106-preview": 128000,
     "text-embedding-ada-002": 8192,
     "chatglm_turbo": 32768
 }
@@ -52,10 +56,12 @@ def count_message_tokens(messages, model="gpt-3.5-turbo-0613"):
     if model in {
         "gpt-3.5-turbo-0613",
         "gpt-3.5-turbo-16k-0613",
+        "gpt-3.5-turbo-1106",
         "gpt-4-0314",
         "gpt-4-32k-0314",
         "gpt-4-0613",
         "gpt-4-32k-0613",
+        "gpt-4-1106-preview",
     }:
         tokens_per_message = 3
         tokens_per_name = 1

tests/conftest.py

@@ -6,14 +6,15 @@
 @File    : conftest.py
 """
 
+import asyncio
+import logging
+import re
 from unittest.mock import Mock
 
 import pytest
 
 from metagpt.logs import logger
 from metagpt.provider.openai_api import OpenAIGPTAPI as GPTAPI
-import asyncio
-import re
 
 
 class Context:
@@ -68,3 +69,14 @@ def proxy():
 
     server = asyncio.get_event_loop().run_until_complete(asyncio.start_server(handle_client, "127.0.0.1", 0))
     return "http://{}:{}".format(*server.sockets[0].getsockname())
+
+
+# see https://github.com/Delgan/loguru/issues/59#issuecomment-466591978
+@pytest.fixture
+def loguru_caplog(caplog):
+    class PropogateHandler(logging.Handler):
+        def emit(self, record):
+            logging.getLogger(record.name).handle(record)
+
+    logger.add(PropogateHandler(), format="{message}")
+    yield caplog

tests/metagpt/test_subscription.py

@@ -0,0 +1,102 @@
+import asyncio
+
+import pytest
+
+from metagpt.roles import Role
+from metagpt.schema import Message
+from metagpt.subscription import SubscriptionRunner
+
+
+@pytest.mark.asyncio
+async def test_subscription_run():
+    callback_done = 0
+
+    async def trigger():
+        while True:
+            yield Message("the latest news about OpenAI")
+            await asyncio.sleep(3600 * 24)
+
+    class MockRole(Role):
+        async def run(self, message=None):
+            return Message("")
+
+    async def callback(message):
+        nonlocal callback_done
+        callback_done += 1
+
+    runner = SubscriptionRunner()
+
+    roles = []
+    for _ in range(2):
+        role = MockRole()
+        roles.append(role)
+        await runner.subscribe(role, trigger(), callback)
+
+    task = asyncio.get_running_loop().create_task(runner.run())
+
+    for _ in range(10):
+        if callback_done == 2:
+            break
+        await asyncio.sleep(0)
+    else:
+        raise TimeoutError("callback not call")
+
+    role = roles[0]
+    assert role in runner.tasks
+    await runner.unsubscribe(roles[0])
+
+    for _ in range(10):
+        if role not in runner.tasks:
+            break
+        await asyncio.sleep(0)
+    else:
+        raise TimeoutError("callback not call")
+
+    task.cancel()
+    for i in runner.tasks.values():
+        i.cancel()
+
+
+@pytest.mark.asyncio
+async def test_subscription_run_error(loguru_caplog):
+    async def trigger1():
+        while True:
+            yield Message("the latest news about OpenAI")
+            await asyncio.sleep(3600 * 24)
+
+    async def trigger2():
+        yield Message("the latest news about OpenAI")
+
+    class MockRole1(Role):
+        async def run(self, message=None):
+            raise RuntimeError
+
+    class MockRole2(Role):
+        async def run(self, message=None):
+            return Message("")
+
+    async def callback(msg: Message):
+        print(msg)
+
+    runner = SubscriptionRunner()
+    await runner.subscribe(MockRole1(), trigger1(), callback)
+    with pytest.raises(RuntimeError):
+        await runner.run()
+
+    await runner.subscribe(MockRole2(), trigger2(), callback)
+    task = asyncio.get_running_loop().create_task(runner.run(False))
+
+    for _ in range(10):
+        if not runner.tasks:
+            break
+        await asyncio.sleep(0)
+    else:
+        raise TimeoutError("wait runner tasks empty timeout")
+
+    task.cancel()
+    for i in runner.tasks.values():
+        i.cancel()
+    assert len(loguru_caplog.records) >= 2
+    logs = "".join(loguru_caplog.messages)
+    assert "run error" in logs
+    assert "has completed" in logs