1  """Init experience pool.
 2  
 3  Put some useful experiences into the experience pool.
 4  """
 5  
 6  import asyncio
 7  import json
 8  from pathlib import Path
 9  
10  from metagpt.const import EXAMPLE_DATA_PATH
11  from metagpt.exp_pool import get_exp_manager
12  from metagpt.exp_pool.schema import EntryType, Experience, Metric, Score
13  from metagpt.logs import logger
14  from metagpt.utils.common import aread
15  
16  
17  async def load_file(filepath) -> list[dict]:
18      """Asynchronously loads and parses a JSON file.
19  
20      Args:
21          filepath: Path to the JSON file.
22  
23      Returns:
24          A list of dictionaries parsed from the JSON file.
25      """
26  
27      return json.loads(await aread(filepath))
28  
29  
30  async def add_exp(req: str, resp: str, tag: str, metric: Metric = None):
31      """Adds a new experience to the experience pool.
32  
33      Args:
34          req: The request string.
35          resp: The response string.
36          tag: A tag for categorizing the experience.
37          metric: Optional metric for the experience. Defaults to a score of 10.
38  
39      """
40  
41      exp = Experience(
42          req=req,
43          resp=resp,
44          entry_type=EntryType.MANUAL,
45          tag=tag,
46          metric=metric or Metric(score=Score(val=10, reason="Manual")),
47      )
48      exp_manager = get_exp_manager()
49      exp_manager.is_writable = True
50  
51      exp_manager.create_exp(exp)
52      logger.info(f"New experience created for the request `{req[:10]}`.")
53  
54  
55  async def add_exps(exps: list, tag: str):
56      """Adds multiple experiences to the experience pool.
57  
58      Args:
59          exps: A list of experience dictionaries.
60          tag: A tag for categorizing the experiences.
61  
62      """
63      tasks = [
64          add_exp(req=exp["req"] if isinstance(exp["req"], str) else json.dumps(exp["req"]), resp=exp["resp"], tag=tag)
65          for exp in exps
66      ]
67      await asyncio.gather(*tasks)
68  
69  
70  async def add_exps_from_file(tag: str, filepath: Path):
71      """Loads experiences from a file and adds them to the experience pool.
72  
73      Args:
74          tag: A tag for categorizing the experiences.
75          filepath: Path to the file containing experiences.
76  
77      """
78  
79      exps = await load_file(filepath)
80      await add_exps(exps, tag)
81  
82  
83  def query_exps_count():
84      """Queries and logs the total count of experiences in the pool."""
85      exp_manager = get_exp_manager()
86      count = exp_manager.get_exps_count()
87      logger.info(f"Experiences Count: {count}")
88  
89  
90  async def main():
91      await add_exps_from_file("TeamLeader.llm_cached_aask", EXAMPLE_DATA_PATH / "exp_pool/team_leader_exps.json")
92      await add_exps_from_file("Engineer2.llm_cached_aask", EXAMPLE_DATA_PATH / "exp_pool/engineer_exps.json")
93      query_exps_count()
94  
95  
96  if __name__ == "__main__":
97      asyncio.run(main())