add many op

experiencemaker/llm/base_llm.py

@@ -1,6 +1,6 @@
 import time
 from abc import ABC
-from typing import List, Literal
+from typing import List, Literal, Callable
 
 from loguru import logger
 from pydantic import Field, BaseModel
@@ -35,10 +35,14 @@ class BaseLLM(BaseModel, ABC):
     def _chat(self, messages: List[Message], tools: List[BaseTool] = None, **kwargs) -> Message:
         raise NotImplementedError
 
-    def chat(self, messages: List[Message], tools: List[BaseTool] = None, **kwargs) -> Message | None:
+    def chat(self, messages: List[Message], tools: List[BaseTool] = None, callback_fn: Callable = None, **kwargs):
         for i in range(self.max_retries):
             try:
-                return self._chat(messages, tools, **kwargs)
+                message: Message = self._chat(messages, tools, **kwargs)
+                if callback_fn:
+                    return callback_fn(message)
+                else:
+                    return message
 
             except Exception as e:
                 logger.exception(f"chat with model={self.model_name} encounter error with e={e.args}")

experiencemaker/op/base_op.py

@@ -1,7 +1,10 @@
 from abc import abstractmethod, ABC
+from concurrent.futures import Future
 from pathlib import Path
+from typing import List
 
 from loguru import logger
+from tqdm import tqdm
 
 from experiencemaker.embedding_model import EMBEDDING_MODEL_REGISTRY
 from experiencemaker.embedding_model.base_embedding_model import BaseEmbeddingModel
@@ -29,11 +32,13 @@ class BaseOp(PromptMixin, ABC):
         self._embedding_model: BaseEmbeddingModel | None = None
         self._vector_store: BaseVectorStore | None = None
 
+        self.task_list: List[Future] = []
+
     def _prepare_prompt(self):
         if self.op_config.prompt_file_path:
             prompt_file_path = self.op_config.prompt_file_path
         else:
-            prompt_name = self.simple_name.replace("op", "").strip("_") + "_prompt.yaml"
+            prompt_name = self.simple_name.replace("_op", "_prompt.yaml")
             prompt_file_path = Path(__file__).parent / prompt_name
 
         # Load custom prompts from prompt file
@@ -62,6 +67,18 @@ class BaseOp(PromptMixin, ABC):
         except Exception as e:
             logger.exception(f"op={self.simple_name} execute failed, error={e.args}")
 
+    def submit_task(self, fn, *args, **kwargs):
+        task = self.context.thread_pool.submit(fn, *args, **kwargs)
+        self.task_list.append(task)
+        return self
+
+    def join_task(self, task_desc: str = None) -> list:
+        result = []
+        for task in tqdm(self.task_list, desc=task_desc or (self.simple_name + ".join_task")):
+            result.append(task.result())
+        self.task_list.clear()
+        return result
+
     @property
     def llm(self) -> BaseLLM:
         if self._llm is None:

experiencemaker/op/mock_op.py

@@ -8,15 +8,11 @@ from experiencemaker.op.base_op import BaseOp
 
 @OP_REGISTRY.register()
 class Mock1Op(BaseOp):
-
-    def __init__(self, a: int = 1, b: str = "2", **kwargs):
-        super().__init__(**kwargs)
-        self.a = a
-        self.b = b
-
     def execute(self):
-        time.sleep(3)
-        logger.info(f"enter class={self.__class__.__name__}. a={self.a} b={self.b}")
+        time.sleep(1)
+        a: int = self.op_params["a"]
+        b: str = self.op_params["b"]
+        logger.info(f"enter class={self.__class__.__name__}. a={a} b={b}")
 
 
 @OP_REGISTRY.register()

experiencemaker/op/prompt_mixin.py

@@ -30,7 +30,9 @@ class PromptMixin:
         for key, value in prompt_dict.items():
             if isinstance(value, str):
                 if key in self._prompt_dict:
-                    logger.warning(f"prompt_dict key={key} already exists")
+                    self._prompt_dict[key] = value
+                    logger.warning(f"prompt_dict key={key} overwrite!")
+
                 else:
                     self._prompt_dict[key] = value
                     logger.info(f"add prompt_dict key={key}")
@@ -67,3 +69,6 @@ class PromptMixin:
             prompt = prompt.format(**other_kwargs)
 
         return prompt
+
+    def get_prompt(self, key: str):
+        return self._prompt_dict[key]

experiencemaker/op/retriever/build_query_op.py

@@ -11,12 +11,13 @@ class BuildQueryOp(BaseOp):
     RETRIEVE_QUERY = "retrieve_query"
 
     def execute(self):
+        # @jiaji
         request: RetrieverRequest = self.context.request
         if request.query:
             query = request.query
 
         elif request.messages:
-            if not self.op_params.get("enable_llm_build"):
+            if self.op_params.get("enable_llm_build") is True:
                 execution_process = merge_messages_content(request.messages)
                 query = self.prompt_format(prompt_name="query_build", execution_process=execution_process)
             else:

experiencemaker/op/retriever/rerank_experience_op.py

@@ -0,0 +1 @@
+# at jiaji

experiencemaker/op/retriever/rewrite_experience_op.py

@@ -0,0 +1 @@
+# @jiaji

experiencemaker/op/summarizer/simple_compare_reward_fn.py

@@ -1,61 +0,0 @@
-import datetime
-from pathlib import Path
-
-from loguru import logger
-from pydantic import Field
-
-from experiencemaker.model.base_llm import BaseLLM
-from experiencemaker.module.prompt.prompt_mixin import PromptMixin
-from experiencemaker.module.reward_fn.base_reward_fn import BaseRewardFn
-from experiencemaker.schema.reward import Reward
-from experiencemaker.schema.trajectory import Trajectory, Message
-from experiencemaker.utils.util_function import get_html_match_content
-
-
-class SimpleCompareRewardFn(BaseRewardFn, PromptMixin):
-    llm: BaseLLM | None = Field(default=None)
-    eval_times: int = Field(default=5)
-    prompt_file_path: Path = Path(__file__).parent / "simple_compare_reward_fn_prompt.yaml"
-
-    def execute(self, trajectory: Trajectory = None, comp_traj: Trajectory = None, **kwargs) -> Reward:
-        query = trajectory.query
-        answer1 = trajectory.answer
-        answer2 = comp_traj.answer
-        logger.info("=" * 10 + f"answer1\n{answer1}\n" + "=" * 10 + f"answer2\n{answer2}\n")
-
-        valid_cnt = 0
-        better_cnt = 0
-        for i in range(self.eval_times):
-            now_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
-            user_prompt = self.prompt_handler.compare_prompt.format(
-                now_time=now_time,
-                query=query,
-                answer1=answer1,
-                answer2=answer2)
-
-            messages = [Message(content=user_prompt)]
-            action_msg = self.llm.chat(messages=messages)
-
-            rule: str = get_html_match_content(action_msg.content, "rule")
-            rule_based_comparison: str = get_html_match_content(action_msg.content, "rule_based_comparison")
-            result: str | None = get_html_match_content(action_msg.content, "result")
-            logger.info(f"round.{i} rule={rule} rule_based_comparison={rule_based_comparison} result={result}")
-            if result:
-                result = result.lower()
-                if "plan1" in result and "plan2" in result:
-                    logger.warning(f"both plan exists in result={result}")
-
-                elif "plan1" in result:
-                    valid_cnt += 1
-                    better_cnt += 1
-
-                elif "plan2" in result:
-                    valid_cnt += 1
-
-                else:
-                    logger.warning(f"no plan exists in result={result}")
-
-        outcome = 0
-        if valid_cnt > 0:
-            outcome = better_cnt / valid_cnt
-        return Reward(outcome=outcome)

experiencemaker/op/summarizer/simple_compare_reward_fn_prompt.yaml

@@ -1,30 +0,0 @@
-compare_prompt: |
-  # Role
-  You are a helpful assistant named BeyondAgent.
-  current time: {now_time}
-
-  # User Question
-  {query}
-
-  # Plan1 Answer
-  {answer1}
-
-  # Plan2 Answer
-  {answer2}
-
-  # Task
-  Based on the **User Question**, determine which plan provides a better answer. Response steps:
-  1. Consider which comparison rules apply, the rules for comparison can be macro-level dimensions or micro-level details.
-  2. Conduct a step-by-step comparison according to the rules.
-  3. Combine all results to arrive at a final answer.
-
-  # Output Example
-  <rule>
-  List the rules that could be used for comparison...
-  </rule>
-  <rule_based_comparison>
-  Conduct a step-by-step comparison according to the rules...
-  </rule_based_comparison>
-  <result>
-  Output only the name of the better plan, either **Plan1** or **Plan2**.
-  </result>

experiencemaker/op/summarizer/simple_summary_op.py

@@ -1,67 +1,73 @@
-from pathlib import Path
+import json
 from typing import List
 
 from loguru import logger
-from pydantic import Field
 
-from experiencemaker.enumeration.role import Role
-from experiencemaker.module.prompt.prompt_mixin import PromptMixin
-from experiencemaker.module.summarizer.base_summarizer import BaseSummarizer, SUMMARIZER_REGISTRY
-from experiencemaker.schema.experience import Experience
-from experiencemaker.schema.trajectory import Trajectory, Message, ActionMessage
-from experiencemaker.utils.util_function import get_html_match_content
+from experiencemaker.op import OP_REGISTRY
+from experiencemaker.op.base_op import BaseOp
+from experiencemaker.schema.experience import TextExperience, ExperienceMeta, BaseExperience
+from experiencemaker.schema.message import Message, Trajectory
+from experiencemaker.schema.request import SummarizerRequest
+from experiencemaker.schema.response import SummarizerResponse
+from experiencemaker.utils.op_utils import merge_messages_content
 
 
-@SUMMARIZER_REGISTRY.register("simple")
-class SimpleSummarizer(BaseSummarizer, PromptMixin):
-    max_retries: int = Field(default=5, description="max retries")
-    prompt_file_path: Path = Path(__file__).parent / "simple_summarizer_prompt.yaml"
+@OP_REGISTRY.register()
+class SimpleSummaryOp(BaseOp):
 
-    def _extract_trajectory_experience(self, trajectory: Trajectory, workspace_id: str) -> Experience | None:
-        step_content_collector: List[str] = []
+    def summary_trajectory(self, trajectory: Trajectory) -> List[BaseExperience]:
+        execution_process = merge_messages_content(trajectory.messages)
+        execution_result = "success" if trajectory.score > 0.9 else "fail"
+        summary_prompt = self.prompt_format(prompt_name="summary_prompt",
+                                            execution_process=execution_process,
+                                            execution_result=execution_result,
+                                            summary_example=self.get_prompt("summary_example"))
 
-        for step in trajectory.steps:
-            step_index = len(step_content_collector)
+        def parse_content(message: Message):
+            content = message.content
+            experience_list = []
+            try:
+                content = content.split("```")[1].strip()
+                if content.startswith("json"):
+                    content.strip("json")
 
-            if step.role is Role.ASSISTANT:
-                line = f"### step.{step_index} role={step.role.value} content=\n{step.content}\n{step.reasoning_content}\n"
-                if step.tool_calls:
-                    for tool_call in step.tool_calls:
-                        line += f" - tool call={tool_call.name}\n   params={tool_call.arguments}\n"
-                step_content_collector.append(line)
+                for exp_dict in json.loads(content):
+                    when_to_use = exp_dict.get("when_to_use", "").strip()
+                    experience = exp_dict.get("experience", "").strip()
+                    if when_to_use and experience:
+                        experience_list.append(TextExperience(workspace_id=self.context.request.workspace_id,
+                                                              when_to_use=when_to_use,
+                                                              content=experience,
+                                                              metadata=ExperienceMeta(author=self.llm.model_name)))
 
-            elif step.role is Role.USER:
-                line = f"### step.{step_index} role={step.role.value} content=\n{step.content}\n"
-                step_content_collector.append(line)
+                return experience_list
 
-            elif step.role is Role.TOOL:
-                line = f"### step.{step_index} role={step.role.value} tool call result=\n{step.content}\n"
-                step_content_collector.append(line)
+            except Exception as e:
+                logger.exception(f"parse content failed!\n{content}")
+                raise e
 
-        prompt = self.prompt_format(prompt_name="summary_prompt",
-                                    query=trajectory.query,
-                                    execution_process="\n".join(step_content_collector).strip(),
-                                    answer=trajectory.answer)
+        return self.llm.chat(messages=[Message(content=summary_prompt)], callback_fn=parse_content)
 
-        for i in range(self.max_retries):
-            action_message: ActionMessage = self.llm.chat(messages=[Message(content=prompt)])
-            experience_str = get_html_match_content(action_message.content, key="experience")
-            condition_str = get_html_match_content(action_message.content, key="condition")
-            if experience_str and condition_str:
-                return Experience(experience_workspace_id=workspace_id,
-                                  experience_role=self.llm.model_name,
-                                  experience_desc=condition_str,
-                                  experience_content=experience_str)
-            else:
-                logger.warning(f"action_message.content={action_message.content} re.search failed.")
+    def execute(self):
+        request: SummarizerRequest = self.context.request
+        for trajectory in request.traj_list:
+            execution_process = merge_messages_content(trajectory.messages)
+            execution_result = "success" if trajectory.score > 0.9 else "fail"
+            summary_prompt = self.prompt_format(prompt_name="summary_prompt",
+                                                execution_process=execution_process,
+                                                execution_result=execution_result,
+                                                summary_example=self.get_prompt("summary_example"))
+            self.submit_task(self.llm.chat, messages=[Message(content=summary_prompt)])
 
-        return None
+        experience_list: List[BaseExperience] = []
+        for task_result in self.join_task():
+            if task_result:
+                experience_list.extend(task_result)
 
-    def _extract_experiences(self, trajectories: List[Trajectory], workspace_id: str = None,
-                             **kwargs) -> List[Experience]:
-        experiences: List[Experience] = []
-        for trajectory in trajectories:
-            experience: Experience = self._extract_trajectory_experience(trajectory, workspace_id=workspace_id)
-            if experience:
-                experiences.append(experience)
-        return experiences
+        response: SummarizerResponse = self.context.response
+        response.experience_list = experience_list
+        for e in experience_list:
+            logger.info(f"add experience {e.when_to_use}\n{e.content}")
+
+        from experiencemaker.op.summarizer.insert_database_op import InsertDatabaseOp
+        self.context.set_context(InsertDatabaseOp.INSERT_NODES, [x.to_vector_node() for x in experience_list])

experiencemaker/op/summarizer/simple_summary_prompt.yaml

@@ -1,15 +1,9 @@
 summary_prompt: |
-  # Role
-  You are a helpful assistant.
-
-  # User Question
-  {query}
-
   # Execution Process
   {execution_process}
-
-  # Answer
-  {answer}
+  
+  # Result
+  {execution_result}
 
   # Task
   Reflect on what went well and what did not go well in the execution path based on the user's question.
@@ -17,22 +11,18 @@ summary_prompt: |
   They can include good suggestions or highlight areas to avoid and potential pitfalls.
   The type of experience can be plain text or a piece of code that solves a specific problem.
   If there are no experiences to summarize, output an empty list []. 
-  For each experience, first state the scenario when it is applicable (when to use), then provide the experience itself, with a maximum of three experiences summarized.
+  For each experience, first state the scenario when it is applicable (when to use), then provide the experience itself, with a maximum of two experiences summarized.
   
   # Output Format
   {summary_example}
 
 summary_example: |
-  ``` json
+  ```json
   [
     {
       "when_to_use": "...",
       "experience": "..."
     },
-    {
-      "when_to_use": "...",
-      "experience": "..."
-    },
     {
       "when_to_use": "...",
       "experience": "..."

experiencemaker/op/vector_store/delete_database_op.py

@@ -0,0 +1 @@
+# jinli

experiencemaker/op/vector_store/insert_database_op.py

@@ -0,0 +1,17 @@
+from typing import List
+
+from experiencemaker.op import OP_REGISTRY
+from experiencemaker.op.base_op import BaseOp
+from experiencemaker.schema.vector_node import VectorNode
+
+
+@OP_REGISTRY.register()
+class InsertDatabaseOp(BaseOp):
+    INSERT_NODES: str = "insert_nodes"
+
+    def execute(self):
+        nodes: List[VectorNode] = self.context.get_context(InsertDatabaseOp.INSERT_NODES)
+        self.vector_store.insert(nodes=nodes, workspace_id=self.context.request.workspace_id)
+
+
+

experiencemaker/op/vector_store/search_database_op.py

@@ -0,0 +1 @@
+# @jinli

experiencemaker/schema/app_config.py

@@ -26,9 +26,9 @@ class APIConfig:
 @dataclass
 class OpConfig:
     backend: str = field(default="")
-    llm: str = field(default="")
     prompt_file_path: str = field(default="")
     prompt_dict: dict = field(default_factory=dict)
+    llm: str = field(default="")
     embedding_model: str = field(default="")
     vector_store: str = field(default="")
     params: dict = field(default_factory=dict)

experiencemaker/service/experience_maker_service.py

@@ -7,8 +7,10 @@ from experiencemaker.config.config_parser import ConfigParser
 from experiencemaker.pipeline.pipeline import Pipeline
 from experiencemaker.pipeline.pipeline_context import PipelineContext
 from experiencemaker.schema.app_config import AppConfig, HttpServiceConfig
-from experiencemaker.schema.request import SummarizerRequest, RetrieverRequest, VectorStoreRequest, AgentRequest, BaseRequest
-from experiencemaker.schema.response import SummarizerResponse, RetrieverResponse, VectorStoreResponse, AgentResponse
+from experiencemaker.schema.request import SummarizerRequest, RetrieverRequest, VectorStoreRequest, AgentRequest, \
+    BaseRequest
+from experiencemaker.schema.response import SummarizerResponse, RetrieverResponse, VectorStoreResponse, AgentResponse, \
+    BaseResponse
 from experiencemaker.vector_store import VECTOR_STORE_REGISTRY
 
 
@@ -30,7 +32,7 @@ class ExperienceMakerService:
     def http_service_config(self) -> HttpServiceConfig:
         return self.init_app_config.http_service
 
-    def __call__(self, api: str, request: dict | BaseRequest) -> dict:
+    def __call__(self, api: str, request: dict | BaseRequest) -> BaseResponse:
         if isinstance(request, dict):
             request = BaseRequest(**request)
         app_config: AppConfig = self.config_parser.get_app_config(**request.config)
@@ -76,4 +78,4 @@ class ExperienceMakerService:
             response.success = False
             response.metadata["error"] = str(e)
 
-        return response.model_dump()
+        return response