new: tools can now be directly called from the prompt by interpolation

Author: evilsocket

examples/changelog/agent.yml

@@ -3,12 +3,15 @@ agent: >
   
   ## Guidance
 
-  - Start by using the get_new_commits tool to get a list of commits since the last release.
-  - Then use the create_changelog tool to generate a changelog for the new commits.
-  - Focus on the new features and fixes, group and summarize other changes into "Miscellaneous" or "Other".
+  - Use the create_changelog tool to generate a changelog for the new commits.
+  - Focus on the new features and major fixes.
+  - Group and summarize other minor changes into "Miscellaneous" or "Other".
   - Add relevant and catchy emojis but ONLY to important changes.
 
-task: Generate a changelog for the current project.
+task: >
+  Generate a changelog for these commits in the current project:
+
+  {{ get_new_commits() }}
 
 defaults:
   output: CHANGELOG.md

nerve/generation/test_conversation.py nerve/generation/conversation_test.pynerve/generation/test_conversation.py renamed to nerve/generation/conversation_test.py

@@ -64,8 +64,6 @@ class Configuration(BaseModel):
     # legacy field used to detect if the user is loading a legacy file
     system_prompt: str | None = Field(default=None, exclude=True)
 
-    # TODO: add a "precall" to call the tool when the agent is started
-
     # TODO: document these fields.
 
     # used for versioning the agents

nerve/models.py

@@ -4,6 +4,7 @@
 from loguru import logger
 
 from nerve.models import Tool
+from nerve.runtime import state
 from nerve.tools import compiler
 
 
@@ -62,6 +63,6 @@ def build(
             logger.debug(f"🧰 importing {len(funcs)} custom tools from functions")
             runtime.tools.extend(funcs)
 
-        logger.debug(f"tools: {runtime.tools}")
+        state.set_tools({tool.__name__: tool for tool in runtime.tools})
 
         return runtime

nerve/test_models.py nerve/models_test.pynerve/test_models.py renamed to nerve/models_test.py

@@ -1,6 +1,8 @@
+import asyncio
 import json
 import os
 import pathlib
+import threading
 import typing as t
 
 import click
@@ -28,6 +30,8 @@
 _defaults: dict[str, t.Any] = {}
 # similar to variables but used by tools
 _knowledge: dict[str, t.Any] = {}
+# tools
+_tools: dict[str, t.Callable[..., t.Any]] = {}
 # extra tools defined at runtime
 _extra_tools: dict[str, t.Callable[..., t.Any]] = {}
 # listeners for events
@@ -222,6 +226,15 @@ def as_dict() -> dict[str, t.Any]:
     }
 
 
+def set_tools(tools: dict[str, t.Callable[..., t.Any]]) -> None:
+    """Set all tools."""
+
+    global _tools
+    _tools = tools
+
+    logger.debug(f"tools: {_tools}")
+
+
 def get_extra_tools() -> dict[str, t.Callable[..., t.Any]]:
     """Get any extra tool registered at runtime."""
 
@@ -329,9 +342,8 @@ def on_user_input_needed(input_name: str, prompt: str) -> str:
         )
 
 
-def interpolate(raw: str, extra: dict[str, t.Any] | None = None) -> str:
-    """Interpolate the current state into a string."""
-
+def _create_jinja_env() -> jinja2.Environment:
+    # we use this to catch undefined variables at runtime
     class OnUndefinedVariable(jinja2.Undefined):
         def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
             super().__init__(*args, **kwargs)
@@ -352,5 +364,46 @@ def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
         def __str__(self) -> str:
             return self.value or "<UNDEFINED>"
 
+    env = jinja2.Environment(undefined=OnUndefinedVariable)
+
+    # allow prompts to call tools
+    for name, tool_fn in _tools.items():
+        # if the tool is async, wrap it in a sync function to make it callable from Jinja
+        if asyncio.iscoroutinefunction(tool_fn):
+
+            def make_sync_wrapper(fn: t.Callable[..., t.Any]) -> t.Callable[..., t.Any]:
+                def sync_wrapper(*args: t.Any, **kwargs: t.Any) -> t.Any:
+                    coro = fn(*args, **kwargs)
+                    # NOTE: we use a list to store the result because nonlocal variables (like scalars) inside a nested
+                    # function do not allow assignment unless explicitly declared nonlocal. However, mutable objects like
+                    # lists can be modified within the nested function without extra declarations.
+                    result_container = []
+
+                    def we_need_an_async_loop_thread() -> None:
+                        loop = asyncio.new_event_loop()
+                        asyncio.set_event_loop(loop)
+                        result_container.append(loop.run_until_complete(coro))
+
+                    thread = threading.Thread(target=we_need_an_async_loop_thread)
+                    thread.start()
+                    thread.join()
+
+                    return result_container[0]
+
+                return sync_wrapper
+
+            env.globals[name] = make_sync_wrapper(tool_fn)
+        else:
+            env.globals[name] = tool_fn
+
+    return env
+
+
+def interpolate(raw: str, extra: dict[str, t.Any] | None = None) -> str:
+    """Interpolate the current state into a string."""
+
+    env = _create_jinja_env()
+    template = env.from_string(raw)
     context = _variables | (extra or {})
-    return jinja2.Environment(undefined=OnUndefinedVariable).from_string(raw).render(**context)
+
+    return template.render(**context)

nerve/runtime/__init__.py

@@ -0,0 +1,105 @@
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from nerve.models import Mode
+from nerve.runtime import state
+
+
+class TestInterpolate:
+    def setup_method(self) -> None:
+        # Reset state before each test
+        state.reset()
+        state._variables = {}
+        state._defaults = {}
+        state._mode = Mode.AUTOMATIC
+        state._tools = {}
+
+    def test_basic_interpolation(self) -> None:
+        # Setup
+        state.update_variables({"name": "John", "age": 30})
+
+        # Test
+        result = state.interpolate("Hello, {{ name }}! You are {{ age }} years old.")
+
+        # Verify
+        assert result == "Hello, John! You are 30 years old."
+
+    def test_interpolation_with_extra_context(self) -> None:
+        # Setup
+        state.update_variables({"name": "John"})
+        extra = {"location": "New York"}
+
+        # Test
+        result = state.interpolate("{{ name }} is from {{ location }}.", extra)
+
+        # Verify
+        assert result == "John is from New York."
+
+    def test_undefined_variable_in_automatic_mode(self) -> None:
+        # Setup
+        state._mode = Mode.AUTOMATIC
+        state._defaults = {"city": "San Francisco"}
+
+        # Test
+        result = state.interpolate("Welcome to {{ city }}!")
+
+        # Verify
+        assert result == "Welcome to San Francisco!"
+
+    def test_undefined_variable_from_environment(self) -> None:
+        # Setup
+        with patch.dict(os.environ, {"TEST_VAR": "test_value"}):
+            # Test
+            result = state.interpolate("Environment value: {{ TEST_VAR }}")
+
+            # Verify
+            assert result == "Environment value: test_value"
+            # The variable should also be saved to state
+            assert state._variables.get("TEST_VAR") == "test_value"
+
+    @patch("builtins.input", return_value="Paris")
+    def test_undefined_variable_in_interactive_mode(self, mock_input: MagicMock) -> None:
+        # Setup
+        state._mode = Mode.INTERACTIVE
+
+        # Test
+        result = state.interpolate("Welcome to {{ city }}!")
+
+        # Verify
+        assert result == "Welcome to Paris!"
+        assert mock_input.called
+        assert state._variables.get("city") == "Paris"
+
+    def test_tool_call_in_template(self) -> None:
+        # Setup
+        mock_tool = MagicMock(return_value="tool result")
+        state._tools = {"test_tool": mock_tool}
+
+        # Test
+        result = state.interpolate("Tool output: {{ test_tool() }}")
+
+        # Verify
+        assert result == "Tool output: tool result"
+        assert mock_tool.called
+
+    @patch("nerve.runtime.state.on_user_input_needed")
+    def test_missing_variable_raises_exception(self, mock_input_needed: MagicMock) -> None:
+        # Setup
+        state._mode = Mode.AUTOMATIC
+        mock_input_needed.side_effect = Exception("Missing parameter")
+
+        # Test & Verify
+        with pytest.raises(Exception):  # noqa: B017
+            state.interpolate("Missing {{ variable }}")
+
+    def test_complex_expression(self) -> None:
+        # Setup
+        state.update_variables({"numbers": [1, 2, 3, 4, 5]})
+
+        # Test
+        result = state.interpolate("Sum: {{ numbers|sum }}, Average: {{ (numbers|sum) / numbers|length }}")
+
+        # Verify
+        assert result == "Sum: 15, Average: 3.0"