This is an automated email from the ASF dual-hosted git repository. xtsong pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-agents.git
commit 6847575f86c1bc03e7606f9c00b4bb06c2ead660 Author: WenjinXie <[email protected]> AuthorDate: Thu Aug 28 11:55:44 2025 +0800 [hotfix] Use re rather than Formatter to format Prompt to support json schema in prompt. --- python/flink_agents/api/prompts/prompt.py | 12 +++++----- python/flink_agents/api/prompts/utils.py | 33 +++++++++++++++------------- python/flink_agents/api/tests/test_prompt.py | 6 ++++- 3 files changed, 28 insertions(+), 23 deletions(-) diff --git a/python/flink_agents/api/prompts/prompt.py b/python/flink_agents/api/prompts/prompt.py index 1635606..4ef07a7 100644 --- a/python/flink_agents/api/prompts/prompt.py +++ b/python/flink_agents/api/prompts/prompt.py @@ -18,7 +18,7 @@ from typing import List, Sequence, Union from flink_agents.api.chat_message import ChatMessage, MessageRole -from flink_agents.api.prompts.utils import FORMATTER +from flink_agents.api.prompts.utils import format_string from flink_agents.api.resource import ResourceType, SerializableResource @@ -51,11 +51,11 @@ class Prompt(SerializableResource): def format_string(self, **kwargs: str) -> str: """Generate text string from template with input arguments.""" if isinstance(self.template, str): - return FORMATTER.format(self.template, **kwargs) + return format_string(self.template, **kwargs) else: msgs = [] for m in self.template: - msg = f"{m.role.value}: {FORMATTER.format(m.content, **kwargs)}" + msg = f"{m.role.value}: {format_string(m.content, **kwargs)}" if m.extra_args is not None and len(m.extra_args) > 0: msg += f"{m.extra_args}" msgs.append(msg) @@ -67,15 +67,13 @@ class Prompt(SerializableResource): """Generate list of ChatMessage from template with input arguments.""" if isinstance(self.template, str): return [ - ChatMessage( - role=role, content=FORMATTER.format(self.template, **kwargs) - ) + ChatMessage(role=role, content=format_string(self.template, **kwargs)) ] else: msgs = [] for m in self.template: msg = ChatMessage( - role=m.role, content=FORMATTER.format(m.content, **kwargs) + role=m.role, content=format_string(m.content, **kwargs) ) msgs.append(msg) return msgs diff --git a/python/flink_agents/api/prompts/utils.py b/python/flink_agents/api/prompts/utils.py index 527fdd5..5becfa7 100644 --- a/python/flink_agents/api/prompts/utils.py +++ b/python/flink_agents/api/prompts/utils.py @@ -15,24 +15,27 @@ # See the License for the specific language governing permissions and # limitations under the License. ################################################################################# -from string import Formatter -from typing import Any +import re +from typing import Dict, Optional -from typing_extensions import override - -class SafeFormatter(Formatter): +class SafeFormatter: """Safe string formatter that does not raise KeyError if key is missing.""" - @override - def get_value(self, key: Any, args: Any, kwargs: Any) -> Any: - if isinstance(key, int): - return args[key] - else: - if key in kwargs: - return kwargs[key] - else: - return str(key) + def __init__(self, kwargs: Optional[Dict[str, str]] = None) -> None: + """Init method.""" + self.kwargs = kwargs or {} + + def format(self, text: str) -> str: + """Format a text with key arguments.""" + return re.sub(r"\{([^{}]+)\}", self._replace_match, text) + + def _replace_match(self, match: re.Match) -> str: + key = match.group(1) + return str(self.kwargs.get(key, match.group(0))) -FORMATTER = SafeFormatter() +def format_string(text: str, **kwargs: str) -> str: + """Format a string with kwargs.""" + formatter = SafeFormatter(kwargs=kwargs) + return formatter.format(text) diff --git a/python/flink_agents/api/tests/test_prompt.py b/python/flink_agents/api/tests/test_prompt.py index 40136a2..011a20a 100644 --- a/python/flink_agents/api/tests/test_prompt.py +++ b/python/flink_agents/api/tests/test_prompt.py @@ -118,6 +118,10 @@ def test_prompt_lack_one_argument(text_prompt: Prompt) -> None: # noqa: D103 review="The headphones broke after one week of use. Very poor quality", ) == ( "You ara a product review analyzer, please generate a score and the " - "dislike reasons(if any) for the review. The product 12345 is description, " + "dislike reasons(if any) for the review. The product 12345 is {description}, " "and user review is 'The headphones broke after one week of use. Very poor quality'." ) + +def test_prompt_contain_json_schema() -> None: # noqa: D103 + prompt = Prompt.from_text(name="prompt", text = f"The json schema is {Prompt.model_json_schema(mode='serialization')}") + prompt.format_string()
