Merge branch 'feat-exp-pool' into 'mgx_ops'

Feat exp pool

See merge request pub/MetaGPT!213

metagpt/actions/action.py

@@ -90,7 +90,7 @@ class Action(SerializationMixin, ContextMixin, BaseModel):
         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(context=context, llm=self.llm)
+        return await self.node.fill(req=context, llm=self.llm)
 
     async def run(self, *args, **kwargs):
         """Run action"""

metagpt/actions/action_node.py

@@ -18,6 +18,8 @@ from tenacity import retry, stop_after_attempt, wait_random_exponential
 
 from metagpt.actions.action_outcls_registry import register_action_outcls
 from metagpt.const import MARKDOWN_TITLE_PREFIX, USE_CONFIG_TIMEOUT
+from metagpt.exp_pool import exp_cache
+from metagpt.exp_pool.serializers import ActionNodeSerializer
 from metagpt.llm import BaseLLM
 from metagpt.logs import logger
 from metagpt.provider.postprocess.llm_output_postprocess import llm_output_postprocess
@@ -465,9 +467,11 @@ class ActionNode:
 
         return self
 
+    @exp_cache(serializer=ActionNodeSerializer())
     async def fill(
         self,
-        context,
+        *,
+        req,
         llm,
         schema="json",
         mode="auto",
@@ -478,7 +482,7 @@ class ActionNode:
     ):
         """Fill the node(s) with mode.
 
-        :param context: Everything we should know when filling node.
+        :param req: Everything we should know when filling node.
         :param llm: Large Language Model with pre-defined system message.
         :param schema: json/markdown, determine example and output format.
          - raw: free form text
@@ -497,7 +501,7 @@ class ActionNode:
         :return: self
         """
         self.set_llm(llm)
-        self.set_context(context)
+        self.set_context(req)
         if self.schema:
             schema = self.schema
 

metagpt/actions/design_api.py

@@ -178,12 +178,12 @@ class WriteDesign(Action):
         )
 
     async def _new_system_design(self, context):
-        node = await DESIGN_API_NODE.fill(context=context, llm=self.llm, schema=self.prompt_schema)
+        node = await DESIGN_API_NODE.fill(req=context, llm=self.llm, schema=self.prompt_schema)
         return node
 
     async def _merge(self, prd_doc, system_design_doc):
         context = NEW_REQ_TEMPLATE.format(old_design=system_design_doc.content, context=prd_doc.content)
-        node = await REFINED_DESIGN_NODE.fill(context=context, llm=self.llm, schema=self.prompt_schema)
+        node = await REFINED_DESIGN_NODE.fill(req=context, llm=self.llm, schema=self.prompt_schema)
         system_design_doc.content = node.instruct_content.model_dump_json()
         return system_design_doc
 

metagpt/actions/generate_questions.py

@@ -22,4 +22,4 @@ class GenerateQuestions(Action):
     name: str = "GenerateQuestions"
 
     async def run(self, context) -> ActionNode:
-        return await QUESTIONS.fill(context=context, llm=self.llm)
+        return await QUESTIONS.fill(req=context, llm=self.llm)

metagpt/actions/prepare_interview.py

@@ -22,4 +22,4 @@ class PrepareInterview(Action):
     name: str = "PrepareInterview"
 
     async def run(self, context):
-        return await QUESTIONS.fill(context=context, llm=self.llm)
+        return await QUESTIONS.fill(req=context, llm=self.llm)

metagpt/actions/project_management.py

@@ -151,12 +151,12 @@ class WriteTasks(Action):
         return task_doc
 
     async def _run_new_tasks(self, context: str):
-        node = await PM_NODE.fill(context, self.llm, schema=self.prompt_schema)
+        node = await PM_NODE.fill(req=context, llm=self.llm, schema=self.prompt_schema)
         return node
 
     async def _merge(self, system_design_doc, task_doc) -> Document:
         context = NEW_REQ_TEMPLATE.format(context=system_design_doc.content, old_task=task_doc.content)
-        node = await REFINED_PM_NODE.fill(context, self.llm, schema=self.prompt_schema)
+        node = await REFINED_PM_NODE.fill(req=context, llm=self.llm, schema=self.prompt_schema)
         task_doc.content = node.instruct_content.model_dump_json()
         return task_doc
 

metagpt/actions/write_code_an_draft.py

@@ -578,7 +578,7 @@ class WriteCodeAN(Action):
 
     async def run(self, context):
         self.llm.system_prompt = "You are an outstanding engineer and can implement any code"
-        return await WRITE_MOVE_NODE.fill(context=context, llm=self.llm, schema="json")
+        return await WRITE_MOVE_NODE.fill(req=context, llm=self.llm, schema="json")
 
 
 async def main():

metagpt/actions/write_code_plan_and_change_an.py

@@ -229,7 +229,7 @@ class WriteCodePlanAndChange(Action):
             code=await self.get_old_codes(),
         )
         logger.info("Writing code plan and change..")
-        return await WRITE_CODE_PLAN_AND_CHANGE_NODE.fill(context=context, llm=self.llm, schema="json")
+        return await WRITE_CODE_PLAN_AND_CHANGE_NODE.fill(req=context, llm=self.llm, schema="json")
 
     async def get_old_codes(self) -> str:
         old_codes = await self.repo.srcs.get_all()

metagpt/actions/write_prd.py

@@ -211,7 +211,7 @@ class WritePRD(Action):
         context = CONTEXT_TEMPLATE.format(requirements=requirement, project_name=project_name)
         exclude = [PROJECT_NAME.key] if project_name else []
         node = await WRITE_PRD_NODE.fill(
-            context=context, llm=self.llm, exclude=exclude, schema=self.prompt_schema
+            req=context, llm=self.llm, exclude=exclude, schema=self.prompt_schema
         )  # schema=schema
         return node
 
@@ -238,7 +238,7 @@ class WritePRD(Action):
     async def _is_bugfix(self, context: str) -> bool:
         if not self.repo.code_files_exists():
             return False
-        node = await WP_ISSUE_TYPE_NODE.fill(context, self.llm)
+        node = await WP_ISSUE_TYPE_NODE.fill(req=context, llm=self.llm)
         return node.get("issue_type") == "BUG"
 
     async def get_related_docs(self, req: Document, docs: list[Document]) -> list[Document]:
@@ -248,14 +248,14 @@ class WritePRD(Action):
 
     async def _is_related(self, req: Document, old_prd: Document) -> bool:
         context = NEW_REQ_TEMPLATE.format(old_prd=old_prd.content, requirements=req.content)
-        node = await WP_IS_RELATIVE_NODE.fill(context, self.llm)
+        node = await WP_IS_RELATIVE_NODE.fill(req=context, llm=self.llm)
         return node.get("is_relative") == "YES"
 
     async def _merge(self, req: Document, related_doc: Document) -> Document:
         if not self.project_name:
             self.project_name = Path(self.project_path).name
         prompt = NEW_REQ_TEMPLATE.format(requirements=req.content, old_prd=related_doc.content)
-        node = await REFINED_PRD_NODE.fill(context=prompt, llm=self.llm, schema=self.prompt_schema)
+        node = await REFINED_PRD_NODE.fill(req=prompt, llm=self.llm, schema=self.prompt_schema)
         related_doc.content = node.instruct_content.model_dump_json()
         await self._rename_workspace(node)
         return related_doc

metagpt/actions/write_review.py

@@ -36,4 +36,4 @@ class WriteReview(Action):
     name: str = "WriteReview"
 
     async def run(self, context):
-        return await WRITE_REVIEW_NODE.fill(context=context, llm=self.llm, schema="json")
+        return await WRITE_REVIEW_NODE.fill(req=context, llm=self.llm, schema="json")

metagpt/config2.py

@@ -13,6 +13,7 @@ from pydantic import BaseModel, model_validator
 
 from metagpt.configs.browser_config import BrowserConfig
 from metagpt.configs.embedding_config import EmbeddingConfig
+from metagpt.configs.exp_pool_config import ExperiencePoolConfig
 from metagpt.configs.llm_config import LLMConfig, LLMType
 from metagpt.configs.mermaid_config import MermaidConfig
 from metagpt.configs.redis_config import RedisConfig
@@ -71,6 +72,9 @@ class Config(CLIParams, YamlModel):
     enable_longterm_memory: bool = False
     code_review_k_times: int = 2
 
+    # Experience Pool Parameters
+    exp_pool: ExperiencePoolConfig = ExperiencePoolConfig()
+
     # Will be removed in the future
     metagpt_tti_url: str = ""
     language: str = "English"

metagpt/configs/exp_pool_config.py

@@ -0,0 +1,9 @@
+from pydantic import Field
+
+from metagpt.utils.yaml_model import YamlModel
+
+
+class ExperiencePoolConfig(YamlModel):
+    enable_read: bool = Field(default=False, description="Enable to read from experience pool.")
+    enable_write: bool = Field(default=False, description="Enable to write to experience pool.")
+    persist_path: str = Field(default=".chroma_exp_data", description="The persist path for experience pool.")

metagpt/exp_pool/__init__.py

@@ -0,0 +1,6 @@
+"""Experience pool init."""
+
+from metagpt.exp_pool.manager import exp_manager
+from metagpt.exp_pool.decorator import exp_cache
+
+__all__ = ["exp_manager", "exp_cache"]

metagpt/exp_pool/context_builders/__init__.py

@@ -0,0 +1,7 @@
+"""Context builders init."""
+
+from metagpt.exp_pool.context_builders.base import BaseContextBuilder
+from metagpt.exp_pool.context_builders.simple import SimpleContextBuilder
+from metagpt.exp_pool.context_builders.role_zero import RoleZeroContextBuilder
+
+__all__ = ["BaseContextBuilder", "SimpleContextBuilder", "RoleZeroContextBuilder"]

metagpt/exp_pool/context_builders/action_node.py

@@ -0,0 +1,30 @@
+"""Action Node context builder."""
+
+from typing import Any
+
+from metagpt.exp_pool.context_builders.base import BaseContextBuilder
+
+ACTION_NODE_CONTEXT_TEMPLATE = """
+{req}
+
+### Experiences
+-----
+{exps}
+-----
+
+## Instruction
+Consider **Experiences** to generate a better answer.
+"""
+
+
+class ActionNodeContextBuilder(BaseContextBuilder):
+    async def build(self, req: Any) -> str:
+        """Builds the action node context string.
+
+        If there are no experiences, returns the original `req`;
+        otherwise returns context with `req` and formatted experiences.
+        """
+
+        exps = self.format_exps()
+
+        return ACTION_NODE_CONTEXT_TEMPLATE.format(req=req, exps=exps) if exps else req

metagpt/exp_pool/context_builders/base.py

@@ -0,0 +1,41 @@
+"""Base context builder."""
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict
+
+from metagpt.exp_pool.schema import Experience
+
+EXP_TEMPLATE = """Given the request: {req}, We can get the response: {resp}, which scored: {score}."""
+
+
+class BaseContextBuilder(BaseModel, ABC):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    exps: list[Experience] = []
+
+    @abstractmethod
+    async def build(self, req: Any) -> Any:
+        """Build context from req.
+
+        Do not modify `req`. If modification is necessary, use copy.deepcopy to create a copy first.
+        """
+
+    def format_exps(self) -> str:
+        """Format experiences into a numbered list of strings.
+
+        Example:
+            1. Given the request: req1, We can get the response: resp1, which scored: 8.
+            2. Given the request: req2, We can get the response: resp2, which scored: 9.
+
+        Returns:
+            str: The formatted experiences as a string.
+        """
+
+        result = []
+        for i, exp in enumerate(self.exps, start=1):
+            score_val = exp.metric.score.val if exp.metric and exp.metric.score else "N/A"
+            result.append(f"{i}. " + EXP_TEMPLATE.format(req=exp.req, resp=exp.resp, score=score_val))
+
+        return "\n".join(result)

metagpt/exp_pool/context_builders/role_zero.py

@@ -0,0 +1,56 @@
+"""RoleZero context builder."""
+
+import copy
+import re
+from typing import Any
+
+from metagpt.exp_pool.context_builders.base import BaseContextBuilder
+
+
+class RoleZeroContextBuilder(BaseContextBuilder):
+    async def build(self, req: Any) -> list[dict]:
+        """Builds the role zero context string.
+
+        Note:
+            1. The expected format for `req`, e.g., [{...}, {"role": "user", "content": "context"}].
+            2. Returns the original `req` if it is empty.
+            3. Creates a copy of req and replaces the example content in the copied req with actual experiences.
+        """
+
+        if not req:
+            return req
+
+        exps = self.format_exps()
+        if not exps:
+            return req
+
+        req_copy = copy.deepcopy(req)
+
+        req_copy[-1]["content"] = self.replace_example_content(req_copy[-1].get("content", ""), exps)
+
+        return req_copy
+
+    def replace_example_content(self, text: str, new_example_content: str) -> str:
+        return self.replace_content_between_markers(text, "# Example", "# Instruction", new_example_content)
+
+    @staticmethod
+    def replace_content_between_markers(text: str, start_marker: str, end_marker: str, new_content: str) -> str:
+        """Replace the content between `start_marker` and `end_marker` in the text with `new_content`.
+
+        Args:
+            text (str): The original text.
+            new_content (str): The new content to replace the old content.
+            start_marker (str): The marker indicating the start of the content to be replaced, such as '# Example'.
+            end_marker (str): The marker indicating the end of the content to be replaced, such as '# Instruction'.
+
+        Returns:
+            str: The text with the content replaced.
+        """
+
+        pattern = re.compile(f"({start_marker}\n)(.*?)(\n{end_marker})", re.DOTALL)
+
+        def replacement(match):
+            return f"{match.group(1)}{new_content}\n{match.group(3)}"
+
+        replaced_text = pattern.sub(replacement, text)
+        return replaced_text

metagpt/exp_pool/context_builders/simple.py

@@ -0,0 +1,26 @@
+"""Simple context builder."""
+
+
+from typing import Any
+
+from metagpt.exp_pool.context_builders.base import BaseContextBuilder
+
+SIMPLE_CONTEXT_TEMPLATE = """
+## Context
+
+### Experiences
+-----
+{exps}
+-----
+
+## User Requirement
+{req}
+
+## Instruction
+Consider **Experiences** to generate a better answer.
+"""
+
+
+class SimpleContextBuilder(BaseContextBuilder):
+    async def build(self, req: Any) -> str:
+        return SIMPLE_CONTEXT_TEMPLATE.format(req=req, exps=self.format_exps())

metagpt/exp_pool/decorator.py

@@ -0,0 +1,211 @@
+"""Experience Decorator."""
+
+import asyncio
+import functools
+from typing import Any, Callable, Optional, TypeVar
+
+from pydantic import BaseModel, ConfigDict, model_validator
+
+from metagpt.config2 import config
+from metagpt.exp_pool.context_builders import BaseContextBuilder, SimpleContextBuilder
+from metagpt.exp_pool.manager import ExperienceManager, exp_manager
+from metagpt.exp_pool.perfect_judges import BasePerfectJudge, SimplePerfectJudge
+from metagpt.exp_pool.schema import Experience, Metric, QueryType, Score
+from metagpt.exp_pool.scorers import BaseScorer, SimpleScorer
+from metagpt.exp_pool.serializers import BaseSerializer, SimpleSerializer
+from metagpt.logs import logger
+from metagpt.utils.async_helper import NestAsyncio
+from metagpt.utils.exceptions import handle_exception
+
+ReturnType = TypeVar("ReturnType")
+
+
+def exp_cache(
+    _func: Optional[Callable[..., ReturnType]] = None,
+    query_type: QueryType = QueryType.SEMANTIC,
+    manager: Optional[ExperienceManager] = None,
+    scorer: Optional[BaseScorer] = None,
+    perfect_judge: Optional[BasePerfectJudge] = None,
+    context_builder: Optional[BaseContextBuilder] = None,
+    serializer: Optional[BaseSerializer] = None,
+    tag: Optional[str] = None,
+):
+    """Decorator to get a perfect experience, otherwise, it executes the function, and create a new experience.
+
+    Note:
+        1. This can be applied to both synchronous and asynchronous functions.
+        2. The function must have a `req` parameter, and it must be provided as a keyword argument.
+        3. If `config.exp_pool.enable_read` is False, the decorator will just directly execute the function.
+        4. If `config.exp_pool.enable_write` is False, the decorator will skip evaluating and saving the experience.
+
+    Args:
+        _func: Just to make the decorator more flexible, for example, it can be used directly with @exp_cache by default, without the need for @exp_cache().
+        query_type: The type of query to be used when fetching experiences.
+        manager: How to fetch, evaluate and save experience, etc. Default to `exp_manager`.
+        scorer: Evaluate experience. Default to `SimpleScorer()`.
+        perfect_judge: Determines if an experience is perfect. Defaults to `SimplePerfectJudge()`.
+        context_builder: Build the context from exps and the function parameters. Default to `SimpleContextBuilder()`.
+        serializer: Serializes the request and the function's return value for storage, deserializes the stored response back to the function's return value. Defaults to `SimpleSerializer()`.
+        tag: An optional tag for the experience. Default to `ClassName.method_name` or `function_name`.
+    """
+
+    def decorator(func: Callable[..., ReturnType]) -> Callable[..., ReturnType]:
+        if not config.exp_pool.enable_read:
+            return func
+
+        @functools.wraps(func)
+        async def get_or_create(args: Any, kwargs: Any) -> ReturnType:
+            handler = ExpCacheHandler(
+                func=func,
+                args=args,
+                kwargs=kwargs,
+                query_type=query_type,
+                exp_manager=manager,
+                exp_scorer=scorer,
+                exp_perfect_judge=perfect_judge,
+                context_builder=context_builder,
+                serializer=serializer,
+                tag=tag,
+            )
+
+            await handler.fetch_experiences()
+
+            if exp := await handler.get_one_perfect_exp():
+                return exp
+
+            await handler.execute_function()
+
+            if config.exp_pool.enable_write:
+                await handler.process_experience()
+
+            return handler._raw_resp
+
+        return ExpCacheHandler.choose_wrapper(func, get_or_create)
+
+    return decorator(_func) if _func else decorator
+
+
+class ExpCacheHandler(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    func: Callable
+    args: Any
+    kwargs: Any
+    query_type: QueryType = QueryType.SEMANTIC
+    exp_manager: Optional[ExperienceManager] = None
+    exp_scorer: Optional[BaseScorer] = None
+    exp_perfect_judge: Optional[BasePerfectJudge] = None
+    context_builder: Optional[BaseContextBuilder] = None
+    serializer: Optional[BaseSerializer] = None
+    tag: Optional[str] = None
+
+    _exps: list[Experience] = None
+    _req: str = ""
+    _resp: str = ""
+    _raw_resp: Any = None
+    _score: Score = None
+
+    @model_validator(mode="after")
+    def initialize(self):
+        """Initialize default values for optional parameters if they are None.
+
+        This is necessary because the decorator might pass None, which would override the default values set by Field.
+        """
+
+        self._validate_params()
+
+        self.exp_manager = self.exp_manager or exp_manager
+        self.exp_scorer = self.exp_scorer or SimpleScorer()
+        self.exp_perfect_judge = self.exp_perfect_judge or SimplePerfectJudge()
+        self.context_builder = self.context_builder or SimpleContextBuilder()
+        self.serializer = self.serializer or SimpleSerializer()
+        self.tag = self.tag or self._generate_tag()
+
+        self._req = self.serializer.serialize_req(**self.kwargs)
+
+        return self
+
+    async def fetch_experiences(self):
+        """Fetch experiences by query_type."""
+
+        self._exps = await self.exp_manager.query_exps(self._req, query_type=self.query_type, tag=self.tag)
+
+    async def get_one_perfect_exp(self) -> Optional[Any]:
+        """Get a potentially perfect experience, and resolve resp."""
+
+        for exp in self._exps:
+            if await self.exp_perfect_judge.is_perfect_exp(exp, self._req, *self.args, **self.kwargs):
+                logger.info(f"Get one perfect experience: {exp.req[:20]}...")
+                return self.serializer.deserialize_resp(exp.resp)
+
+        return None
+
+    async def execute_function(self):
+        """Execute the function, and save resp."""
+
+        self._raw_resp = await self._execute_function()
+        self._resp = self.serializer.serialize_resp(self._raw_resp)
+
+    @handle_exception
+    async def process_experience(self):
+        """Process experience.
+
+        Evaluates and saves experience.
+        Use `handle_exception` to ensure robustness, do not stop subsequent operations.
+        """
+
+        await self.evaluate_experience()
+        self.save_experience()
+
+    async def evaluate_experience(self):
+        """Evaluate the experience, and save the score."""
+
+        self._score = await self.exp_scorer.evaluate(self._req, self._resp)
+
+    def save_experience(self):
+        """Save the new experience."""
+
+        exp = Experience(req=self._req, resp=self._resp, tag=self.tag, metric=Metric(score=self._score))
+        self.exp_manager.create_exp(exp)
+
+    @staticmethod
+    def choose_wrapper(func, wrapped_func):
+        """Choose how to run wrapped_func based on whether the function is asynchronous."""
+
+        async def async_wrapper(*args, **kwargs):
+            return await wrapped_func(args, kwargs)
+
+        def sync_wrapper(*args, **kwargs):
+            NestAsyncio.apply_once()
+            return asyncio.get_event_loop().run_until_complete(wrapped_func(args, kwargs))
+
+        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
+
+    def _validate_params(self):
+        if "req" not in self.kwargs:
+            raise ValueError("`req` must be provided as a keyword argument.")
+
+    def _generate_tag(self) -> str:
+        """Generates a tag for the self.func.
+
+        "ClassName.method_name" if the first argument is a class instance, otherwise just "function_name".
+        """
+
+        if self.args and hasattr(self.args[0], "__class__"):
+            cls_name = type(self.args[0]).__name__
+            return f"{cls_name}.{self.func.__name__}"
+
+        return self.func.__name__
+
+    async def _build_context(self) -> str:
+        self.context_builder.exps = self._exps
+
+        return await self.context_builder.build(self.kwargs["req"])
+
+    async def _execute_function(self):
+        self.kwargs["req"] = await self._build_context()
+
+        if asyncio.iscoroutinefunction(self.func):
+            return await self.func(*self.args, **self.kwargs)
+
+        return self.func(*self.args, **self.kwargs)

metagpt/exp_pool/manager.py

@@ -0,0 +1,116 @@
+"""Experience Manager."""
+
+from typing import TYPE_CHECKING, Any
+
+from pydantic import BaseModel, ConfigDict
+
+from metagpt.config2 import Config, config
+from metagpt.exp_pool.schema import (
+    DEFAULT_COLLECTION_NAME,
+    DEFAULT_SIMILARITY_TOP_K,
+    Experience,
+    QueryType,
+)
+from metagpt.logs import logger
+from metagpt.utils.exceptions import handle_exception
+
+if TYPE_CHECKING:
+    from llama_index.vector_stores.chroma import ChromaVectorStore
+
+
+class ExperienceManager(BaseModel):
+    """ExperienceManager manages the lifecycle of experiences, including CRUD and optimization.
+
+    Args:
+        config (Config): Configuration for managing experiences.
+        _storage (SimpleEngine): Engine to handle the storage and retrieval of experiences.
+        _vector_store (ChromaVectorStore): The actual place where vectors are stored.
+    """
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    config: Config = config
+
+    _storage: Any = None
+    _vector_store: Any = None
+
+    @property
+    def storage(self):
+        if self._storage is None:
+            try:
+                from metagpt.rag.engines import SimpleEngine
+                from metagpt.rag.schema import ChromaRetrieverConfig, LLMRankerConfig
+            except ImportError:
+                raise ImportError("To use the experience pool, you need to install the rag module.")
+
+            retriever_configs = [
+                ChromaRetrieverConfig(
+                    persist_path=self.config.exp_pool.persist_path,
+                    collection_name=DEFAULT_COLLECTION_NAME,
+                    similarity_top_k=DEFAULT_SIMILARITY_TOP_K,
+                )
+            ]
+            ranker_configs = [LLMRankerConfig(top_n=DEFAULT_SIMILARITY_TOP_K)]
+
+            self._storage: SimpleEngine = SimpleEngine.from_objs(
+                retriever_configs=retriever_configs, ranker_configs=ranker_configs
+            )
+            logger.debug(f"exp_pool config: {self.config.exp_pool}")
+
+        return self._storage
+
+    @property
+    def vector_store(self):
+        if not self._vector_store:
+            self._vector_store: ChromaVectorStore = self.storage._retriever._vector_store
+
+        return self._vector_store
+
+    @handle_exception
+    def create_exp(self, exp: Experience):
+        """Adds an experience to the storage if writing is enabled.
+
+        Args:
+            exp (Experience): The experience to add.
+        """
+
+        if not self.config.exp_pool.enable_write:
+            return
+
+        self.storage.add_objs([exp])
+
+    @handle_exception(default_return=[])
+    async def query_exps(self, req: str, tag: str = "", query_type: QueryType = QueryType.SEMANTIC) -> list[Experience]:
+        """Retrieves and filters experiences.
+
+        Args:
+            req (str): The query string to retrieve experiences.
+            tag (str): Optional tag to filter the experiences by.
+            query_type (QueryType): Default semantic to vector matching. exact to same matching.
+
+        Returns:
+            list[Experience]: A list of experiences that match the args.
+        """
+
+        if not self.config.exp_pool.enable_read:
+            return []
+
+        nodes = await self.storage.aretrieve(req)
+        exps: list[Experience] = [node.metadata["obj"] for node in nodes]
+
+        # TODO: filter by metadata
+        if tag:
+            exps = [exp for exp in exps if exp.tag == tag]
+
+        if query_type == QueryType.EXACT:
+            exps = [exp for exp in exps if exp.req == req]
+
+        return exps
+
+    def get_exps_count(self) -> int:
+        """Get the total number of experiences."""
+
+        return self.vector_store._collection.count()
+
+
+exp_manager = ExperienceManager()

metagpt/exp_pool/perfect_judges/__init__.py

@@ -0,0 +1,6 @@
+"""Perfect judges init."""
+
+from metagpt.exp_pool.perfect_judges.base import BasePerfectJudge
+from metagpt.exp_pool.perfect_judges.simple import SimplePerfectJudge
+
+__all__ = ["BasePerfectJudge", "SimplePerfectJudge"]

metagpt/exp_pool/perfect_judges/base.py

@@ -0,0 +1,20 @@
+"""Base perfect judge."""
+
+from abc import ABC, abstractmethod
+
+from pydantic import BaseModel, ConfigDict
+
+from metagpt.exp_pool.schema import Experience
+
+
+class BasePerfectJudge(BaseModel, ABC):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    @abstractmethod
+    async def is_perfect_exp(self, exp: Experience, serialized_req: str, *args, **kwargs) -> bool:
+        """Determine whether the experience is perfect.
+
+        Args:
+            exp (Experience): The experience to evaluate.
+            serialized_req (str): The serialized request to compare against the experience's request.
+        """

metagpt/exp_pool/perfect_judges/simple.py

@@ -0,0 +1,27 @@
+"""Simple perfect judge."""
+
+
+from pydantic import ConfigDict
+
+from metagpt.exp_pool.perfect_judges.base import BasePerfectJudge
+from metagpt.exp_pool.schema import MAX_SCORE, Experience
+
+
+class SimplePerfectJudge(BasePerfectJudge):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    async def is_perfect_exp(self, exp: Experience, serialized_req: str, *args, **kwargs) -> bool:
+        """Determine whether the experience is perfect.
+
+        Args:
+            exp (Experience): The experience to evaluate.
+            serialized_req (str): The serialized request to compare against the experience's request.
+
+        Returns:
+            bool: True if the serialized request matches the experience's request and the experience's score is perfect, False otherwise.
+        """
+
+        if not exp.metric or not exp.metric.score:
+            return False
+
+        return serialized_req == exp.req and exp.metric.score.val == MAX_SCORE

metagpt/exp_pool/schema.py

@@ -0,0 +1,72 @@
+"""Experience schema."""
+
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+MAX_SCORE = 10
+
+DEFAULT_COLLECTION_NAME = "experience_pool"
+DEFAULT_SIMILARITY_TOP_K = 2
+
+
+class QueryType(str, Enum):
+    """Type of query experiences."""
+
+    EXACT = "exact"
+    SEMANTIC = "semantic"
+
+
+class ExperienceType(str, Enum):
+    """Experience Type."""
+
+    SUCCESS = "success"
+    FAILURE = "failure"
+    INSIGHT = "insight"
+
+
+class EntryType(Enum):
+    """Experience Entry Type."""
+
+    AUTOMATIC = "Automatic"
+    MANUAL = "Manual"
+
+
+class Score(BaseModel):
+    """Score in Metric."""
+
+    val: int = Field(default=1, description="Value of the score, Between 1 and 10, higher is better.")
+    reason: str = Field(default="", description="Reason for the value.")
+
+
+class Metric(BaseModel):
+    """Experience Metric."""
+
+    time_cost: float = Field(default=0.000, description="Time cost, the unit is milliseconds.")
+    money_cost: float = Field(default=0.000, description="Money cost, the unit is US dollars.")
+    score: Score = Field(default=None, description="Score, with value and reason.")
+
+
+class Trajectory(BaseModel):
+    """Experience Trajectory."""
+
+    plan: str = Field(default="", description="The plan.")
+    action: str = Field(default="", description="Action for the plan.")
+    observation: str = Field(default="", description="Output of the action.")
+    reward: int = Field(default=0, description="Measure the action.")
+
+
+class Experience(BaseModel):
+    """Experience."""
+
+    req: str = Field(..., description="")
+    resp: str = Field(..., description="The type is string/json/code.")
+    metric: Optional[Metric] = Field(default=None, description="Metric.")
+    exp_type: ExperienceType = Field(default=ExperienceType.SUCCESS, description="The type of experience.")
+    entry_type: EntryType = Field(default=EntryType.AUTOMATIC, description="Type of entry: Manual or Automatic.")
+    tag: str = Field(default="", description="Tagging experience.")
+    traj: Optional[Trajectory] = Field(default=None, description="Trajectory.")
+
+    def rag_key(self):
+        return self.req

metagpt/exp_pool/scorers/__init__.py

@@ -0,0 +1,6 @@
+"""Scorers init."""
+
+from metagpt.exp_pool.scorers.base import BaseScorer
+from metagpt.exp_pool.scorers.simple import SimpleScorer
+
+__all__ = ["BaseScorer", "SimpleScorer"]

metagpt/exp_pool/scorers/base.py

@@ -0,0 +1,15 @@
+"""Base scorer."""
+
+from abc import ABC, abstractmethod
+
+from pydantic import BaseModel, ConfigDict
+
+from metagpt.exp_pool.schema import Score
+
+
+class BaseScorer(BaseModel, ABC):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    @abstractmethod
+    async def evaluate(self, req: str, resp: str) -> Score:
+        """Evaluates the quality of a response relative to a given request."""

metagpt/exp_pool/scorers/simple.py

@@ -0,0 +1,65 @@
+"""Simple scorer."""
+
+import json
+
+from pydantic import Field
+
+from metagpt.exp_pool.schema import Score
+from metagpt.exp_pool.scorers.base import BaseScorer
+from metagpt.llm import LLM
+from metagpt.provider.base_llm import BaseLLM
+from metagpt.utils.common import CodeParser
+
+SIMPLE_SCORER_TEMPLATE = """
+Role: You are a highly efficient assistant, tasked with evaluating a response to a given request. The response is generated by a large language model (LLM). 
+
+I will provide you with a request and a corresponding response. Your task is to assess this response and provide a score from a human perspective.
+
+## Context
+### Request
+{req}
+
+### Response
+{resp}
+
+## Format Example
+```json
+{{
+    "val": "the value of the score, int from 1 to 10, higher is better.",
+    "reason": "an explanation supporting the score."
+}}
+```
+
+## Instructions
+- Understand the request and response given by the user.
+- Evaluate the response based on its quality relative to the given request.
+- Provide a score from 1 to 10, where 10 is the best.
+- Provide a reason supporting your score.
+
+## Constraint
+Format: Just print the result in json format like **Format Example**.
+
+## Action
+Follow instructions, generate output and make sure it follows the **Constraint**.
+"""
+
+
+class SimpleScorer(BaseScorer):
+    llm: BaseLLM = Field(default_factory=LLM)
+
+    async def evaluate(self, req: str, resp: str) -> Score:
+        """Evaluates the quality of a response relative to a given request, as scored by an LLM.
+
+        Args:
+            req (str): The request.
+            resp (str): The response.
+
+        Returns:
+            Score: An object containing the score (1-10) and the reasoning.
+        """
+
+        prompt = SIMPLE_SCORER_TEMPLATE.format(req=req, resp=resp)
+        resp = await self.llm.aask(prompt)
+        resp_json = json.loads(CodeParser.parse_code(resp, lang="json"))
+
+        return Score(**resp_json)

metagpt/exp_pool/serializers/__init__.py

@@ -0,0 +1,9 @@
+"""Serializers init."""
+
+from metagpt.exp_pool.serializers.base import BaseSerializer
+from metagpt.exp_pool.serializers.simple import SimpleSerializer
+from metagpt.exp_pool.serializers.action_node import ActionNodeSerializer
+from metagpt.exp_pool.serializers.role_zero import RoleZeroSerializer
+
+
+__all__ = ["BaseSerializer", "SimpleSerializer", "ActionNodeSerializer", "RoleZeroSerializer"]

metagpt/exp_pool/serializers/action_node.py

@@ -0,0 +1,36 @@
+"""ActionNode Serializer."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Type
+
+# Import ActionNode only for type checking to avoid circular imports
+if TYPE_CHECKING:
+    from metagpt.actions.action_node import ActionNode
+
+from metagpt.exp_pool.serializers.simple import SimpleSerializer
+
+
+class ActionNodeSerializer(SimpleSerializer):
+    def serialize_resp(self, resp: ActionNode) -> str:
+        return resp.instruct_content.model_dump_json()
+
+    def deserialize_resp(self, resp: str) -> ActionNode:
+        """Customized deserialization, it will be triggered when a perfect experience is found.
+
+        ActionNode cannot be serialized, it throws an error 'cannot pickle 'SSLContext' object'.
+        """
+
+        class InstructContent:
+            def __init__(self, json_data):
+                self.json_data = json_data
+
+            def model_dump_json(self):
+                return self.json_data
+
+        from metagpt.actions.action_node import ActionNode
+
+        action_node = ActionNode(key="", expected_type=Type[str], instruction="", example="")
+        action_node.instruct_content = InstructContent(resp)
+
+        return action_node

metagpt/exp_pool/serializers/base.py

@@ -0,0 +1,29 @@
+"""Base serializer."""
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict
+
+
+class BaseSerializer(BaseModel, ABC):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    @abstractmethod
+    def serialize_req(self, **kwargs) -> str:
+        """Serializes the request for storage.
+
+        Do not modify kwargs. If modification is necessary, use copy.deepcopy to create a copy first.
+        Note that copy.deepcopy may raise errors, such as TypeError: cannot pickle '_thread.RLock' object.
+        """
+
+    @abstractmethod
+    def serialize_resp(self, resp: Any) -> str:
+        """Serializes the function's return value for storage.
+
+        Do not modify resp. The rest is the same as `serialize_req`.
+        """
+
+    @abstractmethod
+    def deserialize_resp(self, resp: str) -> Any:
+        """Deserializes the stored response back to the function's return value"""

metagpt/exp_pool/serializers/role_zero.py

@@ -0,0 +1,58 @@
+"""RoleZero Serializer."""
+
+import copy
+import json
+
+from metagpt.exp_pool.serializers.simple import SimpleSerializer
+
+
+class RoleZeroSerializer(SimpleSerializer):
+    def serialize_req(self, **kwargs) -> str:
+        """Serialize the request for database storage, ensuring it is a string.
+
+        Only extracts the necessary content from `req` because `req` may be very lengthy and could cause embedding errors.
+
+        Args:
+            req (list[dict]): The request to be serialized. Example:
+                [
+                    {"role": "user", "content": "..."},
+                    {"role": "assistant", "content": "..."},
+                    {"role": "user", "content": "context"},
+                ]
+
+        Returns:
+            str: The serialized request as a JSON string.
+        """
+        req = kwargs.get("req", [])
+
+        if not req:
+            return ""
+
+        filtered_req = self._filter_req(req)
+
+        if state_data := kwargs.get("state_data"):
+            filtered_req.append({"role": "user", "content": state_data})
+
+        return json.dumps(filtered_req)
+
+    def _filter_req(self, req: list[dict]) -> list[dict]:
+        """Filter the `req` to include only necessary items.
+
+        Args:
+            req (list[dict]): The original request.
+
+        Returns:
+            list[dict]: The filtered request.
+        """
+
+        filtered_req = [copy.deepcopy(item) for item in req if self._is_useful_content(item["content"])]
+
+        return filtered_req
+
+    def _is_useful_content(self, content: str) -> bool:
+        """Currently, only the content of the file is considered, and more judgments can be added later."""
+
+        if "Command Editor.read executed: file_path" in content:
+            return True
+
+        return False

metagpt/exp_pool/serializers/simple.py

@@ -0,0 +1,22 @@
+"""Simple Serializer."""
+
+from typing import Any
+
+from metagpt.exp_pool.serializers.base import BaseSerializer
+
+
+class SimpleSerializer(BaseSerializer):
+    def serialize_req(self, **kwargs) -> str:
+        """Just use `str` to convert the request object into a string."""
+
+        return str(kwargs.get("req", ""))
+
+    def serialize_resp(self, resp: Any) -> str:
+        """Just use `str` to convert the response object into a string."""
+
+        return str(resp)
+
+    def deserialize_resp(self, resp: str) -> Any:
+        """Just return the string response as it is."""
+
+        return resp

metagpt/prompts/di/role_zero.py

@@ -8,7 +8,7 @@ Note:
 2. Carefully review your progress at the current task, if your actions so far has not fulfilled the task instruction, you should continue with current task. Otherwise, finish current task by Plan.finish_current_task explicitly.
 3. Each time you finish a task, use RoleZero.reply_to_human to report your progress.
 """
-
+# To ensure compatibility with hard-coded experience, do not add any other content between "# Example" and "# Available Commands".
 CMD_PROMPT = """
 # Data Structure
 class Task(BaseModel):
@@ -17,7 +17,7 @@ class Task(BaseModel):
     instruction: str = ""
     task_type: str = ""
     assignee: str = ""
-
+    
 # Available Task Types
 {task_type_desc}
 

metagpt/roles/di/role_zero.py

@@ -10,6 +10,9 @@ from pydantic import model_validator
 
 from metagpt.actions import Action, UserRequirement
 from metagpt.actions.di.run_command import RunCommand
+from metagpt.exp_pool import exp_cache
+from metagpt.exp_pool.context_builders import RoleZeroContextBuilder
+from metagpt.exp_pool.serializers import RoleZeroSerializer
 from metagpt.logs import logger
 from metagpt.prompts.di.role_zero import (
     CMD_PROMPT,
@@ -143,25 +146,42 @@ class RoleZero(Role):
         tool_info = json.dumps({tool.name: tool.schemas for tool in tools})
 
         ### Make Decision Dynamically ###
+        instruction = self.instruction.strip()
         prompt = self.cmd_prompt.format(
-            plan_status=plan_status,
-            current_task=current_task,
             example=example,
             available_commands=tool_info,
-            instruction=self.instruction.strip(),
             task_type_desc=self.task_type_desc,
+            plan_status=plan_status,
+            current_task=current_task,
+            instruction=instruction,
         )
         memory = self.rc.memory.get(self.memory_k)
         memory = await self.parse_browser_actions(memory)
-        context = self.llm.format_msg(memory + [UserMessage(content=prompt)])
-        # print(*context, sep="\n" + "*" * 5 + "\n")
+
+        req = self.llm.format_msg(memory + [UserMessage(content=prompt)])
         async with ThoughtReporter(enable_llm_stream=True) as reporter:
             await reporter.async_report({"type": "react"})
-            self.command_rsp = await self.llm.aask(context, system_msgs=self.system_msg)
+            state_data = dict(
+                plan_status=plan_status,
+                current_task=current_task,
+                instruction=instruction,
+            )
+            self.command_rsp = await self.llm_cached_aask(req=req, system_msgs=self.system_msg, state_data=state_data)
+
         self.rc.memory.add(AIMessage(content=self.command_rsp))
 
         return True
 
+    @exp_cache(context_builder=RoleZeroContextBuilder(), serializer=RoleZeroSerializer())
+    async def llm_cached_aask(self, *, req: list[dict], system_msgs: list[str], **kwargs) -> str:
+        """Use `exp_cache` to automatically manage experiences.
+
+        The `RoleZeroContextBuilder` attempts to add experiences to `req`.
+        The `RoleZeroSerializer` extracts essential parts of `req` for the experience pool, trimming lengthy entries to retain only necessary parts.
+        """
+
+        return await self.llm.aask(req, system_msgs=system_msgs)
+
     async def parse_browser_actions(self, memory: List[Message]) -> List[Message]:
         if not self.browser.is_empty_page:
             pattern = re.compile(r"Command Browser\.(\w+) executed")

metagpt/strategy/experience_retriever.py

@@ -791,13 +791,13 @@ Explanation: I will first need to read the system design document and the projec
     {
         "command_name": "Editor.read",
         "args": {
-            "path": "/tmp/docs/project_schedule.json"
+            "path": "/tmp/project_schedule.json"
         }
     },
     {
         "command_name": "Editor.read",
         "args": {
-            "path": "/tmp/docs/system_design.json"
+            "path": "/tmp/system_design.json"
         }
     }
 ]

metagpt/strategy/solver.py

@@ -39,7 +39,7 @@ class NaiveSolver(BaseSolver):
         self.graph.topological_sort()
         for key in self.graph.execution_order:
             op = self.graph.nodes[key]
-            await op.fill(self.context, self.llm, mode="root")
+            await op.fill(req=self.context, llm=self.llm, mode="root")
 
 
 class TOTSolver(BaseSolver):

metagpt/utils/token_counter.py

@@ -154,6 +154,7 @@ TOKEN_MAX = {
     "gpt-4-1106-preview": 128000,
     "gpt-4-vision-preview": 128000,
     "gpt-4-1106-vision-preview": 128000,
+    "gpt-4-turbo": 128000,
     "gpt-4": 8192,
     "gpt-4-0613": 8192,
     "gpt-4-32k": 32768,