prompty: adding Microsoft langchain_prompty package (#21346)

Co-authored-by: Micky Liu <wayliu@microsoft.com>
Co-authored-by: wayliums <wayliums@users.noreply.github.com>
Co-authored-by: Erick Friis <erick@langchain.dev>

libs/partners/prompty/langchain_prompty/utils.py

@@ -0,0 +1,207 @@
+import traceback
+from pathlib import Path
+from typing import Any, Dict, List, Union
+
+from .core import (
+    Frontmatter,
+    InvokerFactory,
+    ModelSettings,
+    Prompty,
+    PropertySettings,
+    SimpleModel,
+    TemplateSettings,
+    param_hoisting,
+)
+
+
+def load(prompt_path: str, configuration: str = "default") -> Prompty:
+    file_path = Path(prompt_path)
+    if not file_path.is_absolute():
+        # get caller's path (take into account trace frame)
+        caller = Path(traceback.extract_stack()[-3].filename)
+        file_path = Path(caller.parent / file_path).resolve().absolute()
+
+    # load dictionary from prompty file
+    matter = Frontmatter.read_file(file_path.__fspath__())
+    attributes = matter["attributes"]
+    content = matter["body"]
+
+    # normalize attribute dictionary resolve keys and files
+    attributes = Prompty.normalize(attributes, file_path.parent)
+
+    # load global configuration
+    if "model" not in attributes:
+        attributes["model"] = {}
+
+    # pull model settings out of attributes
+    try:
+        model = ModelSettings(**attributes.pop("model"))
+    except Exception as e:
+        raise ValueError(f"Error in model settings: {e}")
+
+    # pull template settings
+    try:
+        if "template" in attributes:
+            t = attributes.pop("template")
+            if isinstance(t, dict):
+                template = TemplateSettings(**t)
+            # has to be a string denoting the type
+            else:
+                template = TemplateSettings(type=t, parser="prompty")
+        else:
+            template = TemplateSettings(type="mustache", parser="prompty")
+    except Exception as e:
+        raise ValueError(f"Error in template loader: {e}")
+
+    # formalize inputs and outputs
+    if "inputs" in attributes:
+        try:
+            inputs = {
+                k: PropertySettings(**v) for (k, v) in attributes.pop("inputs").items()
+            }
+        except Exception as e:
+            raise ValueError(f"Error in inputs: {e}")
+    else:
+        inputs = {}
+    if "outputs" in attributes:
+        try:
+            outputs = {
+                k: PropertySettings(**v) for (k, v) in attributes.pop("outputs").items()
+            }
+        except Exception as e:
+            raise ValueError(f"Error in outputs: {e}")
+    else:
+        outputs = {}
+
+    # recursive loading of base prompty
+    if "base" in attributes:
+        # load the base prompty from the same directory as the current prompty
+        base = load(file_path.parent / attributes["base"])
+        # hoist the base prompty's attributes to the current prompty
+        model.api = base.model.api if model.api == "" else model.api
+        model.configuration = param_hoisting(
+            model.configuration, base.model.configuration
+        )
+        model.parameters = param_hoisting(model.parameters, base.model.parameters)
+        model.response = param_hoisting(model.response, base.model.response)
+        attributes["sample"] = param_hoisting(attributes, base.sample, "sample")
+
+        p = Prompty(
+            **attributes,
+            model=model,
+            inputs=inputs,
+            outputs=outputs,
+            template=template,
+            content=content,
+            file=file_path,
+            basePrompty=base,
+        )
+    else:
+        p = Prompty(
+            **attributes,
+            model=model,
+            inputs=inputs,
+            outputs=outputs,
+            template=template,
+            content=content,
+            file=file_path,
+        )
+    return p
+
+
+def prepare(
+    prompt: Prompty,
+    inputs: Dict[str, Any] = {},
+) -> Any:
+    invoker = InvokerFactory()
+
+    inputs = param_hoisting(inputs, prompt.sample)
+
+    if prompt.template.type == "NOOP":
+        render = prompt.content
+    else:
+        # render
+        result = invoker(
+            "renderer",
+            prompt.template.type,
+            prompt,
+            SimpleModel(item=inputs),
+        )
+        render = result.item
+
+    if prompt.template.parser == "NOOP":
+        result = render
+    else:
+        # parse
+        result = invoker(
+            "parser",
+            f"{prompt.template.parser}.{prompt.model.api}",
+            prompt,
+            SimpleModel(item=result.item),
+        )
+
+    if isinstance(result, SimpleModel):
+        return result.item
+    else:
+        return result
+
+
+def run(
+    prompt: Prompty,
+    content: Union[Dict, List, str],
+    configuration: Dict[str, Any] = {},
+    parameters: Dict[str, Any] = {},
+    raw: bool = False,
+) -> Any:
+    invoker = InvokerFactory()
+
+    if configuration != {}:
+        prompt.model.configuration = param_hoisting(
+            configuration, prompt.model.configuration
+        )
+
+    if parameters != {}:
+        prompt.model.parameters = param_hoisting(parameters, prompt.model.parameters)
+
+    # execute
+    result = invoker(
+        "executor",
+        prompt.model.configuration["type"],
+        prompt,
+        SimpleModel(item=content),
+    )
+
+    # skip?
+    if not raw:
+        # process
+        result = invoker(
+            "processor",
+            prompt.model.configuration["type"],
+            prompt,
+            result,
+        )
+
+    if isinstance(result, SimpleModel):
+        return result.item
+    else:
+        return result
+
+
+def execute(
+    prompt: Union[str, Prompty],
+    configuration: Dict[str, Any] = {},
+    parameters: Dict[str, Any] = {},
+    inputs: Dict[str, Any] = {},
+    raw: bool = False,
+    connection: str = "default",
+) -> Any:
+    if isinstance(prompt, str):
+        prompt = load(prompt, connection)
+
+    # prepare content
+    content = prepare(prompt, inputs)
+
+    # run LLM model
+    result = run(prompt, content, configuration, parameters, raw)
+
+    return result