Move hydrate_input to utils

Author: ajhai

llmstack/common/utils/utils.py

@@ -21,6 +21,7 @@
     run_sitemap_spider_in_process,
     run_url_spider_in_process,
 )
+from llmstack.common.utils.liquid import render_template
 
 logger = logging.getLogger(__name__)
 
@@ -515,3 +516,28 @@ def vectorize_text(text):
 
     vectors = embedding_functions.DefaultEmbeddingFunction()([text])
     return vectors[0]
+
+
+def hydrate_input(input, values):
+    def render(value):
+        if isinstance(value, str):
+            try:
+                return render_template(value, values)
+            except Exception:
+                logger.exception("Error rendering template when hydrating input")
+
+        return value
+
+    def traverse(obj):
+        if isinstance(obj, dict):
+            return {key: traverse(render(value)) for key, value in obj.items()}
+        elif isinstance(obj, list):
+            return [traverse(render(item)) for item in obj]
+        elif isinstance(obj, BaseModel):
+            cls = obj.__class__
+            return cls.model_validate(traverse(obj.model_dump()))
+        elif isinstance(obj, str):
+            return render(obj)
+        return obj
+
+    return traverse(input)

llmstack/processors/providers/api_processor_interface.py

@@ -16,8 +16,8 @@
     ProcessorInterface,
 )
 from llmstack.common.blocks.base.schema import BaseSchema as _Schema
-from llmstack.common.utils.liquid import render_template
 from llmstack.common.utils.provider_config import get_matched_provider_config
+from llmstack.common.utils.utils import hydrate_input
 from llmstack.play.actor import Actor, BookKeepingData
 from llmstack.play.actors.agent import ToolInvokeInput
 from llmstack.play.utils import extract_jinja2_variables
@@ -36,31 +36,6 @@
 FILE_WIDGET_NAME = "file"
 
 
-def hydrate_input(input, values):
-    def render(value):
-        if isinstance(value, str):
-            try:
-                return render_template(value, values)
-            except Exception:
-                logger.exception("Error rendering template when hydrating input")
-
-        return value
-
-    def traverse(obj):
-        if isinstance(obj, dict):
-            return {key: traverse(render(value)) for key, value in obj.items()}
-        elif isinstance(obj, list):
-            return [traverse(render(item)) for item in obj]
-        elif isinstance(obj, BaseModel):
-            cls = obj.__class__
-            return cls.model_validate(traverse(obj.model_dump()))
-        elif isinstance(obj, str):
-            return render(obj)
-        return obj
-
-    return traverse(input)
-
-
 class ApiProcessorSchema(_Schema):
     pass