diff --git a/experiencemaker/op/react/__init__.py b/cookbook/react/__init__.py similarity index 100% rename from experiencemaker/op/react/__init__.py rename to cookbook/react/__init__.py diff --git a/experiencemaker/op/react/agent.py b/cookbook/react/agent.py similarity index 100% rename from experiencemaker/op/react/agent.py rename to cookbook/react/agent.py diff --git a/experiencemaker/op/react/simple_agent.py b/cookbook/react/simple_agent.py similarity index 100% rename from experiencemaker/op/react/simple_agent.py rename to cookbook/react/simple_agent.py diff --git a/experiencemaker/op/react/simple_agent_prompt.yaml b/cookbook/react/simple_agent_prompt.yaml similarity index 100% rename from experiencemaker/op/react/simple_agent_prompt.yaml rename to cookbook/react/simple_agent_prompt.yaml diff --git a/experiencemaker/llm/base_llm.py b/experiencemaker/llm/base_llm.py index 8489f819..b45a8c98 100644 --- a/experiencemaker/llm/base_llm.py +++ b/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}") diff --git a/experiencemaker/op/base_op.py b/experiencemaker/op/base_op.py index aba4b263..94b5330d 100644 --- a/experiencemaker/op/base_op.py +++ b/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: diff --git a/experiencemaker/op/mock_op.py b/experiencemaker/op/mock_op.py index 7f6a05d5..32afbaf6 100644 --- a/experiencemaker/op/mock_op.py +++ b/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() diff --git a/experiencemaker/op/prompt_mixin.py b/experiencemaker/op/prompt_mixin.py index 41e6ae65..b933fc8d 100644 --- a/experiencemaker/op/prompt_mixin.py +++ b/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] diff --git a/experiencemaker/op/retriever/build_query_op.py b/experiencemaker/op/retriever/build_query_op.py index 0f903ddd..5a4c74b8 100644 --- a/experiencemaker/op/retriever/build_query_op.py +++ b/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: diff --git a/experiencemaker/op/retriever/rerank_experience_op.py b/experiencemaker/op/retriever/rerank_experience_op.py new file mode 100644 index 00000000..c98c96bc --- /dev/null +++ b/experiencemaker/op/retriever/rerank_experience_op.py @@ -0,0 +1 @@ +# at jiaji \ No newline at end of file diff --git a/experiencemaker/op/retriever/rewrite_experience_op.py b/experiencemaker/op/retriever/rewrite_experience_op.py new file mode 100644 index 00000000..122c231b --- /dev/null +++ b/experiencemaker/op/retriever/rewrite_experience_op.py @@ -0,0 +1 @@ +# @jiaji \ No newline at end of file diff --git a/experiencemaker/op/summarizer/compare_summary_op.py b/experiencemaker/op/summarizer/compare_summary_op.py new file mode 100644 index 00000000..e69de29b diff --git a/experiencemaker/op/summarizer/extract_experience_op.py b/experiencemaker/op/summarizer/extract_experience_op.py new file mode 100644 index 00000000..e69de29b diff --git a/experiencemaker/op/summarizer/simple_compare_reward_fn.py b/experiencemaker/op/summarizer/simple_compare_reward_fn.py deleted file mode 100644 index b06339c9..00000000 --- a/experiencemaker/op/summarizer/simple_compare_reward_fn.py +++ /dev/null @@ -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) diff --git a/experiencemaker/op/summarizer/simple_compare_reward_fn_prompt.yaml b/experiencemaker/op/summarizer/simple_compare_reward_fn_prompt.yaml deleted file mode 100644 index 24e57a3f..00000000 --- a/experiencemaker/op/summarizer/simple_compare_reward_fn_prompt.yaml +++ /dev/null @@ -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 - - List the rules that could be used for comparison... - - - Conduct a step-by-step comparison according to the rules... - - - Output only the name of the better plan, either **Plan1** or **Plan2**. - \ No newline at end of file diff --git a/experiencemaker/op/summarizer/simple_summary_op.py b/experiencemaker/op/summarizer/simple_summary_op.py index 761b4e59..4451fab1 100644 --- a/experiencemaker/op/summarizer/simple_summary_op.py +++ b/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]) diff --git a/experiencemaker/op/summarizer/simple_summary_prompt.yaml b/experiencemaker/op/summarizer/simple_summary_prompt.yaml index 1cdb5427..bb20a51e 100644 --- a/experiencemaker/op/summarizer/simple_summary_prompt.yaml +++ b/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": "..." diff --git a/experiencemaker/op/summarizer/update_experience_op.py b/experiencemaker/op/summarizer/update_experience_op.py new file mode 100644 index 00000000..e69de29b diff --git a/experiencemaker/op/summarizer/validate_experience_op.py b/experiencemaker/op/summarizer/validate_experience_op.py new file mode 100644 index 00000000..e69de29b diff --git a/experiencemaker/op/vector_store/__init__.py b/experiencemaker/op/vector_store/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/experiencemaker/op/vector_store/delete_database_op.py b/experiencemaker/op/vector_store/delete_database_op.py new file mode 100644 index 00000000..3b4780ba --- /dev/null +++ b/experiencemaker/op/vector_store/delete_database_op.py @@ -0,0 +1 @@ +# jinli \ No newline at end of file diff --git a/experiencemaker/op/vector_store/insert_database_op.py b/experiencemaker/op/vector_store/insert_database_op.py new file mode 100644 index 00000000..6cbe7c5d --- /dev/null +++ b/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) + + + diff --git a/experiencemaker/op/vector_store/search_database_op.py b/experiencemaker/op/vector_store/search_database_op.py new file mode 100644 index 00000000..c8022666 --- /dev/null +++ b/experiencemaker/op/vector_store/search_database_op.py @@ -0,0 +1 @@ +# @jinli \ No newline at end of file diff --git a/experiencemaker/schema/app_config.py b/experiencemaker/schema/app_config.py index d50d7884..407431c6 100644 --- a/experiencemaker/schema/app_config.py +++ b/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) diff --git a/experiencemaker/service/experience_maker_service.py b/experiencemaker/service/experience_maker_service.py index aa86a122..dd2e8b79 100644 --- a/experiencemaker/service/experience_maker_service.py +++ b/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