Move OpenAI decorator to DFApp class

azure/durable_functions/decorators/durable_app.py

@@ -1,5 +1,6 @@
 #  Copyright (c) Microsoft Corporation. All rights reserved.
 #  Licensed under the MIT License.
+
 from .metadata import OrchestrationTrigger, ActivityTrigger, EntityTrigger,\
     DurableClient
 from typing import Callable, Optional
@@ -45,6 +46,7 @@ def __init__(self,
             New instance of a Durable Functions app
         """
         super().__init__(auth_level=http_auth_level)
+        self._is_durable_openai_agent_setup = False
 
     def _configure_entity_callable(self, wrap) -> Callable:
         """Obtain decorator to construct an Entity class from a user-defined Function.
@@ -250,6 +252,54 @@ def decorator():
 
         return wrap
 
+    def _create_invoke_model_activity(self, model_provider):
+        """Create and register the invoke_model_activity function with the provided FunctionApp."""
+
+        @self.activity_trigger(input_name="input")
+        async def invoke_model_activity(input: str):
+            from azure.durable_functions.openai_agents.orchestrator_generator\
+                import durable_openai_agent_activity
+
+            return await durable_openai_agent_activity(input, model_provider)
+
+        return invoke_model_activity
+
+    def _setup_durable_openai_agent(self, model_provider):
+        if not self._is_durable_openai_agent_setup:
+            self._create_invoke_model_activity(model_provider)
+            self._is_durable_openai_agent_setup = True
+
+    def durable_openai_agent_orchestrator(self, _func=None, *, model_provider=None):
+        """Decorate Azure Durable Functions orchestrators that use OpenAI Agents.
+
+        Parameters
+        ----------
+        model_provider: Optional[ModelProvider]
+            Use a non-default ModelProvider instead of the default OpenAIProvider,
+            such as when testing.
+        """
+        from agents import ModelProvider
+        from azure.durable_functions.openai_agents.orchestrator_generator\
+            import durable_openai_agent_orchestrator_generator
+
+        if model_provider is not None and type(model_provider) is not ModelProvider:
+            raise TypeError("Provided model provider must be of type ModelProvider")
+
+        self._setup_durable_openai_agent(model_provider)
+
+        def generator_wrapper_wrapper(func):
+
+            @wraps(func)
+            def generator_wrapper(context):
+                return durable_openai_agent_orchestrator_generator(func, context)
+
+            return generator_wrapper
+
+        if _func is None:
+            return generator_wrapper_wrapper
+        else:
+            return generator_wrapper_wrapper(_func)
+
 
 class DFApp(Blueprint, FunctionRegister):
     """Durable Functions (DF) app.

azure/durable_functions/openai_agents/__init__.py

@@ -4,10 +4,8 @@
 with Azure Durable Functions orchestration patterns.
 """
 
-from .decorators import durable_openai_agent_orchestrator
 from .context import DurableAIAgentContext
 
 __all__ = [
-    'durable_openai_agent_orchestrator',
     'DurableAIAgentContext',
 ]

azure/durable_functions/openai_agents/orchestrator_generator.py

@@ -0,0 +1,75 @@
+import inspect
+from agents import ModelProvider, ModelResponse
+from agents.run import set_default_agent_runner
+from azure.durable_functions.models.DurableOrchestrationContext import DurableOrchestrationContext
+from azure.durable_functions.openai_agents.model_invocation_activity\
+    import ActivityModelInput, ModelInvoker
+from .runner import DurableOpenAIRunner
+from .exceptions import YieldException
+from .context import DurableAIAgentContext
+from .event_loop import ensure_event_loop
+
+
+async def durable_openai_agent_activity(input: str, model_provider: ModelProvider):
+    """Activity logic that handles OpenAI model invocations."""
+    activity_input = ActivityModelInput.from_json(input)
+
+    model_invoker = ModelInvoker(model_provider=model_provider)
+    result = await model_invoker.invoke_model_activity(activity_input)
+
+    json_obj = ModelResponse.__pydantic_serializer__.to_json(result)
+    return json_obj.decode()
+
+
+def durable_openai_agent_orchestrator_generator(
+        func,
+        durable_orchestration_context: DurableOrchestrationContext):
+    """Adapts the synchronous OpenAI Agents function to an Durable orchestrator generator."""
+    ensure_event_loop()
+    durable_ai_agent_context = DurableAIAgentContext(durable_orchestration_context)
+    durable_openai_runner = DurableOpenAIRunner(context=durable_ai_agent_context)
+    set_default_agent_runner(durable_openai_runner)
+
+    if inspect.isgeneratorfunction(func):
+        gen = iter(func(durable_ai_agent_context))
+        try:
+            # prime the subiterator
+            value = next(gen)
+            yield from durable_ai_agent_context._yield_and_clear_tasks()
+            while True:
+                try:
+                    # send whatever was sent into us down to the subgenerator
+                    yield from durable_ai_agent_context._yield_and_clear_tasks()
+                    sent = yield value
+                except GeneratorExit:
+                    # ensure the subgenerator is closed
+                    if hasattr(gen, "close"):
+                        gen.close()
+                    raise
+                except BaseException as exc:
+                    # forward thrown exceptions if possible
+                    if hasattr(gen, "throw"):
+                        value = gen.throw(type(exc), exc, exc.__traceback__)
+                    else:
+                        raise
+                else:
+                    # normal path: forward .send (or .__next__)
+                    if hasattr(gen, "send"):
+                        value = gen.send(sent)
+                    else:
+                        value = next(gen)
+        except StopIteration as e:
+            yield from durable_ai_agent_context._yield_and_clear_tasks()
+            return e.value
+        except YieldException as e:
+            yield from durable_ai_agent_context._yield_and_clear_tasks()
+            yield e.task
+    else:
+        try:
+            result = func(durable_ai_agent_context)
+            return result
+        except YieldException as e:
+            yield from durable_ai_agent_context._yield_and_clear_tasks()
+            yield e.task
+        finally:
+            yield from durable_ai_agent_context._yield_and_clear_tasks()

samples-v2/openai_agents/function_app.py

@@ -1,7 +1,8 @@
 import os
 
 import azure.functions as func
-from azure.durable_functions.openai_agents import durable_openai_agent_orchestrator
+import azure.durable_functions as df
+
 from azure.identity import DefaultAzureCredential
 from openai import AsyncAzureOpenAI
 
@@ -31,7 +32,7 @@ def get_azure_token():
 # endregion
 
 
-app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)
+app = df.DFApp(http_auth_level=func.AuthLevel.FUNCTION)
 
 @app.route(route="orchestrators/{functionName}")
 @app.durable_client_input(client_name="client")
@@ -44,59 +45,59 @@ async def orchestration_starter(req: func.HttpRequest, client):
 
 
 @app.orchestration_trigger(context_name="context")
-@durable_openai_agent_orchestrator
+@app.durable_openai_agent_orchestrator
 def hello_world(context):
     import basic.hello_world
     return basic.hello_world.main()
 
 @app.orchestration_trigger(context_name="context")
-@durable_openai_agent_orchestrator
+@app.durable_openai_agent_orchestrator
 def agent_lifecycle_example(context):
     import basic.agent_lifecycle_example
     return basic.agent_lifecycle_example.main()
 
 
 @app.orchestration_trigger(context_name="context")
-@durable_openai_agent_orchestrator
+@app.durable_openai_agent_orchestrator
 def dynamic_system_prompt(context):
     import basic.dynamic_system_prompt
     return basic.dynamic_system_prompt.main()
 
 @app.orchestration_trigger(context_name="context")
-@durable_openai_agent_orchestrator
+@app.durable_openai_agent_orchestrator
 def lifecycle_example(context):
     import basic.lifecycle_example
     return basic.lifecycle_example.main()
 
 
 @app.orchestration_trigger(context_name="context")
-@durable_openai_agent_orchestrator
+@app.durable_openai_agent_orchestrator
 def local_image(context):
     import basic.local_image
     return basic.local_image.main()
 
 
 @app.orchestration_trigger(context_name="context")
-@durable_openai_agent_orchestrator
+@app.durable_openai_agent_orchestrator
 def non_strict_output_type(context):
     import basic.non_strict_output_type
     return basic.non_strict_output_type.main()
 
 
 @app.orchestration_trigger(context_name="context")
-@durable_openai_agent_orchestrator
+@app.durable_openai_agent_orchestrator
 def previous_response_id(context):
     import basic.previous_response_id
     return basic.previous_response_id.main()
 
 @app.orchestration_trigger(context_name="context")
-@durable_openai_agent_orchestrator
+@app.durable_openai_agent_orchestrator
 def remote_image(context):
     import basic.remote_image
     return basic.remote_image.main()
 
 @app.orchestration_trigger(context_name="context")
-@durable_openai_agent_orchestrator
+@app.durable_openai_agent_orchestrator
 def tools(context):
     import basic.tools
     return basic.tools.main()