Moved CustomerServiceWorkflow logic into the update handler for clarity.

Author: mfateev

openai_agents/customer_service_client.py

@@ -1,12 +1,13 @@
 import argparse
 import asyncio
+from ftplib import print_line
 
-from temporalio.client import Client, WorkflowQueryRejectedError
+from temporalio.client import Client, WorkflowQueryRejectedError, WorkflowUpdateFailedError
 from temporalio.common import WorkflowIDReusePolicy, QueryRejectCondition
 from temporalio.service import RPCError, RPCStatusCode
 
-from openai_agents.workflows.customer_service_workflow import CustomerServiceWorkflow
 from openai_agents.adapters.open_ai_converter import open_ai_data_converter
+from openai_agents.workflows.customer_service_workflow import CustomerServiceWorkflow, ProcessUserMessageInput
 
 
 async def main():
@@ -42,8 +43,17 @@ async def main():
     # Loop to send messages to the workflow
     while True:
         user_input = input("Enter your message: ")
-        new_history = await handle.execute_update(CustomerServiceWorkflow.process_user_message, user_input)
-        print(*new_history, sep="\n")
+        message_input = ProcessUserMessageInput(user_input=user_input, chat_length=len(history))
+        try:
+            new_history = await handle.execute_update(CustomerServiceWorkflow.process_user_message, message_input)
+            history.extend(new_history)
+            print(*new_history, sep="\n")
+        except WorkflowUpdateFailedError:
+            print_line("** Stale conversation. Reloading...")
+            length = len(history)
+            history = await handle.query(CustomerServiceWorkflow.get_chat_history,
+                                         reject_condition=QueryRejectCondition.NOT_OPEN)
+            print(*history[length:], sep="\n")
 
 
 if __name__ == "__main__":

openai_agents/workflows/customer_service_workflow.py

@@ -2,7 +2,6 @@
 
 from temporalio import workflow
 
-
 with workflow.unsafe.imports_passed_through():
     from pydantic import BaseModel
     from openai_agents.adapters.activity_model import ModelStubProvider
@@ -136,55 +135,64 @@ def init_agents() -> Agent[AirlineAgentContext]:
     return triage_agent
 
 
-### RUN
+class ProcessUserMessageInput(BaseModel):
+    user_input: str
+    chat_length: int
+
 
 @workflow.defn(sandboxed=False)
 class CustomerServiceWorkflow:
 
-    def __init__(self):
+    def __init__(self, input_items: list[TResponseInputItem] = None):
         self.run_config = RunConfig(model_provider=ModelStubProvider())
         self.chat_history = []
-        self.user_input = None
+        self.current_agent: Agent[AirlineAgentContext] = init_agents()
+        self.context = AirlineAgentContext()
+        self.input_items = [] if input_items is None else input_items
 
     @workflow.run
     async def run(self, input_items: list[TResponseInputItem] = None):
-        input_items = [] #input_items or []
-        current_agent: Agent[AirlineAgentContext] = init_agents()
-        context = AirlineAgentContext()
-        while not workflow.info().is_continue_as_new_suggested():
-            await workflow.wait_condition(lambda: self.user_input is not None)
-            self.chat_history.append(f"User: {self.user_input}")
-            with trace("Customer service", group_id=workflow.info().workflow_id):
-                input_items.append({"content": self.user_input, "role": "user"})
-                result = await Runner.run(current_agent, input_items, context=context, run_config=self.run_config)
-
-                for new_item in result.new_items:
-                    agent_name = new_item.agent.name
-                    if isinstance(new_item, MessageOutputItem):
-                        self.chat_history.append(f"{agent_name}: {ItemHelpers.text_message_output(new_item)}")
-                    elif isinstance(new_item, HandoffOutputItem):
-                        self.chat_history.append(
-                            f"Handed off from {new_item.source_agent.name} to {new_item.target_agent.name}"
-                        )
-                    elif isinstance(new_item, ToolCallItem):
-                        self.chat_history.append(f"{agent_name}: Calling a tool")
-                    elif isinstance(new_item, ToolCallOutputItem):
-                        self.chat_history.append(f"{agent_name}: Tool call output: {new_item.output}")
-                    else:
-                        self.chat_history.append(f"{agent_name}: Skipping item: {new_item.__class__.__name__}")
-                input_items = result.to_input_list()
-                current_agent = result.last_agent
-            self.user_input = None
-        workflow.continue_as_new(input_items)
+        await workflow.wait_condition(
+            lambda: workflow.info().is_continue_as_new_suggested() and workflow.all_handlers_finished())
+        workflow.continue_as_new(self.input_items)
 
     @workflow.query
     def get_chat_history(self) -> list[str]:
         return self.chat_history
 
     @workflow.update
-    async def process_user_message(self, user_input: str) -> list[str]:
-        await workflow.wait_condition(lambda: self.user_input is None)
-        self.user_input = user_input
+    async def process_user_message(self, input: ProcessUserMessageInput) -> list[str]:
         length = len(self.chat_history)
-        await workflow.wait_condition(lambda: self.user_input is None)
+        self.chat_history.append(f"User: {input.user_input}")
+        with trace("Customer service", group_id=workflow.info().workflow_id):
+            self.input_items.append({"content": input.user_input, "role": "user"})
+            result = await Runner.run(self.current_agent, self.input_items, context=self.context,
+                                      run_config=self.run_config)
+
+            for new_item in result.new_items:
+                agent_name = new_item.agent.name
+                if isinstance(new_item, MessageOutputItem):
+                    self.chat_history.append(f"{agent_name}: {ItemHelpers.text_message_output(new_item)}")
+                elif isinstance(new_item, HandoffOutputItem):
+                    self.chat_history.append(
+                        f"Handed off from {new_item.source_agent.name} to {new_item.target_agent.name}"
+                    )
+                elif isinstance(new_item, ToolCallItem):
+                    self.chat_history.append(f"{agent_name}: Calling a tool")
+                elif isinstance(new_item, ToolCallOutputItem):
+                    self.chat_history.append(f"{agent_name}: Tool call output: {new_item.output}")
+                else:
+                    self.chat_history.append(f"{agent_name}: Skipping item: {new_item.__class__.__name__}")
+            self.input_items = result.to_input_list()
+            self.current_agent = result.last_agent
+        workflow.set_current_details("\n\n".join(self.chat_history))
         return self.chat_history[length:]
+
+    @process_user_message.validator
+    def validate_process_user_message(self, input: ProcessUserMessageInput) -> None:
+        if not input.user_input:
+            raise ValueError("User input cannot be empty.")
+        if len(input.user_input) > 1000:
+            raise ValueError("User input is too long. Please limit to 1000 characters.")
+        if input.chat_length != len(self.chat_history):
+            raise ValueError("Stale chat history. Please refresh the chat.")