Allow invoking durable activities

Author: AnatoliB

samples-v2/openai_agents/durable_ai_agent_context.py

@@ -25,6 +25,11 @@ def get_activity_call_result(self, activity_name, input: str):
             result_json = completed_tasks[self.activities_called - 1].Result
             result = json.loads(result_json)
             return result
+        
+    def call_activity_yieldable(self, activity_name, input: str):
+        task = self.context.call_activity(activity_name, input)
+        self.activities_called += 1
+        return task
 
     def yield_and_clear_tasks(self):
         """Yield all accumulated tasks and clear the tasks list."""

samples-v2/openai_agents/durable_decorators.py

@@ -1,4 +1,5 @@
 from functools import wraps
+import inspect
 from agents.run import set_default_agent_runner
 from azure.durable_functions.models.DurableOrchestrationContext import DurableOrchestrationContext
 from durable_openai_runner import DurableOpenAIRunner
@@ -15,13 +16,47 @@ def wrapper(durable_orchestration_context: DurableOrchestrationContext):
         durable_openai_runner = DurableOpenAIRunner(context=durable_ai_agent_context)
         set_default_agent_runner(durable_openai_runner)
 
-        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()
+        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
+                        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()
 
     return wrapper

samples-v2/openai_agents/function_app.py

@@ -10,6 +10,12 @@
 create_invoke_model_activity(app) # TODO: Can we hide this line?
 
 
+@app.activity_trigger(input_name="name")
+async def hello(name: str):
+    """A simple durable activity that returns a greeting message."""
+    return f"Hello {name}!"
+
+
 @app.route(route="orchestrators/{functionName}")
 @app.durable_client_input(client_name="client")
 async def orchestration_starter(req: func.HttpRequest, client):
@@ -47,12 +53,21 @@ def haiku_judge(context):
 
     haiku = Runner.run_sync(writer, "Tell me about recursion in programming.")
 
+    response = yield context.call_activity_yieldable("hello", "World")
+
     judge = Agent(
         name="Judge",
         instructions="You judge haikus.",
     )
 
     evaluation = Runner.run_sync(judge, f"Is this a good haiku? {haiku.final_output}")
 
-    return [haiku.final_output, "\n\n" + evaluation.final_output]
+    return [haiku.final_output, "\n\n" + evaluation.final_output, response]
 
+
+@app.orchestration_trigger(context_name="context")
+@durable_openai_agent_orchestrator
+def hello_cities(context):
+    task = context.call_activity_yieldable("hello", "Seattle")
+    result = yield task
+    return result