use template to render all prompts

Author: peteryangms

rdagent/app/qlib_rd_loop/factor_from_report.py

@@ -3,27 +3,23 @@
 from typing import Any, Dict, Tuple
 
 import fire
-from jinja2 import Environment, StrictUndefined
 
 from rdagent.app.qlib_rd_loop.conf import FACTOR_FROM_REPORT_PROP_SETTING
 from rdagent.app.qlib_rd_loop.factor import FactorRDLoop
 from rdagent.components.document_reader.document_reader import (
     extract_first_page_screenshot_from_pdf,
     load_and_process_pdfs_by_langchain,
 )
-from rdagent.core.prompts import Prompts
 from rdagent.core.proposal import Hypothesis
 from rdagent.log import rdagent_logger as logger
 from rdagent.oai.llm_utils import APIBackend
 from rdagent.scenarios.qlib.experiment.factor_experiment import QlibFactorExperiment
 from rdagent.scenarios.qlib.factor_experiment_loader.pdf_loader import (
     FactorExperimentLoaderFromPDFfiles,
 )
+from rdagent.utils.agent.tpl import T
 from rdagent.utils.workflow import LoopMeta
 
-prompts_path = Path(__file__).parent / "prompts.yaml"
-prompts = Prompts(file_path=prompts_path)
-
 
 def generate_hypothesis(factor_result: dict, report_content: str) -> str:
     """
@@ -36,13 +32,9 @@ def generate_hypothesis(factor_result: dict, report_content: str) -> str:
     Returns:
         str: The generated hypothesis.
     """
-    system_prompt = (
-        Environment(undefined=StrictUndefined).from_string(prompts["hypothesis_generation"]["system"]).render()
-    )
-    user_prompt = (
-        Environment(undefined=StrictUndefined)
-        .from_string(prompts["hypothesis_generation"]["user"])
-        .render(factor_descriptions=json.dumps(factor_result), report_content=report_content)
+    system_prompt = T(".prompts:hypothesis_generation.system").r()
+    user_prompt = T(".prompts:hypothesis_generation.user").r(
+        factor_descriptions=json.dumps(factor_result), report_content=report_content
     )
 
     response = APIBackend().build_messages_and_create_chat_completion(

rdagent/components/coder/CoSTEER/evolving_strategy.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from abc import abstractmethod
-from pathlib import Path
 
 from rdagent.components.coder.CoSTEER.config import CoSTEERSettings
 from rdagent.components.coder.CoSTEER.evaluators import (
@@ -15,12 +14,9 @@
 from rdagent.core.conf import RD_AGENT_SETTINGS
 from rdagent.core.evolving_framework import EvolvingStrategy, EvoStep, QueriedKnowledge
 from rdagent.core.experiment import FBWorkspace, Task
-from rdagent.core.prompts import Prompts
 from rdagent.core.scenario import Scenario
 from rdagent.core.utils import multiprocessing_wrapper
 
-implement_prompts = Prompts(file_path=Path(__file__).parent / "prompts.yaml")
-
 
 class MultiProcessEvolvingStrategy(EvolvingStrategy):
     def __init__(self, scen: Scenario, settings: CoSTEERSettings):

rdagent/components/coder/CoSTEER/knowledge_management.py

@@ -8,8 +8,6 @@
 from pathlib import Path
 from typing import List, Union
 
-from jinja2 import Environment, StrictUndefined
-
 from rdagent.components.coder.CoSTEER.config import CoSTEERSettings
 from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
 from rdagent.components.knowledge_management.graph import (
@@ -26,12 +24,12 @@
     RAGStrategy,
 )
 from rdagent.core.experiment import FBWorkspace, Task
-from rdagent.core.prompts import Prompts
 from rdagent.log import rdagent_logger as logger
 from rdagent.oai.llm_utils import (
     APIBackend,
     calculate_embedding_distance_between_str_list,
 )
+from rdagent.utils.agent.tpl import T
 
 
 class CoSTEERKnowledge(Knowledge):
@@ -216,8 +214,6 @@ def __init__(
 
 
 class CoSTEERRAGStrategyV2(RAGStrategy):
-    prompt = Prompts(file_path=Path(__file__).parent / "prompts.yaml")
-
     def __init__(self, knowledgebase: CoSTEERKnowledgeBaseV2, settings: CoSTEERSettings) -> None:
         super().__init__(knowledgebase)
         self.current_generated_trace_count = 0
@@ -324,12 +320,8 @@ def analyze_component(
         all_component_content = ""
         for _, component_node in enumerate(all_component_nodes):
             all_component_content += f"{component_node.content}, \n"
-        analyze_component_system_prompt = (
-            Environment(undefined=StrictUndefined)
-            .from_string(self.prompt["analyze_component_prompt_v1_system"])
-            .render(
-                all_component_content=all_component_content,
-            )
+        analyze_component_system_prompt = T(".prompts:analyze_component_prompt_v1_system").r(
+            all_component_content=all_component_content,
         )
 
         analyze_component_user_prompt = target_task_information

rdagent/components/coder/data_science/ensemble/__init__.py

@@ -11,9 +11,7 @@
     - Each coder could be tested.
 """
 
-import json
 from pathlib import Path
-from typing import Dict
 
 from jinja2 import Environment, StrictUndefined
 

rdagent/components/coder/factor_coder/eva_utils.py

@@ -1,20 +1,16 @@
 import io
 import json
 from abc import abstractmethod
-from pathlib import Path
 from typing import Dict, Tuple
 
 import pandas as pd
-from jinja2 import Environment, StrictUndefined
 
 from rdagent.components.coder.factor_coder.config import FACTOR_COSTEER_SETTINGS
 from rdagent.components.coder.factor_coder.factor import FactorTask
 from rdagent.core.experiment import Task, Workspace
-from rdagent.core.prompts import Prompts
 from rdagent.oai.llm_conf import LLM_SETTINGS
 from rdagent.oai.llm_utils import APIBackend
-
-evaluate_prompts = Prompts(file_path=Path(__file__).parent / "prompts.yaml")
+from rdagent.utils.agent.tpl import T
 
 
 class FactorEvaluator:
@@ -81,36 +77,26 @@ def evaluate(
         factor_information = target_task.get_task_information()
         code = implementation.all_codes
 
-        system_prompt = (
-            Environment(undefined=StrictUndefined)
-            .from_string(evaluate_prompts["evaluator_code_feedback_v1_system"])
-            .render(
-                scenario=(
-                    self.scen.get_scenario_all_desc(
-                        target_task,
-                        filtered_tag="feature",
-                        simple_background=FACTOR_COSTEER_SETTINGS.simple_background,
-                    )
-                    if self.scen is not None
-                    else "No scenario description."
+        system_prompt = T(".prompts:evaluator_code_feedback_v1_system").r(
+            scenario=(
+                self.scen.get_scenario_all_desc(
+                    target_task,
+                    filtered_tag="feature",
+                    simple_background=FACTOR_COSTEER_SETTINGS.simple_background,
                 )
+                if self.scen is not None
+                else "No scenario description."
             )
         )
 
         execution_feedback_to_render = execution_feedback
         for _ in range(10):  # 10 times to split the content is enough
-            user_prompt = (
-                Environment(undefined=StrictUndefined)
-                .from_string(
-                    evaluate_prompts["evaluator_code_feedback_v1_user"],
-                )
-                .render(
-                    factor_information=factor_information,
-                    code=code,
-                    execution_feedback=execution_feedback_to_render,
-                    value_feedback=value_feedback,
-                    gt_code=gt_implementation.code if gt_implementation else None,
-                )
+            user_prompt = T(".prompts:evaluator_code_feedback_v1_user").r(
+                factor_information=factor_information,
+                code=code,
+                execution_feedback=execution_feedback_to_render,
+                value_feedback=value_feedback,
+                gt_code=gt_implementation.code if gt_implementation else None,
             )
             if (
                 APIBackend().build_messages_and_calculate_token(
@@ -189,17 +175,11 @@ def evaluate(
         buffer = io.StringIO()
         gen_df.info(buf=buffer)
         gen_df_info_str = f"The user is currently working on a feature related task.\nThe output dataframe info is:\n{buffer.getvalue()}"
-        system_prompt = (
-            Environment(undefined=StrictUndefined)
-            .from_string(
-                evaluate_prompts["evaluator_output_format_system"],
-            )
-            .render(
-                scenario=(
-                    self.scen.get_scenario_all_desc(implementation.target_task, filtered_tag="feature")
-                    if self.scen is not None
-                    else "No scenario description."
-                )
+        system_prompt = T(".prompts:evaluator_output_format_system").r(
+            scenario=(
+                self.scen.get_scenario_all_desc(implementation.target_task, filtered_tag="feature")
+                if self.scen is not None
+                else "No scenario description."
             )
         )
 
@@ -504,35 +484,25 @@ def evaluate(
         code_feedback: str,
         **kwargs,
     ) -> Tuple:
-        system_prompt = (
-            Environment(undefined=StrictUndefined)
-            .from_string(evaluate_prompts["evaluator_final_decision_v1_system"])
-            .render(
-                scenario=(
-                    self.scen.get_scenario_all_desc(target_task, filtered_tag="feature")
-                    if self.scen is not None
-                    else "No scenario description."
-                )
+        system_prompt = T(".prompts:evaluator_final_decision_v1_system").r(
+            scenario=(
+                self.scen.get_scenario_all_desc(target_task, filtered_tag="feature")
+                if self.scen is not None
+                else "No scenario description."
             )
         )
         execution_feedback_to_render = execution_feedback
 
         for _ in range(10):  # 10 times to split the content is enough
-            user_prompt = (
-                Environment(undefined=StrictUndefined)
-                .from_string(
-                    evaluate_prompts["evaluator_final_decision_v1_user"],
-                )
-                .render(
-                    factor_information=target_task.get_task_information(),
-                    execution_feedback=execution_feedback_to_render,
-                    code_feedback=code_feedback,
-                    value_feedback=(
-                        value_feedback
-                        if value_feedback is not None
-                        else "No Ground Truth Value provided, so no evaluation on value is performed."
-                    ),
-                )
+            user_prompt = T(".prompts:evaluator_final_decision_v1_user").r(
+                factor_information=target_task.get_task_information(),
+                execution_feedback=execution_feedback_to_render,
+                code_feedback=code_feedback,
+                value_feedback=(
+                    value_feedback
+                    if value_feedback is not None
+                    else "No Ground Truth Value provided, so no evaluation on value is performed."
+                ),
             )
             if (
                 APIBackend().build_messages_and_calculate_token(

rdagent/components/coder/factor_coder/evolving_strategy.py

@@ -1,11 +1,8 @@
 from __future__ import annotations
 
 import json
-from pathlib import Path
 from typing import Dict
 
-from jinja2 import Environment, StrictUndefined
-
 from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
 from rdagent.components.coder.CoSTEER.evolving_strategy import (
     MultiProcessEvolvingStrategy,
@@ -17,11 +14,9 @@
 from rdagent.components.coder.factor_coder.config import FACTOR_COSTEER_SETTINGS
 from rdagent.components.coder.factor_coder.factor import FactorFBWorkspace, FactorTask
 from rdagent.core.experiment import FBWorkspace
-from rdagent.core.prompts import Prompts
 from rdagent.oai.llm_conf import LLM_SETTINGS
 from rdagent.oai.llm_utils import APIBackend
-
-implement_prompts = Prompts(file_path=Path(__file__).parent / "prompts.yaml")
+from rdagent.utils.agent.tpl import T
 
 
 class FactorMultiProcessEvolvingStrategy(MultiProcessEvolvingStrategy):
@@ -36,24 +31,14 @@ def error_summary(
         queried_former_failed_knowledge_to_render: list,
         queried_similar_error_knowledge_to_render: list,
     ) -> str:
-        error_summary_system_prompt = (
-            Environment(undefined=StrictUndefined)
-            .from_string(implement_prompts["evolving_strategy_error_summary_v2_system"])
-            .render(
-                scenario=self.scen.get_scenario_all_desc(target_task),
-                factor_information_str=target_task.get_task_information(),
-                code_and_feedback=queried_former_failed_knowledge_to_render[-1].get_implementation_and_feedback_str(),
-            )
-            .strip("\n")
+        error_summary_system_prompt = T(".prompts:evolving_strategy_error_summary_v2_system").r(
+            scenario=self.scen.get_scenario_all_desc(target_task),
+            factor_information_str=target_task.get_task_information(),
+            code_and_feedback=queried_former_failed_knowledge_to_render[-1].get_implementation_and_feedback_str(),
         )
         for _ in range(10):  # max attempt to reduce the length of error_summary_user_prompt
-            error_summary_user_prompt = (
-                Environment(undefined=StrictUndefined)
-                .from_string(implement_prompts["evolving_strategy_error_summary_v2_user"])
-                .render(
-                    queried_similar_error_knowledge=queried_similar_error_knowledge_to_render,
-                )
-                .strip("\n")
+            error_summary_user_prompt = T(".prompts:evolving_strategy_error_summary_v2_user").r(
+                queried_similar_error_knowledge=queried_similar_error_knowledge_to_render,
             )
             if (
                 APIBackend().build_messages_and_calculate_token(
@@ -106,16 +91,9 @@ def implement_one_task(
         latest_attempt_to_latest_successful_execution = queried_knowledge.task_to_former_failed_traces[
             target_factor_task_information
         ][1]
-
-        system_prompt = (
-            Environment(undefined=StrictUndefined)
-            .from_string(
-                implement_prompts["evolving_strategy_factor_implementation_v1_system"],
-            )
-            .render(
-                scenario=self.scen.get_scenario_all_desc(target_task, filtered_tag="feature"),
-                queried_former_failed_knowledge=queried_former_failed_knowledge_to_render,
-            )
+        system_prompt = T(".prompts:evolving_strategy_factor_implementation_v1_system").r(
+            scenario=self.scen.get_scenario_all_desc(target_task, filtered_tag="feature"),
+            queried_former_failed_knowledge=queried_former_failed_knowledge_to_render,
         )
         queried_similar_successful_knowledge_to_render = queried_similar_successful_knowledge
         queried_similar_error_knowledge_to_render = queried_similar_error_knowledge
@@ -136,19 +114,12 @@ def implement_one_task(
             else:
                 error_summary_critics = None
             # 构建user_prompt。开始写代码
-            user_prompt = (
-                Environment(undefined=StrictUndefined)
-                .from_string(
-                    implement_prompts["evolving_strategy_factor_implementation_v2_user"],
-                )
-                .render(
-                    factor_information_str=target_factor_task_information,
-                    queried_similar_successful_knowledge=queried_similar_successful_knowledge_to_render,
-                    queried_similar_error_knowledge=queried_similar_error_knowledge_to_render,
-                    error_summary_critics=error_summary_critics,
-                    latest_attempt_to_latest_successful_execution=latest_attempt_to_latest_successful_execution,
-                )
-                .strip("\n")
+            user_prompt = T(".prompts:evolving_strategy_factor_implementation_v2_user").r(
+                factor_information_str=target_factor_task_information,
+                queried_similar_successful_knowledge=queried_similar_successful_knowledge_to_render,
+                queried_similar_error_knowledge=queried_similar_error_knowledge_to_render,
+                error_summary_critics=error_summary_critics,
+                latest_attempt_to_latest_successful_execution=latest_attempt_to_latest_successful_execution,
             )
             if (
                 APIBackend().build_messages_and_calculate_token(user_prompt=user_prompt, system_prompt=system_prompt)