RDBC-1151 RavenDB-24824 Add per-call output schema override and string output for AI conversations

ravendb/documents/ai/ai_conversation.py

@@ -150,14 +150,18 @@ def run(self) -> AiAnswer:
             if self._handle_server_reply(r):
                 return r
 
-    def run_with_schema(self, output_options: "AiOutputOptions") -> AiAnswer:
+    def run_with_schema(self, output_options: Union["AiOutputOptions", str, Any]) -> AiAnswer:
         """
         Runs one turn with the output format overridden for that turn only, leaving the
-        agent's own schema in place for later turns. Pass
-        ``AiOutputOptions(no_schema=True)`` to get free-form text back instead of JSON.
+        agent's own schema in place for later turns.
+
+        ``output_options`` is one of:
+
+        - an ``AiOutputOptions`` instance; ``AiOutputOptions(no_schema=True)`` gets free-form text back
+        - a JSON schema string, sent as the explicit output schema
+        - a sample object (dict or entity), which the server converts to a JSON schema
         """
-        if output_options is None:
-            raise ValueError("output_options cannot be None")
+        output_options = self._to_output_options(output_options)
 
         self._dispatched_tool_ids.clear()
 
@@ -166,6 +170,12 @@ def run_with_schema(self, output_options: "AiOutputOptions") -> AiAnswer:
             if self._handle_server_reply(r):
                 return r
 
+    def run_text(self) -> AiAnswer:
+        """
+        Runs one turn without structured output. The answer is the model's free-form text, as a string.
+        """
+        return self.run_with_schema(AiOutputOptions(no_schema=True))
+
     def stream(self, stream_property_path: str = None, on_chunk: Optional[Callable[[str], None]] = None) -> AiAnswer:
         while True:
             r = self._run_internal(stream_property_path=stream_property_path, streamed_chunks_callback=on_chunk)
@@ -176,15 +186,18 @@ def stream_with_schema(
         self,
         stream_property_path: str = None,
         on_chunk: Optional[Callable[[str], None]] = None,
-        output_options: "AiOutputOptions" = None,
+        output_options: Union["AiOutputOptions", str, Any] = None,
     ) -> AiAnswer:
         """
         Streams one turn with the output format overridden for that turn only.
+        ``output_options`` accepts the same values as in ``run_with_schema``.
         ``stream_property_path`` is ignored when the options ask for no schema, since
         free-form text has no property to stream from.
         """
-        if output_options is None:
-            raise ValueError("output_options cannot be None")
+        output_options = self._to_output_options(output_options)
+
+        if output_options.no_schema and stream_property_path is None:
+            stream_property_path = ""
 
         while True:
             r = self._run_internal(
@@ -195,6 +208,25 @@ def stream_with_schema(
             if self._handle_server_reply(r):
                 return r
 
+    def stream_text(self, on_chunk: Callable[[str], None]) -> AiAnswer:
+        """
+        Streams one turn without structured output: ``on_chunk`` receives each text chunk as it
+        arrives, and the answer is the full free-form text, as a string.
+        """
+        if on_chunk is None:
+            raise ValueError("on_chunk cannot be None")
+        return self.stream_with_schema("", on_chunk, AiOutputOptions(no_schema=True))
+
+    @staticmethod
+    def _to_output_options(value: Union["AiOutputOptions", str, Any]) -> "AiOutputOptions":
+        if value is None:
+            raise ValueError("output_options cannot be None")
+        if isinstance(value, AiOutputOptions):
+            return value
+        if isinstance(value, str):
+            return AiOutputOptions(output_schema=value)
+        return AiOutputOptions(sample_object=value)
+
     def _run_internal(
         self,
         stream_property_path: Optional[str] = None,
@@ -340,6 +372,36 @@ def receive(
         t = self.AiActionContext(self, lambda request, args: action(request, args), ai_handle_error)
         self._add_action(action_name, t.execute)
 
+    def handle_no_args(
+        self,
+        action_name: str,
+        action: Callable[[], Any],
+        ai_handle_error: AiHandleErrorStrategy = AiHandleErrorStrategy.SEND_ERRORS_TO_MODEL,
+    ) -> None:
+        """
+        Registers a handler for an action tool that takes no parameters. The handler is called
+        without arguments and its return value is sent back to the model as the tool result,
+        the same way as with ``handle``.
+        """
+
+        def respond(request: AiAgentActionRequest) -> None:
+            self.add_action_response(request.tool_id, action())
+
+        self.receive_no_args(action_name, respond, ai_handle_error)
+
+    def receive_no_args(
+        self,
+        action_name: str,
+        action: Callable[[AiAgentActionRequest], None],
+        ai_handle_error: AiHandleErrorStrategy = AiHandleErrorStrategy.SEND_ERRORS_TO_MODEL,
+    ) -> None:
+        """
+        Registers a receiver for an action tool that takes no parameters. The receiver gets only the
+        action request and sends the tool result itself with ``add_action_response``.
+        """
+        t = self.AiActionContext(self, lambda request, _: action(request), ai_handle_error, parse_arguments=False)
+        self._add_action(action_name, t.execute)
+
     def _add_action(self, action_name: str, action: Callable[[AiAgentActionRequest], Any]):
         if action_name in self._invocations:
             raise ValueError(f"Action '{action_name}' already exists")
@@ -352,13 +414,16 @@ def __init__(
             conversation: AiConversation,
             action: Callable[[AiAgentActionRequest, dict], Any],
             ai_handle_error: AiHandleErrorStrategy,
+            parse_arguments: bool = True,
         ):
             self._conversation = conversation
             self._action = action
             self._ai_handle_error = ai_handle_error
+            self._parse_arguments = parse_arguments
 
         def execute(self, action_request: AiAgentActionRequest):
-            args = json.loads(action_request.arguments)
+            # No-args tools get the raw arguments text, which the model may leave empty.
+            args = json.loads(action_request.arguments) if self._parse_arguments else action_request.arguments
             self.invoke(action_request, args)
 
         def invoke(self, action_request: AiAgentActionRequest, args: dict):

ravendb/documents/ai/ai_output_options.py

@@ -45,6 +45,9 @@ def to_json(self) -> Dict[str, Any]:
             sample = self.sample_object
             if callable(getattr(sample, "to_json", None)):
                 sample = sample.to_json()
+            elif not isinstance(sample, (dict, list, str)) and hasattr(sample, "__dict__"):
+                # A plain entity: its attributes describe the expected answer.
+                sample = sample.__dict__
             # The server reads this as a JSON string, not as a nested object.
             json_dict["SampleObject"] = json.dumps(sample)
 

ravendb/documents/operations/ai/agents/run_conversation_operation.py

@@ -421,7 +421,8 @@ def create_request(self, node: ServerNode) -> requests.Request:
         )
         if self._change_vector:
             url += f"&changeVector={quote(self._change_vector)}"
-        if self._stream_property_path:
+        # An empty path still streams: raw text output has no property to point at.
+        if self._stream_property_path is not None:
             url += f"&streaming=true&streamPropertyPath={quote(self._stream_property_path)}"
 
         # Add debug flag if requested
@@ -469,7 +470,7 @@ def create_request(self, node: ServerNode) -> requests.Request:
 
     # todo: rewrite via custom set_response_raw + RAW response type
     def process_response(self, cache, response: requests.Response, url) -> ResponseDisposeHandling:
-        if not self._stream_property_path:
+        if self._stream_property_path is None:
             return super().process_response(cache, response, url)
 
         for line in response.iter_lines(decode_unicode=True):
@@ -490,7 +491,7 @@ def process_response(self, cache, response: requests.Response, url) -> ResponseD
         return ResponseDisposeHandling.AUTOMATIC
 
     def send(self, session: requests.Session, request: requests.Request) -> requests.Response:
-        if self._stream_property_path:
+        if self._stream_property_path is not None:
             from ravendb.util.request_utils import RequestUtils
 
             prepared_request = session.prepare_request(request)

ravendb/tests/ai_agent_tests/test_ai_output_options.py

@@ -5,22 +5,44 @@
 
 import json
 import unittest
+from unittest.mock import MagicMock
 
+from ravendb.documents.ai.ai_answer import AiConversationStatus
 from ravendb.documents.ai.ai_conversation import AiConversation
 from ravendb.documents.ai.ai_output_options import AiOutputOptions
 from ravendb.documents.operations.ai.agents.run_conversation_operation import (
+    AiAgentActionRequest,
     ConversationRequestBody,
     ConversationResult,
     RunConversationOperation,
 )
 from ravendb.http.server_node import ServerNode
 
 
+def _request(operation: RunConversationOperation):
+    return operation.get_command(None).create_request(ServerNode("http://localhost:8080", "db"))
+
+
 def _request_body(operation: RunConversationOperation) -> dict:
-    request = operation.get_command(None).create_request(ServerNode("http://localhost:8080", "db"))
+    request = _request(operation)
     return json.loads(request.data) if isinstance(request.data, str) else request.data
 
 
+def _conversation(*results: ConversationResult):
+    # The store only has to hand each operation to maintenance.send; the replies are canned.
+    store = MagicMock()
+    store.maintenance.send.side_effect = list(results)
+    return AiConversation(store, agent_id="agents/1", conversation_id="chats/1"), store.maintenance.send
+
+
+def _done(response) -> ConversationResult:
+    return ConversationResult(conversation_id="chats/1", change_vector="A:1", response=response, action_requests=[])
+
+
+def _sent_operations(send) -> list:
+    return [call.args[0] for call in send.call_args_list]
+
+
 class TestAiOutputOptions(unittest.TestCase):
     def test_a_sample_object_is_sent_as_json_text(self):
         # The server reads SampleObject as a string, not as a nested object.
@@ -114,7 +136,7 @@ def test_the_other_body_fields_are_untouched(self):
 
 class TestConversationRunEntryPoints(unittest.TestCase):
     def test_the_schema_overriding_entry_points_exist(self):
-        for name in ("run", "run_with_schema", "stream", "stream_with_schema"):
+        for name in ("run", "run_with_schema", "run_text", "stream", "stream_with_schema", "stream_text"):
             self.assertTrue(hasattr(AiConversation, name), name)
 
     def test_they_refuse_to_run_without_options(self):
@@ -137,3 +159,155 @@ def test_a_structured_answer_is_still_read_as_an_object(self):
         result = ConversationResult.from_json({"ConversationId": "chats/1", "Response": {"Name": "sample"}})
 
         self.assertEqual({"Name": "sample"}, result.response)
+
+
+class _AlternativeSchema:
+    def __init__(self, summary: str = None, score: int = None):
+        self.Summary = summary
+        self.Score = score
+
+
+class TestSchemaOverrideShorthands(unittest.TestCase):
+    def test_a_string_is_taken_as_an_explicit_schema(self):
+        schema = '{"name":"alt","strict":true,"schema":{"type":"object"}}'
+        conversation, send = _conversation(_done({"Summary": "x", "Score": 1}))
+        conversation.set_user_prompt("Rate yourself")
+
+        conversation.run_with_schema(schema)
+
+        self.assertEqual({"OutputSchema": schema}, _request_body(_sent_operations(send)[0])["OutputOptions"])
+
+    def test_a_dict_is_taken_as_a_sample_object(self):
+        conversation, send = _conversation(_done({"Summary": "x", "Score": 1}))
+        conversation.set_user_prompt("Rate yourself")
+
+        conversation.run_with_schema({"Summary": "a short summary", "Score": 5})
+
+        options = _request_body(_sent_operations(send)[0])["OutputOptions"]
+        self.assertEqual({"Summary": "a short summary", "Score": 5}, json.loads(options["SampleObject"]))
+
+    def test_an_entity_is_taken_as_a_sample_object(self):
+        conversation, send = _conversation(_done({"Summary": "x", "Score": 1}))
+        conversation.set_user_prompt("Rate yourself")
+
+        answer = conversation.run_with_schema(_AlternativeSchema("a short summary", 5))
+
+        options = _request_body(_sent_operations(send)[0])["OutputOptions"]
+        self.assertEqual({"Summary": "a short summary", "Score": 5}, json.loads(options["SampleObject"]))
+        self.assertEqual(AiConversationStatus.DONE, answer.status)
+
+    def test_an_empty_schema_string_is_rejected(self):
+        conversation, send = _conversation()
+        conversation.set_user_prompt("Rate yourself")
+
+        with self.assertRaises(ValueError):
+            conversation.run_with_schema("  ")
+        send.assert_not_called()
+
+    def test_stream_with_schema_takes_a_sample_object(self):
+        conversation, send = _conversation(_done({"Summary": "x", "Score": 1}))
+        conversation.set_user_prompt("Rate yourself")
+
+        conversation.stream_with_schema("Summary", lambda chunk: None, _AlternativeSchema("a short summary", 5))
+
+        operation = _sent_operations(send)[0]
+        self.assertIn("streamPropertyPath=Summary", _request(operation).url)
+        self.assertIn("SampleObject", _request_body(operation)["OutputOptions"])
+
+
+class TestRawTextShorthands(unittest.TestCase):
+    def test_run_text_asks_for_no_schema_and_returns_the_text(self):
+        conversation, send = _conversation(_done("Hello! I am a helpful assistant."))
+        conversation.set_user_prompt("Say hello")
+
+        answer = conversation.run_text()
+
+        self.assertEqual("Hello! I am a helpful assistant.", answer.answer)
+        self.assertEqual({"NoSchema": True}, _request_body(_sent_operations(send)[0])["OutputOptions"])
+
+    def test_stream_text_streams_the_whole_answer_without_a_schema(self):
+        conversation, send = _conversation(_done("Hello"))
+        conversation.set_user_prompt("Say hello")
+
+        conversation.stream_text(lambda chunk: None)
+
+        operation = _sent_operations(send)[0]
+        self.assertIn("streaming=true&streamPropertyPath=&", _request(operation).url)
+        self.assertEqual({"NoSchema": True}, _request_body(operation)["OutputOptions"])
+
+    def test_stream_text_needs_a_callback(self):
+        conversation, _ = _conversation()
+
+        with self.assertRaises(ValueError):
+            conversation.stream_text(None)
+
+    def test_no_schema_streaming_does_not_need_a_property_path(self):
+        conversation, send = _conversation(_done("Hello"))
+        conversation.set_user_prompt("Say hello")
+
+        conversation.stream_with_schema(None, lambda chunk: None, AiOutputOptions(no_schema=True))
+
+        self.assertIn("streaming=true", _request(_sent_operations(send)[0]).url)
+
+    def test_a_turn_without_streaming_does_not_ask_for_it(self):
+        operation = RunConversationOperation(agent_id="agent", conversation_id="chats/1", prompt_parts=[])
+
+        self.assertNotIn("streaming", _request(operation).url)
+
+
+class TestNoArgsTools(unittest.TestCase):
+    @staticmethod
+    def _tool_call(arguments: str) -> ConversationResult:
+        return ConversationResult(
+            conversation_id="chats/1",
+            change_vector="A:1",
+            response=None,
+            action_requests=[AiAgentActionRequest(name="get_current_time", tool_id="call-1", arguments=arguments)],
+        )
+
+    def test_handle_no_args_calls_the_handler_without_arguments(self):
+        conversation, send = _conversation(self._tool_call(""), _done({"Answer": "noon"}))
+        conversation.handle_no_args("get_current_time", lambda: "2026-01-01T12:00:00Z")
+        conversation.set_user_prompt("What time is it?")
+
+        answer = conversation.run()
+
+        self.assertEqual(AiConversationStatus.DONE, answer.status)
+        responses = _request_body(_sent_operations(send)[1])["ActionResponses"]
+        self.assertEqual([{"ToolId": "call-1", "Content": "2026-01-01T12:00:00Z"}], responses)
+
+    def test_receive_no_args_gets_the_request(self):
+        received = []
+        conversation, send = _conversation(self._tool_call("{}"), _done({"Answer": "pong"}))
+
+        def on_ping(request):
+            received.append(request)
+            conversation.add_action_response(request.tool_id, "pong")
+
+        conversation.receive_no_args("get_current_time", on_ping)
+        conversation.set_user_prompt("Ping")
+
+        conversation.run()
+
+        self.assertEqual(["call-1"], [request.tool_id for request in received])
+        self.assertEqual("pong", _request_body(_sent_operations(send)[1])["ActionResponses"][0]["Content"])
+
+    def test_a_failing_no_args_handler_reports_the_error_to_the_model(self):
+        def broken():
+            raise RuntimeError("clock unavailable")
+
+        conversation, send = _conversation(self._tool_call(""), _done({"Answer": "unknown"}))
+        conversation.handle_no_args("get_current_time", broken)
+        conversation.set_user_prompt("What time is it?")
+
+        conversation.run()
+
+        content = _request_body(_sent_operations(send)[1])["ActionResponses"][0]["Content"]
+        self.assertIn("clock unavailable", content)
+
+    def test_a_no_args_tool_name_can_be_registered_only_once(self):
+        conversation, _ = _conversation()
+        conversation.handle_no_args("get_current_time", lambda: "now")
+
+        with self.assertRaises(ValueError):
+            conversation.receive_no_args("get_current_time", lambda request: None)