Repository: run-llama/llama_index

Issue:
[Question]: Resuming a Workflow
### Question Validation

- [x] I have searched both the documentation and discord for an answer.

### Question

I'm attempting to build a workflow that includes multiple steps that can return InputRequiredEvents. For simplicity, let's say the workflow looks like this:

Steps:
1. ask_question_1 (accepts **StartEvent**, produces **InputRequiredQuestion1Event**)
2. process_answer1 (accepts **HumanResponseQuestion1Event**, produces **InputRequiredQuestion2Event**)
3. process_answer2 (accepts **HumanResponseQuestion2Event**, produces **StopEvent**)

When an InputRequiredEvent is produced, I want to pause the workflow and send the question and the workflow's current context back to the client.

`handler = workflow.run()`
`async for event in handler.stream_events():`
`   if isinstance(event, InputRequiredQuestion1Event):`
`        return {`
`            "response": event.prefix,`
`            "context": handler.ctx.to_dict(serializer=JsonSerializer()),`
`             "question": "q1"`
`         }`
`    elif ....handle other input required event types....`

Once I receive a response back from the client, I attempt to resume the workflow where it left off by initializing the previous context and calling:
`handler: WorkflowHandler = workflow.run(ctx=ctx)`
`handler.ctx.send_event(HumanResponseQuestion1Event(response="my response"))`

However, this seems to start the workflow from the very beginning at the **ask_question_1** step as opposed to jumping to the **process_answer1** step. This causes `handler.stream_events()` to always produce a InputRequiredQuestion1Event instead of picking up directly where it left off.

Is there a way to "resume" a workflow using context in a way that it starts where it left off as opposed to rerunning from the beginning?
Is calling `workflow.run(ctx=ctx)` incorrect? And if so, is there a way to instantiate the previous WorkflowHandler directly?

Discussion:
@dosubot: <!-- Greeting -->
Hey @michaelbluegenai! 👋 I'm here to help you with any bugs, questions, or contributions you have while we wait for a human maintainer.

<!-- Answer -->
To resume a workflow from where it left off, you need to ensure that the context is correctly maintaining the state of the workflow, including the last completed step. The `workflow.run(ctx=ctx)` method should indeed start the workflow using the provided context, but it seems like the context might not be correctly capturing the state needed to resume from the desired step.

Here are a few things to check and try:

1. **Ensure Context Serialization**: Make sure that the context is being serialized and deserialized correctly. The context should include information about the last completed step. Use a serializer like `JsonSerializer` or `JsonPickleSerializer` to handle this.

2. **Check Context State**: Before resuming, verify that the context's state reflects the last completed step. The context should have a record of the last event processed, which should be `HumanResponseQuestion1Event` in your case.

3. **Use Workflow Checkpointing**: Consider using the `WorkflowCheckpointer` to create checkpoints at each step. This allows you to resume from a specific checkpoint rather than relying solely on the context. You can use `WorkflowCheckpointer.run_from()` to start from a specific checkpoint, which might be more reliable for your use case <sup>[[1]](https://github.com/run-llama/llama_index/blob/main/docs/docs/examples/workflow/checkpointing_workflows.ipynb)</sup>.

4. **Verify Event Handling**: Ensure that the event handling logic correctly updates the context with the new state after processing each event. This includes marking the step as completed and updating any necessary state variables.

By ensuring that the context or checkpoints accurately reflect the workflow's state, you should be able to resume the workflow from the desired step. If the context alone is not sufficient, implementing checkpointing might be the solution you need.


<!-- Dosu Comment Footer -->
*To continue talking to [Dosu](https://dosu.dev), mention @dosu.*

---
<sup>Help Dosu learn! Give it feedback: [Great Response](https://app.dosu.dev/response-feedback/876aa40d-9776-435c-9978-0bc1ce30ef4f?feedback_type=great_response) | [Irrelevant Answer](https://app.dosu.dev/response-feedback/876aa40d-9776-435c-9978-0bc1ce30ef4f?feedback_type=irrelevant_answer) | [Incorrect Sources](https://app.dosu.dev/response-feedback/876aa40d-9776-435c-9978-0bc1ce30ef4f?feedback_type=incorrect_sources) | [Too Verbose](https://app.dosu.dev/response-feedback/876aa40d-9776-435c-9978-0bc1ce30ef4f?feedback_type=too_verbose) | [Hallucination](https://app.dosu.dev/response-feedback/876aa40d-9776-435c-9978-0bc1ce30ef4f?feedback_type=hallucination) | [Bug Report](https://app.dosu.dev/response-feedback/876aa40d-9776-435c-9978-0bc1ce30ef4f?feedback_type=bug_report) | [Other](https://app.dosu.dev/response-feedback/876aa40d-9776-435c-9978-0bc1ce30ef4f?feedback_type=other)</sup>
@michaelbluegenai: I should note that after running a quick test, the issue only seems to occur if I serialize the context and then reinitialize it before passing it to the workflow. If I simply pass around the original Context object then the workflow behaves as expected. But I'm having trouble figuring out what I'm doing wrong in the serialization step:

`handler.ctx.to_dict(serializer=JsonSerializer())`
and
`Context.from_dict(workflow, existing_ctx, serializer=JsonSerializer())`

Attaching my test scripts here - any help is appreciated!

WITHOUT SERIALIZATION (working)
```python
import asyncio
import json
from llama_index.core.workflow import (
    Workflow,
    StartEvent,
    StopEvent,
    InputRequiredEvent,
    HumanResponseEvent,
    Context,
    JsonSerializer,
    step,
)
from llama_index.core.workflow.handler import WorkflowHandler

class InputRequiredQuestion1Event(InputRequiredEvent):
    prefix:str

class InputRequiredQuestion2Event(InputRequiredEvent):
    prefix:str

class HumanResponseForQuestion1Event(HumanResponseEvent):
    response:str

class HumanResponseForQuestion2Event(HumanResponseEvent):
    response:str

class ExtendedWorkFlow(Workflow):
    workflow_name = "extended_workflow_1"

    @step
    async def ask_question1(self, ctx: Context, ev: StartEvent) -> InputRequiredQuestion1Event:
        return InputRequiredQuestion1Event(
            prefix="Please provide your favourite color:"
        )

    @step
    async def process_answer1(self, ctx: Context, ev: HumanResponseForQuestion1Event) -> InputRequiredQuestion2Event:
        await ctx.set("color", ev.response)
        
        return InputRequiredQuestion2Event(
            prefix="Please provide your favourite animal:"
        )

    @step
    async def process_answer2(self, ctx: Context, ev: HumanResponseForQuestion2Event) -> StopEvent:
        color = await ctx.get("color")
        animal = ev.response

        return StopEvent(result=f"Your favorite color is {color} and your favorite animal is {animal}")
    



async def run_workflow(existing_ctx=None, next_event: HumanResponseForQuestion1Event | HumanResponseForQuestion1Event | None = None):
    workflow = ExtendedWorkFlow(verbose=True)
    
    if existing_ctx is None:
        handler: WorkflowHandler = workflow.run()
    else:
        handler: WorkflowHandler = workflow.run(ctx=existing_ctx)
        handler.ctx.send_event(next_event)

    async for event in handler.stream_events():
        if isinstance(event, (InputRequiredQuestion1Event, InputRequiredQuestion2Event)):
            print(f"Returning due to event of type {type(event)}")
            return event.prefix, handler.ctx




async def process_workflow_steps():
    question, context = await run_workflow()
    print(f"QUESTION: {question}")
    print("RESPONDING: blue\n\n")
    question, context = await run_workflow(context, HumanResponseForQuestion1Event(response="blue"))
    print(f"QUESTION: {question}")
    print("RESPONDING: dog\n\n")
    await run_workflow(context, HumanResponseForQuestion2Event(response="dog"))
    print("FINAL")

asyncio.run(process_workflow_steps())
```

WITH SERIALIZATION (not working)
```python
import asyncio
import json
from llama_index.core.workflow import (
    Workflow,
    StartEvent,
    StopEvent,
    InputRequiredEvent,
    HumanResponseEvent,
    Context,
    JsonSerializer,
    step,
)
from llama_index.core.workflow.handler import WorkflowHandler

class InputRequiredQuestion1Event(InputRequiredEvent):
    prefix:str

class InputRequiredQuestion2Event(InputRequiredEvent):
    prefix:str

class HumanResponseForQuestion1Event(HumanResponseEvent):
    response:str

class HumanResponseForQuestion2Event(HumanResponseEvent):
    response:str

class ExtendedWorkFlow(Workflow):
    workflow_name = "extended_workflow_1"

    @step
    async def ask_question1(self, ctx: Context, ev: StartEvent) -> InputRequiredQuestion1Event:
        return InputRequiredQuestion1Event(
            prefix="Please provide your favourite color:"
        )

    @step
    async def process_answer1(self, ctx: Context, ev: HumanResponseForQuestion1Event) -> InputRequiredQuestion2Event:
        await ctx.set("color", ev.response)
        
        return InputRequiredQuestion2Event(
            prefix="Please provide your favourite animal:"
        )

    @step
    async def process_answer2(self, ctx: Context, ev: HumanResponseForQuestion2Event) -> StopEvent:
        color = await ctx.get("color")
        animal = ev.response

        return StopEvent(result=f"Your favorite color is {color} and your favorite animal is {animal}")
    



async def run_workflow(existing_ctx=None, next_event: HumanResponseForQuestion1Event | HumanResponseForQuestion1Event | None = None):
    workflow = ExtendedWorkFlow(verbose=True)
    
    if existing_ctx is None:
        handler: WorkflowHandler = workflow.run()
    else:
        ctx = Context.from_dict(workflow, existing_ctx, serializer=JsonSerializer())
        handler: WorkflowHandler = workflow.run(ctx=ctx)
        handler.ctx.send_event(next_event)

    async for event in handler.stream_events():
        if isinstance(event, (InputRequiredQuestion1Event, InputRequiredQuestion2Event)):
            print(f"Returning due to event of type {type(event)}")
            return event.prefix, handler.ctx.to_dict(serializer=JsonSerializer())




async def process_workflow_steps():
    question, context = await run_workflow()
    print(f"QUESTION: {question}")
    print("RESPONDING: blue\n\n")
    question, context = await run_workflow(context, HumanResponseForQuestion1Event(response="blue"))
    print(f"QUESTION: {question}")
    print("RESPONDING: dog\n\n")
    await run_workflow(context, HumanResponseForQuestion2Event(response="dog"))
    print("FINAL")

asyncio.run(process_workflow_steps())
```@logan-markewich: hmm yea, something is sus here, looking into it@logan-markewich: Found the bug! PR in a sec@logan-markewich: Merged into `main`, will be in the next release (likely later tonight)

Solution patch:
diff --git a/llama-index-core/llama_index/core/node_parser/file/markdown.py b/llama-index-core/llama_index/core/node_parser/file/markdown.py
index ef3a5f6a2..30ec317bb 100644
--- a/llama-index-core/llama_index/core/node_parser/file/markdown.py
+++ b/llama-index-core/llama_index/core/node_parser/file/markdown.py
@@ -2,6 +2,7 @@
 import re
 from typing import Any, List, Optional, Sequence
 
+from llama_index.core.bridge.pydantic import Field
 from llama_index.core.callbacks.base import CallbackManager
 from llama_index.core.node_parser.interface import NodeParser
 from llama_index.core.node_parser.node_utils import build_nodes_from_splits
@@ -18,19 +19,26 @@ class MarkdownNodeParser(NodeParser):
     Args:
         include_metadata (bool): whether to include metadata in nodes
         include_prev_next_rel (bool): whether to include prev/next relationships
+        header_path_separator (str): separator char used for section header path metadata
     """
 
+    header_path_separator: str = Field(
+        default="/", description="Separator char used for section header path metadata."
+    )
+
     @classmethod
     def from_defaults(
         cls,
         include_metadata: bool = True,
         include_prev_next_rel: bool = True,
+        header_path_separator: str = "/",
         callback_manager: Optional[CallbackManager] = None,
     ) -> "MarkdownNodeParser":
         callback_manager = callback_manager or CallbackManager([])
         return cls(
             include_metadata=include_metadata,
             include_prev_next_rel=include_prev_next_rel,
+            header_path_separator=header_path_separator,
             callback_manager=callback_manager,
         )
 
@@ -61,7 +69,9 @@ class MarkdownNodeParser(NodeParser):
                             self._build_node_from_split(
                                 current_section.strip(),
                                 node,
-                                "/".join(h[1] for h in header_stack[:-1]),
+                                self.header_path_separator.join(
+                                    h[1] for h in header_stack[:-1]
+                                ),
                             )
                         )
 
@@ -87,7 +97,7 @@ class MarkdownNodeParser(NodeParser):
                 self._build_node_from_split(
                     current_section.strip(),
                     node,
-                    "/".join(h[1] for h in header_stack[:-1]),
+                    self.header_path_separator.join(h[1] for h in header_stack[:-1]),
                 )
             )
 
@@ -103,8 +113,12 @@ class MarkdownNodeParser(NodeParser):
         node = build_nodes_from_splits([text_split], node, id_func=self.id_func)[0]
 
         if self.include_metadata:
+            separator = self.header_path_separator
             node.metadata["header_path"] = (
-                "/" + header_path + "/" if header_path else "/"
+                # ex: "/header1/header2/" || "/"
+                separator + header_path + separator
+                if header_path
+                else separator
             )
 
         return node
diff --git a/llama-index-core/llama_index/core/workflow/workflow.py b/llama-index-core/llama_index/core/workflow/workflow.py
index cbcbc277b..b86dc0730 100644
--- a/llama-index-core/llama_index/core/workflow/workflow.py
+++ b/llama-index-core/llama_index/core/workflow/workflow.py
@@ -362,8 +362,6 @@ class Workflow(metaclass=WorkflowMeta):
                         warnings.warn(
                             f"Step function {name} returned {type(new_ev).__name__} instead of an Event instance."
                         )
-                    elif isinstance(new_ev, InputRequiredEvent):
-                        ctx.write_event_to_stream(new_ev)
                     else:
                         if stepwise:
                             async with ctx._step_condition:
@@ -393,7 +391,13 @@ class Workflow(metaclass=WorkflowMeta):
                                     input_ev=ev,
                                     output_ev=new_ev,
                                 )
-                            ctx.send_event(new_ev)
+
+                            # InputRequiredEvent's are special case and need to be written to the stream
+                            # this way, the user can access and respond to the event
+                            if isinstance(new_ev, InputRequiredEvent):
+                                ctx.write_event_to_stream(new_ev)
+                            else:
+                                ctx.send_event(new_ev)
 
             for _ in range(step_config.num_workers):
                 ctx._tasks.add(
diff --git a/llama-index-integrations/llms/llama-index-llms-anthropic/llama_index/llms/anthropic/base.py b/llama-index-integrations/llms/llama-index-llms-anthropic/llama_index/llms/anthropic/base.py
index 81a8c6ac1..278388d35 100644
--- a/llama-index-integrations/llms/llama-index-llms-anthropic/llama_index/llms/anthropic/base.py
+++ b/llama-index-integrations/llms/llama-index-llms-anthropic/llama_index/llms/anthropic/base.py
@@ -467,14 +467,20 @@ class Anthropic(FunctionCallingLLM):
             chat_history.append(user_msg)
 
         tool_dicts = []
-        for tool in tools:
-            tool_dicts.append(
-                {
-                    "name": tool.metadata.name,
-                    "description": tool.metadata.description,
-                    "input_schema": tool.metadata.get_parameters_dict(),
-                }
-            )
+        if tools:
+            for tool in tools:
+                tool_dicts.append(
+                    {
+                        "name": tool.metadata.name,
+                        "description": tool.metadata.description,
+                        "input_schema": tool.metadata.get_parameters_dict(),
+                    }
+                )
+            if "prompt-caching" in kwargs.get("extra_headers", {}).get(
+                "anthropic-beta", ""
+            ):
+                tool_dicts[-1]["cache_control"] = {"type": "ephemeral"}
+
         return {"messages": chat_history, "tools": tool_dicts, **kwargs}
 
     def _validate_chat_with_tools_response(
diff --git a/llama-index-integrations/llms/llama-index-llms-anthropic/llama_index/llms/anthropic/utils.py b/llama-index-integrations/llms/llama-index-llms-anthropic/llama_index/llms/anthropic/utils.py
index 06e27c5ff..42731637d 100644
--- a/llama-index-integrations/llms/llama-index-llms-anthropic/llama_index/llms/anthropic/utils.py
+++ b/llama-index-integrations/llms/llama-index-llms-anthropic/llama_index/llms/anthropic/utils.py
@@ -2,7 +2,7 @@
 Utility functions for the Anthropic SDK LLM integration.
 """
 
-from typing import Dict, Sequence, Tuple
+from typing import Any, Dict, Sequence, Tuple
 
 from llama_index.core.base.llms.types import (
     ChatMessage,
@@ -139,13 +139,16 @@ def messages_to_anthropic_messages(
         - System prompt
     """
     anthropic_messages = []
-    system_prompt = ""
+    system_prompt = []
     for message in messages:
         if message.role == MessageRole.SYSTEM:
-            # For system messages, concatenate all text blocks
             for block in message.blocks:
-                if isinstance(block, TextBlock):
-                    system_prompt += block.text + "\n"
+                if isinstance(block, TextBlock) and block.text:
+                    system_prompt.append(
+                        _text_block_to_anthropic_message(
+                            block, message.additional_kwargs
+                        )
+                    )
         elif message.role == MessageRole.FUNCTION or message.role == MessageRole.TOOL:
             content = ToolResultBlockParam(
                 tool_use_id=message.additional_kwargs["tool_call_id"],
@@ -161,19 +164,12 @@ def messages_to_anthropic_messages(
             content: list[TextBlockParam | ImageBlockParam] = []
             for block in message.blocks:
                 if isinstance(block, TextBlock):
-                    content_ = (
-                        TextBlockParam(
-                            text=block.text,
-                            type="text",
-                            cache_control=CacheControlEphemeralParam(type="ephemeral"),
+                    if block.text:
+                        content.append(
+                            _text_block_to_anthropic_message(
+                                block, message.additional_kwargs
+                            )
                         )
-                        if "cache_control" in message.additional_kwargs
-                        else TextBlockParam(text=block.text, type="text")
-                    )
-
-                    # avoid empty text blocks
-                    if content_["text"]:
-                        content.append(content_)
                 elif isinstance(block, ImageBlock):
                     # FUTURE: Claude does not support URLs, so we need to always convert to base64
                     img_bytes = block.resolve_image(as_base64=True).read()
@@ -214,7 +210,19 @@ def messages_to_anthropic_messages(
                 content=content,
             )
             anthropic_messages.append(anth_message)
-    return __merge_common_role_msgs(anthropic_messages), system_prompt.strip()
+    return __merge_common_role_msgs(anthropic_messages), system_prompt
+
+
+def _text_block_to_anthropic_message(
+    block: TextBlock, kwargs: dict[str, Any]
+) -> TextBlockParam:
+    if "cache_control" in kwargs:
+        return TextBlockParam(
+            text=block.text,
+            type="text",
+            cache_control=CacheControlEphemeralParam(type="ephemeral"),
+        )
+    return TextBlockParam(text=block.text, type="text")
 
 
 # Function used in bedrock
diff --git a/llama-index-integrations/llms/llama-index-llms-anthropic/pyproject.toml b/llama-index-integrations/llms/llama-index-llms-anthropic/pyproject.toml
index 51bc9a6d9..b7ae443cb 100644
--- a/llama-index-integrations/llms/llama-index-llms-anthropic/pyproject.toml
+++ b/llama-index-integrations/llms/llama-index-llms-anthropic/pyproject.toml
@@ -27,7 +27,7 @@ exclude = ["**/BUILD"]
 license = "MIT"
 name = "llama-index-llms-anthropic"
 readme = "README.md"
-version = "0.6.7"
+version = "0.6.8"
 
 [tool.poetry.dependencies]
 python = ">=3.9,<4.0"


Test patch:
diff --git a/llama-index-core/tests/node_parser/test_markdown.py b/llama-index-core/tests/node_parser/test_markdown.py
index c53d481b3..ffb7e69a2 100644
--- a/llama-index-core/tests/node_parser/test_markdown.py
+++ b/llama-index-core/tests/node_parser/test_markdown.py
@@ -25,6 +25,42 @@ Header 2 content
     assert splits[1].text == "# Header 2\nHeader 2 content"
 
 
+def test_header_splits_with_forwardslash() -> None:
+    markdown_parser = MarkdownNodeParser(
+        header_path_separator="\u203A"
+    )  # Unicode for "›", infrequently used char
+
+    splits = markdown_parser.get_nodes_from_documents(
+        [
+            Document(
+                text="""# Main Header
+
+Header 1 content
+
+## FAQ
+FAQ content
+
+### 24/7 Support
+Support content
+
+#### Contact info
+Contact info content
+    """
+            )
+        ]
+    )
+    assert len(splits) == 4
+    assert splits[0].metadata == {"header_path": "›"}
+    assert splits[1].metadata == {"header_path": "›Main Header›"}
+    assert splits[2].metadata == {"header_path": "›Main Header›FAQ›"}
+    assert splits[3].metadata == {"header_path": "›Main Header›FAQ›24/7 Support›"}
+
+    assert splits[0].text == "# Main Header\n\nHeader 1 content"
+    assert splits[1].text == "## FAQ\nFAQ content"
+    assert splits[2].text == "### 24/7 Support\nSupport content"
+    assert splits[3].text == "#### Contact info\nContact info content"
+
+
 def test_header_splits_with_indented_code_blocks() -> None:
     markdown_parser = MarkdownNodeParser()
 
diff --git a/llama-index-core/tests/workflow/test_workflow.py b/llama-index-core/tests/workflow/test_workflow.py
index 8dbf4de39..a64bb5162 100644
--- a/llama-index-core/tests/workflow/test_workflow.py
+++ b/llama-index-core/tests/workflow/test_workflow.py
@@ -618,17 +618,22 @@ async def test_workflow_context_to_dict(workflow):
     assert new_ctx._queues["start_step"].get_nowait().name == "test"
 
 
-@pytest.mark.asyncio()
-async def test_human_in_the_loop():
-    class HumanInTheLoopWorkflow(Workflow):
-        @step
-        async def step1(self, ev: StartEvent) -> InputRequiredEvent:
-            return InputRequiredEvent(prefix="Enter a number: ")
+class HumanInTheLoopWorkflow(Workflow):
+    @step
+    async def step1(self, ctx: Context, ev: StartEvent) -> InputRequiredEvent:
+        cur_runs = await ctx.get("step1_runs", default=0)
+        await ctx.set("step1_runs", cur_runs + 1)
+        return InputRequiredEvent(prefix="Enter a number: ")
 
-        @step
-        async def step2(self, ev: HumanResponseEvent) -> StopEvent:
-            return StopEvent(result=ev.response)
+    @step
+    async def step2(self, ctx: Context, ev: HumanResponseEvent) -> StopEvent:
+        cur_runs = await ctx.get("step2_runs", default=0)
+        await ctx.set("step2_runs", cur_runs + 1)
+        return StopEvent(result=ev.response)
 
+
+@pytest.mark.asyncio()
+async def test_human_in_the_loop():
     workflow = HumanInTheLoopWorkflow(timeout=1)
 
     # workflow should raise a timeout error because hitl only works with streaming
@@ -653,6 +658,34 @@ async def test_human_in_the_loop():
     assert final_result == "42"
 
 
+@pytest.mark.asyncio()
+async def test_human_in_the_loop_with_resume():
+    # workflow should work with streaming
+    workflow = HumanInTheLoopWorkflow()
+
+    handler: WorkflowHandler = workflow.run()
+    assert handler.ctx
+
+    ctx_dict = None
+    async for event in handler.stream_events():
+        if isinstance(event, InputRequiredEvent):
+            ctx_dict = handler.ctx.to_dict()
+            await handler.cancel_run()
+            break
+
+    new_handler = workflow.run(ctx=Context.from_dict(workflow, ctx_dict))
+    new_handler.ctx.send_event(HumanResponseEvent(response="42"))
+
+    final_result = await new_handler
+    assert final_result == "42"
+
+    # ensure the workflow ran each step once
+    step1_runs = await new_handler.ctx.get("step1_runs")
+    step2_runs = await new_handler.ctx.get("step2_runs")
+    assert step1_runs == 1
+    assert step2_runs == 1
+
+
 class DummyWorkflowForConcurrentRunsTest(Workflow):
     def __init__(self, **kwargs) -> None:
         super().__init__(**kwargs)