自主选择流程（人工介入工作流）¶

在某些工作流应用中，可能需要或要求人类参与其执行过程。例如，工作流的某个步骤可能需要人类专业知识或输入才能运行。另一种场景中，可能要求人类验证工作流的初始输出。

本笔记本将展示如何在工作流中实现人机协同模式。我们将构建一个"选择你自己的冒险"风格的故事生成工作流，其中LLM负责生成故事片段及潜在行动选项，而人类需从这些选项中进行选择。

使用大语言模型生成故事片段¶

在这里，我们将利用大语言模型（LLM）生成结构化输出的能力。我们的任务是让LLM根据先前生成的故事片段和行动选择，继续创作后续的故事段落。

In [ ]:

from typing import Any, List

from llama_index.llms.openai import OpenAI
from llama_index.core.bridge.pydantic import BaseModel, Field
from llama_index.core.prompts import PromptTemplate

from typing import Any, List from llama_index.llms.openai import OpenAI from llama_index.core.bridge.pydantic import BaseModel, Field from llama_index.core.prompts import PromptTemplate

In [ ]:

class Segment(BaseModel):
    """Data model for generating segments of a story."""

    plot: str = Field(
        description="The plot of the adventure for the current segment. The plot should be no longer than 3 sentences."
    )
    actions: List[str] = Field(
        default=[],
        description="The list of actions the protaganist can take that will shape the plot and actions of the next segment.",
    )

class Segment(BaseModel): """Data model for generating segments of a story.""" plot: str = Field( description="The plot of the adventure for the current segment. The plot should be no longer than 3 sentences." ) actions: List[str] = Field( default=[], description="The list of actions the protaganist can take that will shape the plot and actions of the next segment.", )

In [ ]:

SEGMENT_GENERATION_TEMPLATE = """
You are working with a human to create a story in the style of choose your own adventure.

The human is playing the role of the protaganist in the story which you are tasked to
help write. To create the story, we do it in steps, where each step produces a BLOCK.
Each BLOCK consists of a PLOT, a set of ACTIONS that the protaganist can take, and the
chosen ACTION. 

Below we attach the history of the adventure so far.

PREVIOUS BLOCKS:
---
{running_story}

Continue the story by generating the next block's PLOT and set of ACTIONs. If there are
no previous BLOCKs, start an interesting brand new story. Give the protaganist a name and an
interesting challenge to solve.


Use the provided data model to structure your output.
"""

SEGMENT_GENERATION_TEMPLATE = """ You are working with a human to create a story in the style of choose your own adventure. The human is playing the role of the protaganist in the story which you are tasked to help write. To create the story, we do it in steps, where each step produces a BLOCK. Each BLOCK consists of a PLOT, a set of ACTIONS that the protaganist can take, and the chosen ACTION. Below we attach the history of the adventure so far. PREVIOUS BLOCKS: --- {running_story} Continue the story by generating the next block's PLOT and set of ACTIONs. If there are no previous BLOCKs, start an interesting brand new story. Give the protaganist a name and an interesting challenge to solve. Use the provided data model to structure your output. """

In [ ]:

FINAL_SEGMENT_GENERATION_TEMPLATE = """
You are working with a human to create a story in the style of choose your own adventure.

The human is playing the role of the protaganist in the story which you are tasked to
help write. To create the story, we do it in steps, where each step produces a BLOCK.
Each BLOCK consists of a PLOT, a set of ACTIONS that the protaganist can take, and the
chosen ACTION. Below we attach the history of the adventure so far.

PREVIOUS BLOCKS:
---
{running_story}

The story is now coming to an end. With the previous blocks, wrap up the story with a
closing PLOT. Since it is a closing plot, DO NOT GENERATE a new set of actions.

Use the provided data model to structure your output.
"""

FINAL_SEGMENT_GENERATION_TEMPLATE = """ You are working with a human to create a story in the style of choose your own adventure. The human is playing the role of the protaganist in the story which you are tasked to help write. To create the story, we do it in steps, where each step produces a BLOCK. Each BLOCK consists of a PLOT, a set of ACTIONS that the protaganist can take, and the chosen ACTION. Below we attach the history of the adventure so far. PREVIOUS BLOCKS: --- {running_story} The story is now coming to an end. With the previous blocks, wrap up the story with a closing PLOT. Since it is a closing plot, DO NOT GENERATE a new set of actions. Use the provided data model to structure your output. """

In [ ]:

# Let's see an example segment
llm = OpenAI("gpt-4o")
segment = llm.structured_predict(
    Segment,
    PromptTemplate(SEGMENT_GENERATION_TEMPLATE),
    running_story="",
)

# Let's see an example segment llm = OpenAI("gpt-4o") segment = llm.structured_predict( Segment, PromptTemplate(SEGMENT_GENERATION_TEMPLATE), running_story="", )

In [ ]:

segment

segment

Out[ ]:

Segment(plot="In the bustling city of Eldoria, a young adventurer named Aric discovered a mysterious map hidden inside an old bookshop. The map hinted at a hidden treasure buried deep within the enchanted Whispering Woods. Intrigued and eager for adventure, Aric decided to follow the map's clues.", actions=['Follow the map to the Whispering Woods', 'Seek advice from the old bookshop owner', 'Gather supplies for the journey', 'Ignore the map and continue with daily life'])

整合先前片段¶

我们需要将故事片段拼接起来，并将其作为 running_story 的值传入提示词中。我们定义了一个 Block 数据类，用于存放 Segment 以及行动 choice。

In [ ]:

import uuid
from typing import Optional

BLOCK_TEMPLATE = """
BLOCK
===
PLOT: {plot}
ACTIONS: {actions}
CHOICE: {choice}
"""


class Block(BaseModel):
    id_: str = Field(default_factory=lambda: str(uuid.uuid4()))
    segment: Segment
    choice: Optional[str] = None
    block_template: str = BLOCK_TEMPLATE

    def __str__(self):
        return self.block_template.format(
            plot=self.segment.plot,
            actions=", ".join(self.segment.actions),
            choice=self.choice or "",
        )

import uuid from typing import Optional BLOCK_TEMPLATE = """ BLOCK === PLOT: {plot} ACTIONS: {actions} CHOICE: {choice} """ class Block(BaseModel): id_: str = Field(default_factory=lambda: str(uuid.uuid4())) segment: Segment choice: Optional[str] = None block_template: str = BLOCK_TEMPLATE def __str__(self): return self.block_template.format( plot=self.segment.plot, actions=", ".join(self.segment.actions), choice=self.choice or "", )

In [ ]:

block = Block(segment=segment)
print(block)

block = Block(segment=segment) print(block)

BLOCK
===
PLOT: In the bustling city of Eldoria, a young adventurer named Aric discovered a mysterious map hidden inside an old bookshop. The map hinted at a hidden treasure buried deep within the enchanted Whispering Woods. Intrigued and eager for adventure, Aric decided to follow the map's clues.
ACTIONS: Follow the map to the Whispering Woods, Seek advice from the old bookshop owner, Gather supplies for the journey, Ignore the map and continue with daily life
CHOICE: 

创建自主选择式工作流¶

该工作流将包含两个步骤，这两个步骤会循环执行，直到生成最大数量的步骤（即片段）。第一步将由大语言模型（LLM）创建一个新的Segment，该片段将用于生成新的故事Block。第二步会提示人类用户从新创建的Segment中指定的动作列表中选择他们的冒险路径。

In [ ]:

from llama_index.core.workflow import (
    Context,
    Event,
    StartEvent,
    StopEvent,
    Workflow,
    step,
)

from llama_index.core.workflow import ( Context, Event, StartEvent, StopEvent, Workflow, step, )

In [ ]:

class NewBlockEvent(Event):
    block: Block


class HumanChoiceEvent(Event):
    block_id: str

class NewBlockEvent(Event): block: Block class HumanChoiceEvent(Event): block_id: str

In [ ]:

class ChooseYourOwnAdventureWorkflow(Workflow):
    def __init__(self, max_steps: int = 3, **kwargs):
        super().__init__(**kwargs)
        self.llm = OpenAI("gpt-4o")
        self.max_steps = max_steps

    @step
    async def create_segment(
        self, ctx: Context, ev: StartEvent | HumanChoiceEvent
    ) -> NewBlockEvent | StopEvent:
        blocks = await ctx.store.get("blocks", [])
        running_story = "\n".join(str(b) for b in blocks)

        if len(blocks) < self.max_steps:
            new_segment = self.llm.structured_predict(
                Segment,
                PromptTemplate(SEGMENT_GENERATION_TEMPLATE),
                running_story=running_story,
            )
            new_block = Block(segment=new_segment)
            blocks.append(new_block)
            await ctx.store.set("blocks", blocks)
            return NewBlockEvent(block=new_block)
        else:
            final_segment = self.llm.structured_predict(
                Segment,
                PromptTemplate(FINAL_SEGMENT_GENERATION_TEMPLATE),
                running_story=running_story,
            )
            final_block = Block(segment=final_segment)
            blocks.append(final_block)
            return StopEvent(result=blocks)

    @step
    async def prompt_human(
        self, ctx: Context, ev: NewBlockEvent
    ) -> HumanChoiceEvent:
        block = ev.block

        # get human input
        human_prompt = f"\n===\n{ev.block.segment.plot}\n\n"
        human_prompt += "Choose your adventure:\n\n"
        human_prompt += "\n".join(ev.block.segment.actions)
        human_prompt += "\n\n"
        human_input = input(human_prompt)

        blocks = await ctx.store.get("blocks")
        block.choice = human_input
        blocks[-1] = block
        await ctx.store.set("block", blocks)

        return HumanChoiceEvent(block_id=ev.block.id_)

class ChooseYourOwnAdventureWorkflow(Workflow): def __init__(self, max_steps: int = 3, **kwargs): super().__init__(**kwargs) self.llm = OpenAI("gpt-4o") self.max_steps = max_steps @step async def create_segment( self, ctx: Context, ev: StartEvent | HumanChoiceEvent ) -> NewBlockEvent | StopEvent: blocks = await ctx.store.get("blocks", []) running_story = "\n".join(str(b) for b in blocks) if len(blocks) < self.max_steps: new_segment = self.llm.structured_predict( Segment, PromptTemplate(SEGMENT_GENERATION_TEMPLATE), running_story=running_story, ) new_block = Block(segment=new_segment) blocks.append(new_block) await ctx.store.set("blocks", blocks) return NewBlockEvent(block=new_block) else: final_segment = self.llm.structured_predict( Segment, PromptTemplate(FINAL_SEGMENT_GENERATION_TEMPLATE), running_story=running_story, ) final_block = Block(segment=final_segment) blocks.append(final_block) return StopEvent(result=blocks) @step async def prompt_human( self, ctx: Context, ev: NewBlockEvent ) -> HumanChoiceEvent: block = ev.block # get human input human_prompt = f"\n===\n{ev.block.segment.plot}\n\n" human_prompt += "Choose your adventure:\n\n" human_prompt += "\n".join(ev.block.segment.actions) human_prompt += "\n\n" human_input = input(human_prompt) blocks = await ctx.store.get("blocks") block.choice = human_input blocks[-1] = block await ctx.store.set("block", blocks) return HumanChoiceEvent(block_id=ev.block.id_)

运行工作流¶

由于工作流默认采用异步优先模式，这些操作在笔记本环境中都能顺畅运行。若在自主编写的代码中执行，当不存在已激活的异步事件循环时，需使用 asyncio.run() 来启动。

async def main():
    <async code>

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

In [ ]:

import nest_asyncio

nest_asyncio.apply()

import nest_asyncio nest_asyncio.apply()

In [ ]:

w = ChooseYourOwnAdventureWorkflow(timeout=None)

w = ChooseYourOwnAdventureWorkflow(timeout=None)

In [ ]:

result = await w.run()

result = await w.run()

打印最终故事¶

In [ ]:

final_story = "\n\n".join(b.segment.plot for b in result)
print(final_story)

final_story = "\n\n".join(b.segment.plot for b in result) print(final_story)

In the bustling city of Eldoria, a young adventurer named Aric discovered a mysterious map hidden in an old bookshop. The map hinted at a hidden treasure buried deep within the Whispering Woods, a place known for its eerie silence and ancient secrets. Determined to uncover the treasure, Aric set off on his journey, leaving the city behind.

Aric found a seasoned guide named Elara, who knew the Whispering Woods like the back of her hand. Elara agreed to help Aric, intrigued by the promise of hidden treasure. Together, they ventured into the forest, the map leading them to a fork in the path where the trees seemed to whisper secrets.

Elara examined the map closely and listened to the whispers of the trees. She suggested taking the left path, as it seemed to align with the ancient markings on the map. Trusting her expertise, Aric and Elara proceeded down the left path, where the forest grew denser and the air filled with an eerie stillness.

As Aric and Elara searched for hidden clues along the dense path, they stumbled upon an ancient stone altar covered in moss and vines. Upon closer inspection, they discovered a hidden compartment within the altar containing the long-lost treasure—a chest filled with gold, jewels, and ancient artifacts. With their mission complete, Aric and Elara returned to Eldoria, their bond strengthened by the adventure and their hearts filled with the thrill of discovery.

实现人机协同的其他方法¶

也可以通过创建专门用于收集人工输入的独立工作流，并利用嵌套工作流来实现"人在回路"机制。这种设计适用于需要将人工输入收集与工作流其他部分分离的场景，这正是使用llama-deploy部署嵌套工作流时会出现的情况。