LlamaIndexのイベント駆動型エージェントアーキテクチャ「Workflows」を試す

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

from llama_index.llms.openai import OpenAI

class OpenAIGenerator(Workflow):
    @step()
    async def generate(self, ev: StartEvent) -> StopEvent:
        query = ev.get("query")
        llm = OpenAI()
        response = await llm.acomplete(query)
        return StopEvent(result=str(response))

w = OpenAIGenerator(timeout=10, verbose=False)
result = await w.run(query="What's LlamaIndex?")
print(result)

class ExtractionDone(Event):
    output: str
    passage: str


class ValidationErrorEvent(Event):
    error: str
    wrong_output: str
    passage: str
    
    
class ReflectionWorkflow(Workflow):
    @step()
    async def extract(
        self, ev: StartEvent | ValidationErrorEvent
    ) -> StopEvent | ExtractionDone:
        if isinstance(ev, StartEvent):
            passage = ev.get("passage")
            if not passage:
                return StopEvent(result="Please provide some text in input")
            reflection_prompt = ""
        elif isinstance(ev, ValidationErrorEvent):
            passage = ev.passage
            reflection_prompt = REFLECTION_PROMPT.format(
                wrong_answer=ev.wrong_output, error=ev.error
            )

        llm = Ollama(model="llama3", request_timeout=30)
        prompt = EXTRACTION_PROMPT.format(
            passage=passage, schema=CarCollection.schema_json()
        )
        if reflection_prompt:
            prompt += reflection_prompt

        output = await llm.acomplete(prompt)

        return ExtractionDone(output=str(output), passage=passage)

    @step()
    async def validate(
        self, ev: ExtractionDone
    ) -> StopEvent | ValidationErrorEvent:
        try:
            json.loads(ev.output)
        except Exception as e:
            print("Validation failed, retrying...")
            return ValidationErrorEvent(
                error=str(e), wrong_output=ev.output, passage=ev.passage
            )

        return StopEvent(result=ev.output)

w = ReflectionWorkflow(timeout=60, verbose=True)
result = await w.run(
    passage="There are two cars available: a Fiat Panda with 45Hp and a Honda Civic with 330Hp."
)
print(result)

class RAGWorkflow(Workflow):
    @step(pass_context=True)
    async def ingest(self, ctx: Context, ev: StartEvent) -> Optional[StopEvent]:
        dataset_name = ev.get("dataset")
        _, documents = download_llama_dataset(dsname, "./data")
        ctx.data["INDEX"] = VectorStoreIndex.from_documents(documents=documents)
        return StopEvent(result=f"Indexed {len(documents)} documents.")
        
    ...

class MyWorkflow(RAGWorkflow):
    @step(pass_context=True)
    def rerank(
        self, ctx: Context, ev: Union[RetrieverEvent, StartEvent]
    ) -> Optional[QueryResult]:
        # my custom reranking logic here
        
 
w = MyWorkflow(timeout=60, verbose=True)
result = await w.run(query="Who is Paul Graham?")

!pip install llama-index llama-index-callbacks-arize-phoenix

!pip freeze | egrep -i "llama-|arize|open-"

arize-phoenix==4.19.0
arize-phoenix-evals==0.14.1
llama-cloud==0.0.11
llama-index==0.10.59
llama-index-agent-openai==0.2.9
llama-index-callbacks-arize-phoenix==0.1.6
llama-index-cli==0.1.13
llama-index-core==0.10.59
llama-index-embeddings-openai==0.1.11
llama-index-indices-managed-llama-cloud==0.2.7
llama-index-legacy==0.9.48
llama-index-llms-openai==0.1.27
llama-index-multi-modal-llms-openai==0.1.8
llama-index-program-openai==0.1.7
llama-index-question-gen-openai==0.1.3
llama-index-readers-file==0.1.32
llama-index-readers-llama-parse==0.1.6
llama-parse==0.4.9
openinference-instrumentation-llama-index==2.2.1

import llama_index.core
import os
from google.colab import userdata

PHOENIX_API_KEY = userdata.get('PHOENIX_API_KEY')
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"api_key={PHOENIX_API_KEY}"
llama_index.core.set_global_handler(
    "arize_phoenix", endpoint="https://llamatrace.com/v1/traces"
)

import os
from google.colab import userdata

os.environ["OPENAI_API_KEY"] = userdata.get('OPENAI_API_KEY')

import nest_asyncio

nest_asyncio.apply()

from llama_index.core.workflow import (
    Event,
    StartEvent,
    StopEvent,
    Workflow,
    step,
)
from llama_index.llms.openai import OpenAI


class JokeEvent(Event):
    joke: str


class JokeFlow(Workflow):
    llm = OpenAI(model="gpt-4o-mini")

    @step()
    async def generate_joke(self, ev: StartEvent) -> JokeEvent:
        topic = ev.get("topic")

        prompt = f"{topic}についてのジョークを生成して。"
        response = await self.llm.acomplete(prompt)
        return JokeEvent(joke=str(response))

    @step()
    async def critique_joke(self, ev: JokeEvent) -> StopEvent:
        joke = ev.joke

        prompt = f"次のジョークを徹底的に分析し、批評して: {joke}"
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=str(response))


w = JokeFlow(timeout=60, verbose=False)
result = await w.run(topic="パイレーツ")
print(str(result))

このジョークは、言葉遊びを基にしたユーモアの一例です。以下に、ジョークの構造や要素を分析し、批評します。

### 構造分析

1. **設定**: 海賊というキャラクターが登場します。海賊は一般的に冒険や航海を象徴する存在であり、コンパスは航海において重要な道具です。この設定は、聴衆にとって親しみやすく、期待感を持たせます。

2. **質問形式**: 「なぜ海賊はいつもコンパスを持っているの？」という質問は、聴衆の興味を引き、答えを期待させます。この形式は、ジョークの効果を高めるためによく使われます。

3. **ダブルミーニング**: ジョークの核心は「北を見失いたくない」という部分です。「失う」という言葉が二重の意味を持っています。一つは物理的な方向を失うこと、もう一つは「北」を「失う」という言葉遊びです。この言葉のかけ方が、ジョークのユーモアの源泉です。

### 批評

- **ユーモアの質**: このジョークは、シンプルでわかりやすい言葉遊びを用いており、聴衆にすぐに理解されやすいです。しかし、言葉遊びのレベルは比較的低いため、特に高度なユーモアを求める人には物足りないかもしれません。

- **文化的背景**: 海賊やコンパスというテーマは、特に西洋文化において広く知られているため、国や文化によっては共感を得やすいです。ただし、海賊文化に馴染みのない人には、ジョークの面白さが伝わりにくい可能性があります。

- **オリジナリティ**: 言葉遊び自体は一般的な手法であり、特に新しいアイデアや視点を提供しているわけではありません。そのため、オリジナリティに欠けると感じる人もいるでしょう。

- **感情的な反応**: 聴衆がこのジョークを聞いたとき、軽い笑いや微笑みを引き起こすことが期待されますが、深い感情的な反応を引き起こすことは難しいでしょう。

### 結論

このジョークは、シンプルで親しみやすい言葉遊びを用いたものであり、特に子供やカジュアルな場面での笑いを狙ったものと言えます。ユーモアの質やオリジナリティには限界がありますが、軽い気持ちで楽しむには適した内容です。

class JokeEvent(Event):
    joke: str

class JokeFlow(Workflow):
    llm = OpenAI(model="gpt-4o-mini")
    (snip)

class JokeFlow(Workflow):
    llm = OpenAI(model="gpt-4o-mini")

    @step()
    async def generate_joke(self, ev: StartEvent) -> JokeEvent:
        topic = ev.get("topic")

        prompt = f"{topic}についてのジョークを生成して。"
        response = await self.llm.acomplete(prompt)
        return JokeEvent(joke=str(response))

        (snip)

class JokeFlow(Workflow):
    (snip)

    @step()
    async def critique_joke(self, ev: JokeEvent) -> StopEvent:
        joke = ev.joke

        prompt = f"次のジョークを徹底的に分析し、批評して: {joke}"
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=str(response))

w = JokeFlow(timeout=60, verbose=False)
result = await w.run(topic="パイレーツ")
print(str(result))

Running step generate_joke
Step generate_joke produced event JokeEvent
Running step critique_joke
Step critique_joke produced event StopEvent

!pip install pyvis

from llama_index.core.workflow import (
    # 注: ドキュメントには`draw_all_possible_paths`とあるが、おそらく間違い。
    draw_all_possible_flows,
    draw_most_recent_execution,
)

# 定義したワークフロークラスに対して可視化する場合は`draw_all_possible_flows`を使う
draw_all_possible_flows(JokeFlow, filename="joke_flow_all.html")

# 実行したワークフローに対して可視化する場合は`draw_most_recent_execution`を使う
w = JokeFlow(timeout=60, verbose=True)
result = await w.run(topic="パイレーツ")
draw_most_recent_execution(w, filename="joke_flow_recent.html")

from llama_index.core.workflow import Context


@step(pass_context=True)
async def query(self, ctx: Context, ev: QueryEvent) -> StopEvent:
    # コンテキストからデータを取り出す
    query = ctx.data.get("query")

    # コンテキストにデータを保存する
    ctx["key"] = "val"

    result = run_query(query)
    return StopEvent(result=result)

@step(pass_context=True)
async def query(
    self, ctx: Context, ev: QueryEvent | RetrieveEvent
) -> StopEvent | None:
    data = ctx.collect_events(evm[QueryEvent, RetrieveEvent])
    # 実行してよいかをチェックする
    if data is None:
        return None

    # バッファリングされたイベントを使う
    print(data[0])  # QueryEvent
    print(data[1])  # RetrieveEvent

import time
import asyncio

w = JokeFlow(timeout=60, verbose=True)

# ワークフローを開始
print("開始")
await w.run_step(topic="パイレーツ")

# 完了するまで繰り返す
while not w.is_done():
    print("ステップ継続")
    await w.run_step()
    await asyncio.sleep(10)    

print("完了")
# 最終結果を取得
result = w.get_result()

print("===== 出力 =====")
print(result[:50] + "...")

開始
Running step generate_joke
ステップ継続
Step generate_joke produced event JokeEvent
Running step critique_joke
ステップ継続
Step critique_joke produced event StopEvent
ステップ継続
完了
===== 出力 =====
このジョークは、言葉遊びとダブルミーニングを利用したユーモアの一例です。以下に、ジョークの構造や要素...

from llama_index.core.workflow import (
    Event,
    StartEvent,
    StopEvent,
    Workflow,
    step,
)
from llama_index.llms.openai import OpenAI


class JokeEvent(Event):
    joke: str


llm = OpenAI(model="gpt-4o-mini")

joke_flow = Workflow(timeout=60, verbose=True)


@step(workflow=joke_flow)
async def generate_joke(ev: StartEvent) -> JokeEvent:
    # ドキュメントだと`ev.topic`になっているが、そんなattirbuteはないと言われる
    topic = ev.get("topic")

    prompt = f"{topic}についてのジョークを生成して。"
    response = await llm.acomplete(prompt)
    return JokeEvent(joke=str(response))


@step(workflow=joke_flow)
async def critique_joke(ev: JokeEvent) -> StopEvent:
    # こちらはイベント定義があるので、`ev.joke`で取り出せる
    joke = ev.joke

    prompt = f"次のジョークを徹底的に分析し、批評して: {joke}"

    response = await llm.acomplete(prompt)
    return StopEvent(result=str(response))

result = await joke_flow.run(topic="パイレーツ")
print(result)

このジョークは、海賊の特徴や文化に基づいた言葉遊びを利用しています。以下に、ジョークの構造や要素を分析し、批評します。

### 構造分析

1. **質問形式**: ジョークは「なぜ海賊はいつも『アーrrr！』と言うの？」という質問から始まります。この形式は、聞き手の興味を引き、期待感を高める効果があります。

2. **答え**: 「だって、彼らは『海の男』だからさ！」という答えは、海賊のイメージを強調しています。「アーrrr！」という言葉は、海賊の典型的なセリフとして知られており、海賊文化の象徴的な表現です。

### 言葉遊び

- **言葉の響き**: 「アーrrr！」という表現は、海賊の粗野で無骨なイメージを想起させます。この音の響きが、海賊のキャラクターを強調し、ユーモアを生み出しています。

- **「海の男」**: このフレーズは、海賊が海で生活し、冒険をする男たちであることを示しています。ここでの「男」という言葉は、海賊の勇敢さや自由さを象徴しています。

### 批評

1. **ユーモアの質**: このジョークはシンプルで、子供から大人まで楽しめるタイプのユーモアです。特に、海賊のキャラクターが好きな人や、海賊映画や物語に親しんでいる人には受け入れられやすいでしょう。

2. **文化的背景**: 海賊というテーマは、映画や文学で広く知られているため、聞き手が共通の理解を持っていることが前提となっています。このため、特定の文化や背景を持つ人々には特に響くでしょう。

3. **深みの欠如**: 一方で、このジョークは非常にシンプルであり、深い意味や複雑な構造を持っていないため、あまりに直球すぎて物足りなさを感じる人もいるかもしれません。特に、より洗練されたユーモアを求める人には物足りないかもしれません。

### 結論

このジョークは、海賊のイメージを利用したシンプルで親しみやすいユーモアです。聞き手の文化的背景や興味によって受け取り方が変わるため、特定の状況や聴衆においては非常に効果的です。しかし、深みや複雑さを求める人には物足りないと感じられる可能性もあります。全体として、軽い気持ちで楽しむには良いジョークと言えるでしょう。

from pathlib import Path
import requests
import re

def replace_heading(match):
    level = len(match.group(1))
    return '#' * level + ' ' + match.group(2).strip()

# Wikipediaからのデータ読み込み
wiki_titles = ["オグリキャップ"]
for title in wiki_titles:
    response = requests.get(
        "https://ja.wikipedia.org/w/api.php",
        params={
            "action": "query",
            "format": "json",
            "titles": title,
            "prop": "extracts",
            # 'exintro': True,
            "explaintext": True,
        },
    ).json()
    page = next(iter(response["query"]["pages"].values()))
    wiki_text = f"# {title}\n\n## 概要\n\n"
    wiki_text += page["extract"]

    wiki_text = re.sub(r"(=+)([^=]+)\1", replace_heading, wiki_text)
    wiki_text = re.sub(r"\t+", "", wiki_text)
    wiki_text = re.sub(r"\n{3,}", "\n\n", wiki_text)
    data_path = Path("data")
    if not data_path.exists():
        Path.mkdir(data_path)

    # markdown(.md)ファイルとして出力
    with open(data_path / f"{title}.txt", "w") as fp:
        fp.write(wiki_text)

from llama_index.core.workflow import Event
from llama_index.core.schema import NodeWithScore


class RetrieverEvent(Event):
    """検索実行の結果"""

    nodes: list[NodeWithScore]


class RerankEvent(Event):
    """検索されたノードのりランキング結果"""

    nodes: list[NodeWithScore]

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex
from llama_index.core.response_synthesizers import CompactAndRefine
from llama_index.core.postprocessor.llm_rerank import LLMRerank
from llama_index.core.workflow import (
    Context,
    Workflow,
    StartEvent,
    StopEvent,
    step,
)

from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding


class RAGWorkflow(Workflow):
    @step(pass_context=True)
    async def ingest(self, ctx: Context, ev: StartEvent) -> StopEvent | None:
        """StartEventに`dir_name`が渡されるとトリガーされる、RAGのIndexing Stage用のエントリーポイント"""
        dirname = ev.get("dirname")
        if not dirname:
            return None

        documents = SimpleDirectoryReader(dirname).load_data()
        # ドキュメントからインデックスを作成して、グローバルコンテキストに保存
        ctx.data["index"] = VectorStoreIndex.from_documents(
            documents=documents,
            embed_model=OpenAIEmbedding(model_name="text-embedding-3-small"),
        )
        return StopEvent(result=f"{len(documents)}件のドキュメントをインデックス化しました。")

    @step(pass_context=True)
    async def retrieve(
        self, ctx: Context, ev: StartEvent
    ) -> RetrieverEvent | None:
        """StartEventに`query`が渡されるとトリガーされる、RAGのQuerying Stage用のエントリーポイント"""
        query = ev.get("query")
        if not query:
            return None

        print(f"データベース検索クエリ: {query}")

        # グローバルコンテキストにクエリを保存
        ctx.data["query"] = query

        # グローバルコンテキストからインデックスを取得
        index = ctx.data.get("index")
        if index is None:
            print("インデックスは空です。クエリを行う前に、ドキュメントのインデックスを作成してください！")
            return None

        retriever = index.as_retriever(similarity_top_k=2)
        nodes = retriever.retrieve(query)
        print(f"{len(nodes)}件のノードを取得しました。")
        return RetrieverEvent(nodes=nodes)

    @step(pass_context=True)
    async def rerank(self, ctx: Context, ev: RetrieverEvent) -> RerankEvent:
        # ノードをリランキング
        ranker = LLMRerank(
            choice_batch_size=5, top_n=3, llm=OpenAI(model="gpt-4o-mini")
        )
        print(ctx.data.get("query"), flush=True)
        new_nodes = ranker.postprocess_nodes(
            ev.nodes, query_str=ctx.data.get("query")  # グローバルコンテキストからクエリを取得
        )
        print(f"{len(new_nodes)}件のノードをリランキングしました。")
        return RerankEvent(nodes=new_nodes)

    @step(pass_context=True)
    async def synthesize(self, ctx: Context, ev: RerankEvent) -> StopEvent:
        """リランキングされたノードを使って、ストリーミングレスポンスを返す"""
        llm = OpenAI(model="gpt-4o-mini")
        summarizer = CompactAndRefine(llm=llm, streaming=True, verbose=True)
        # グローバルコンテキストからクエリを取得
        query = ctx.data.get("query")

        response = await summarizer.asynthesize(query, nodes=ev.nodes)
        return StopEvent(result=response)

# ワークフローインスタンスを生成
w = RAGWorkflow()

# ワークフローを実行して、インデックスを作成
await w.run(dirname="data")

1件のドキュメントをインデックス化しました。

# ワークフローを実行して、クエリを投げる
result = await w.run(query="オグリキャップの血統について教えて。")

async for chunk in result.async_response_gen():
    print(chunk, end="", flush=True)

データベース検索クエリ: オグリキャップの血統について教えて。
2件のノードを取得しました。
オグリキャップの血統について教えて。
2件のノードをリランキングしました。
オグリキャップの血統は、父がダンシングキャップで、母がホワイトナルビーです。ダンシングキャップの種牡馬成績はあまり優れていなかったため、オグリキャップは「突然変異で生まれた」とか「ネイティヴダンサーの隔世遺伝で生まれた競走馬」とも言われています。しかし、血統評論家はダンシングキャップを「一発ある血統」と評価し、ネイティヴダンサー系の種牡馬が時折大物を出すことから、オグリキャップもそのような可能性を秘めていたと分析しています。

母のホワイトナルビーは現役時代に笠松で4勝を挙げ、産駒は全て競馬で勝利を収めています。5代母のクインナルビーは1953年の天皇賞（秋）を制した実績があります。オグリキャップには兄弟もおり、オグリローマンは1994年の桜花賞優勝馬、オグリイチバンはシンジケートを組まれて種牡馬となりました。また、オグリトウショウもオグリキャップの活躍を受けて誕生し、競走馬引退後に種牡馬となっています。

from llama_index.core.workflow import (
    draw_all_possible_flows,
    draw_most_recent_execution,
)

draw_all_possible_flows(RAGWorkflow, filename="rag_workflow_all.html")
draw_most_recent_execution(w, filename="jrag_workflow_recent.html")

from llama_index.core.llms import ChatMessage
from llama_index.core.tools import ToolSelection, ToolOutput
from llama_index.core.workflow import Event


class InputEvent(Event):
    input: list[ChatMessage]


class ToolCallEvent(Event):
    tool_calls: list[ToolSelection]

# ドキュメントでは定義されているが、実際には使用されていない
#class FunctionOutputEvent(Event):
#    output: ToolOutput

from typing import Any, List

from llama_index.core.llms.function_calling import FunctionCallingLLM
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.tools.types import BaseTool
from llama_index.core.workflow import Workflow, StartEvent, StopEvent, step


class FuncationCallingAgent(Workflow):
    def __init__(
        self,
        *args: Any,
        llm: FunctionCallingLLM | None = None,
        tools: List[BaseTool] | None = None,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)
        
        self.tools = tools or []

        self.llm = llm or OpenAI()
        assert self.llm.metadata.is_function_calling_model

        self.memory = ChatMemoryBuffer.from_defaults(llm=llm)

        # toolの実行結果をソースとして保持する
        self.sources = []

    @step()
    async def prepare_chat_history(self, ev: StartEvent) -> InputEvent:
        # ソースを削除
        self.sources = []

        # ユーザの入力を取得
        # ドキュメントではev.inputだけど、
        # `'StartEvent' object has no attribute 'input'`
        # となってしまう。
        user_input = ev.get("input")
        user_msg = ChatMessage(role="user", content=user_input)
        self.memory.put(user_msg)

        # 会話履歴を取得
        chat_history = self.memory.get()
        return InputEvent(input=chat_history)

    @step()
    async def handle_llm_input(
        self, ev: InputEvent
    ) -> ToolCallEvent | StopEvent:
        chat_history = ev.input  
        # （クエリとツール実行結果を含む）会話履歴をLLMに投げる
        response = await self.llm.achat_with_tools(
            self.tools, chat_history=chat_history
        )
        self.memory.put(response.message)

        # LLMからのレスポンスからtool callを取り出す
        tool_calls = self.llm.get_tool_calls_from_response(
            response, error_on_no_tool_call=False
        )

        if not tool_calls:
            # tool callがなければ、StopEvnetを呼び出して最終解答を出力へ
            return StopEvent(
                result={"response": response, "sources": [*self.sources]}
            )
        else:
            # tool callがあれば、ToolCallEventを呼び出してツール実行へ
             return ToolCallEvent(tool_calls=tool_calls)

    @step()
    async def handle_tool_calls(self, ev: ToolCallEvent) -> InputEvent:
        tool_calls = ev.tool_calls
        tools_by_name = {tool.metadata.get_name(): tool for tool in self.tools}

        tool_msgs = []

        # ツールの呼び出し
        for tool_call in tool_calls:
            tool = tools_by_name.get(tool_call.tool_name)
            additional_kwargs = {
                "tool_call_id": tool_call.tool_id,
                "name": tool.metadata.get_name(),
            }
            if not tool:
                tool_msgs.append(
                    ChatMessage(
                        role="tool",
                        content=f"ツール「{tool_call.tool_name}」は存在しません。",
                        additional_kwargs=additional_kwargs,
                    )
                )
                continue

            # ツールの実行
            try:
                tool_output = tool(**tool_call.tool_kwargs)
                self.sources.append(tool_output)
                tool_msgs.append(
                    ChatMessage(
                        role="tool",
                        content=tool_output.content,
                        additional_kwargs=additional_kwargs,
                    )
                )
            except Exception as e:
                tool_msgs.append(
                    ChatMessage(
                        role="tool",
                        content=f"ツール呼び出しでエラーが発生しました: {e}",
                        additional_kwargs=additional_kwargs,
                    )
                )

        # ツール実行結果を会話履歴に追加
        for msg in tool_msgs:
            self.memory.put(msg)

        chat_history = self.memory.get()
        # 会話履歴をInputEventに渡す
        return InputEvent(input=chat_history)

from llama_index.core.workflow import (
    draw_all_possible_flows,
    draw_most_recent_execution,
)

draw_all_possible_flows(FuncationCallingAgent, filename="function_calling_agent_workflow_all.html")

from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI


def add(x: int, y: int) -> int:
    """2つの数値を加算する便利な関数"""
    return x + y


def multiply(x: int, y: int) -> int:
    """2つの数値を乗算する便利な関数"""
    return x * y


tools = [
    FunctionTool.from_defaults(add),
    FunctionTool.from_defaults(multiply),
]

agent = FuncationCallingAgent(
    llm=OpenAI(model="gpt-4o-mini"), tools=tools, timeout=120, verbose=True
)

ret = await agent.run(input="こんにちは！")
print(ret["response"])

Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event StopEvent
assistant: こんにちは！今日はどんなことをお手伝いできますか？

draw_most_recent_execution(agent, filename="function_calling_agent_workflow_recent.html")

ret = await agent.run(input="(2123 + 2321) * 312 は?")
print(ret["response"])

Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event ToolCallEvent
Running step handle_tool_calls
Step handle_tool_calls produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event ToolCallEvent
Running step handle_tool_calls
Step handle_tool_calls produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event StopEvent
assistant: 計算の結果、(2123 + 2321) * 312 は 1,386,528 です。何か他にお手伝いできることはありますか？

draw_most_recent_execution(agent, filename="function_calling_agent_workflow_recent.html")

ret = await agent.run(input="それを2で割って。")
print(ret["response"])

Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event ToolCallEvent
Running step handle_tool_calls
Step handle_tool_calls produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event ToolCallEvent
Running step handle_tool_calls
Step handle_tool_calls produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event StopEvent
assistant: 1,386,528 を 2 で割ると 693,264 になります。他に何かお手伝いできることはありますか？

print(ret["sources"])

[ToolOutput(content='2773056', tool_name='multiply', raw_input={'args': (), 'kwargs': {'x': 1386528, 'y': 2}}, raw_output=2773056, is_error=False), ToolOutput(content='693264.0', tool_name='multiply', raw_input={'args': (), 'kwargs': {'x': 1386528, 'y': 0.5}}, raw_output=693264.0, is_error=False)]

from llama_index.core.llms import ChatMessage
from llama_index.core.tools import ToolSelection, ToolOutput
from llama_index.core.workflow import Event


class PrepEvent(Event):
    pass


class InputEvent(Event):
    input: list[ChatMessage]


class ToolCallEvent(Event):
    tool_calls: list[ToolSelection]


# ドキュメントのコードでは定義されているが、実際には使用されていない
#class FunctionOutputEvent(Event):
#    output: ToolOutput

from typing import Any, List

from llama_index.core.agent.react import ReActChatFormatter, ReActOutputParser
from llama_index.core.agent.react.types import (
    ActionReasoningStep,
    ObservationReasoningStep,
)
from llama_index.core.llms.llm import LLM
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.tools.types import BaseTool
from llama_index.core.workflow import (
    Context,
    Workflow,
    StartEvent,
    StopEvent,
    step,
)
from llama_index.llms.openai import OpenAI


class ReActAgent(Workflow):
    def __init__(
        self,
        *args: Any,
        llm: LLM | None = None,
        tools: list[BaseTool] | None = None,
        extra_context: str | None = None,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)
        self.tools = tools or []

        self.llm = llm or OpenAI()

        self.memory = ChatMemoryBuffer.from_defaults(llm=llm)
        self.formatter = ReActChatFormatter(context=extra_context or "")
        self.output_parser = ReActOutputParser()
        self.sources = []

    @step(pass_context=True)
    async def new_user_msg(self, ctx: Context, ev: StartEvent) -> PrepEvent:
        # ソースをクリア
        self.sources = []

        # ユーザの入力を取得
        # ドキュメントではev.inputだけど、
        # `'StartEvent' object has no attribute 'input'`
        # となってしまう。
        user_input = ev.get("input")
        user_msg = ChatMessage(role="user", content=user_input)
        self.memory.put(user_msg)

        # 現在の推論状態をクリア
        ctx.data["current_reasoning"] = []

        return PrepEvent()

    @step(pass_context=True)
    async def prepare_chat_history(
        self, ctx: Context, ev: PrepEvent
    ) -> InputEvent:
        # 会話履歴を取得
        chat_history = self.memory.get()
        current_reasoning = ctx.data.get("current_reasoning", [])
        # ReActの推論プロセスの状態を踏まえて、ReActプロンプトを生成
        llm_input = self.formatter.format(
            self.tools, chat_history, current_reasoning=current_reasoning
        )
        return InputEvent(input=llm_input)

    @step(pass_context=True)
    async def handle_llm_input(
        self, ctx: Context, ev: InputEvent
    ) -> ToolCallEvent | StopEvent:
        chat_history = ev.input

        response = await self.llm.achat(chat_history)

        # ReActの推論プロセスの状態により分岐
        try:
            reasoning_step = self.output_parser.parse(response.message.content)
            ctx.data.get("current_reasoning", []).append(reasoning_step)
            if reasoning_step.is_done:
                # 推論プロセスが完了している場合
                self.memory.put(
                    ChatMessage(
                        role="assistant", content=reasoning_step.response
                    )
                )
                # StopEventを発火（して最終回答を出力）
                return StopEvent(
                    result={
                        "response": reasoning_step.response,
                        "sources": [*self.sources],
                        "reasoning": ctx.data.get("current_reasoning", []),
                    }
                )
            elif isinstance(reasoning_step, ActionReasoningStep):
                # 推論プロセスが"Action"の場合はToolEventを発火（してツール実行）
                tool_name = reasoning_step.action
                tool_args = reasoning_step.action_input
                return ToolCallEvent(
                    tool_calls=[
                        ToolSelection(
                            tool_id="fake",
                            tool_name=tool_name,
                            tool_kwargs=tool_args,
                        )
                    ]
                )
        except Exception as e:
            ctx.data.get("current_reasoning", []).append(
                ObservationReasoningStep(
                    observation=f"推論結果のパースに失敗しました: {e}"
                )
            )

        # tool callがレスポンスに含まれない、もしくはReAct推論プロセスが最終回答ではないなら、PrepEventでループさせる
        return PrepEvent()

    @step(pass_context=True)
    async def handle_tool_calls(
        self, ctx: Context, ev: ToolCallEvent
    ) -> PrepEvent:
        tool_calls = ev.tool_calls
        tools_by_name = {tool.metadata.get_name(): tool for tool in self.tools}

        # ツールの実行
        for tool_call in tool_calls:
            tool = tools_by_name.get(tool_call.tool_name)
            if not tool:
                ctx.data.get("current_reasoning", []).append(
                    ObservationReasoningStep(
                        observation=f"ツール「{tool_call.tool_name}」は存在しません。"
                    )
                )
                continue

            try:
                tool_output = tool(**tool_call.tool_kwargs)
                self.sources.append(tool_output)
                ctx.data.get("current_reasoning", []).append(
                    ObservationReasoningStep(observation=tool_output.content)
                )
            except Exception as e:
                ctx.data.get("current_reasoning", []).append(
                    ObservationReasoningStep(
                        observation=f"Error calling tool {tool.metadata.get_name()}: {e}"
                    )
                )

        # 次の繰り返しへ
        return PrepEvent()

from llama_index.core.workflow import (
    draw_all_possible_flows,
    draw_most_recent_execution,
)

draw_all_possible_flows(ReActAgent, filename="react_agent_workflow_all.html")

from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI


def add(x: int, y: int) -> int:
    """2つの数値を加算する便利な関数"""
    return x + y


def multiply(x: int, y: int) -> int:
    """2つの数値を乗算する便利な関数"""
    return x * y


tools = [
    FunctionTool.from_defaults(add),
    FunctionTool.from_defaults(multiply),
]

agent = ReActAgent(
    llm=OpenAI(model="gpt-4o-mini"), tools=tools, timeout=120, verbose=True
)

ret = await agent.run(input="こんにちは!")
print(ret["response"])

Running step new_user_msg
Step new_user_msg produced event PrepEvent
Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event StopEvent
こんにちは！どのようにお手伝いできますか？

ret = await agent.run(input="(2123 + 2321) * 312 は？")
print(ret["response"])

Running step new_user_msg
Step new_user_msg produced event PrepEvent
Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event ToolCallEvent
Running step handle_tool_calls
Step handle_tool_calls produced event PrepEvent
Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event ToolCallEvent
Running step handle_tool_calls
Step handle_tool_calls produced event PrepEvent
Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event StopEvent
(2123 + 2321) * 312 の結果は1386528です。

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


class TopicEvent(Event):
    pass

class AdvantageEvent(Event):
    advantage: str


class DisadvantageEvent(Event):
    disadvantage: str


class EvaluationFlow(Workflow):
    llm = OpenAI(model="gpt-4o-mini")

    @step(pass_context=True)
    async def query(self, ctx: Context, ev: StartEvent) -> TopicEvent:
        topic = ev.get("topic")
        ctx.data["topic"] = topic
        return TopicEvent()

    @step(pass_context=True)
    async def generate_advantage(self, ctx: Context, ev: TopicEvent) -> AdvantageEvent:
        topic = ctx.data["topic"]
        
        prompt = f"{topic}についてのメリットをリストアップして"
        response = await self.llm.acomplete(prompt)
        return AdvantageEvent(advantage=str(response))

    @step(pass_context=True)
    async def generate_disadvantage(self, ctx: Context, ev: TopicEvent) -> DisadvantageEvent:
        topic = ctx.data["topic"]

        prompt = f"{topic}についてのデメリットをリストアップして"
        response = await self.llm.acomplete(prompt)
        return DisadvantageEvent(disadvantage=str(response))

    @step(pass_context=True)
    async def evaluate(self, ctx: Context, ev: AdvantageEvent | DisadvantageEvent ) -> StopEvent:
        data = ctx.collect_events(ev, [AdvantageEvent, DisadvantageEvent])
        if data is None:
            return None

        topic = ctx.data["topic"]
        advantage_event, disadvantage_event = data

        advantage = advantage_event.advantage
        disadvantage = disadvantage_event.disadvantage

        prompt = f"""\
{topic}についてメリット・デメリットを比較してどちらのほうが有利か判断して。

メリット:
{advantage}

デメリット:
{disadvantage}

判定: 
"""
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=str(response))


w = EvaluationFlow(timeout=60, verbose=True)
result = await w.run(topic="円高")
print(str(result))

Running step query
Step query produced event TopicEvent
Running step generate_advantage
Running step generate_disadvantage
Step generate_disadvantage produced event DisadvantageEvent
Running step evaluate
Step evaluate produced no event
Step generate_advantage produced event AdvantageEvent
Running step evaluate
Step evaluate produced event StopEvent

from llama_index.core.workflow import (
    draw_all_possible_flows,
    draw_most_recent_execution,
)

draw_all_possible_flows(EvaluationFlow, filename="evaluate_flow_all.html")
draw_most_recent_execution(w, filename="evaluate_flow_recent.html")

from llama_index.core.workflow import step, Context, Event, Workflow


class MyEvent(Event):
    pass


class MyEventResult(Event):
    result: str


class GatherEvent(Event):
    pass


class MyWorkflow(Workflow):
    @step()
    async def dispatch_step(self, ev: StartEvent) -> MyEvent | GatherEvent:
        self.send_event(MyEvent())
        self.send_event(MyEvent())

        return GatherEvent()

    @step()
    async def handle_my_event(self, ev: MyEvent) -> MyEventResult:
        return MyEventResult(result="result")

    @step(pass_context=True)
    async def gather(
        self, ctx: Context, ev: GatherEvent | MyEventResult
    ) -> StopEvent | None:
        # wait for events to finish
        events = ctx.collect_events([MyEventResult, MyEventResult])
        if not events:
            return None

        return StopEvent(result=events)

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



class AdvantageEvent(Event):
    topic: str


class DisadvantageEvent(Event):
    topic: str


class AdvantageResultEvent(Event):
    result: str


class DisadvantageResultEvent(Event):
    result: str


class GatherEvent(Event):
    topic: str


class DecisionFlow(Workflow):
    llm = OpenAI(model="gpt-4o-mini")

    @step()
    async def dispatch_step(self, ev: StartEvent) -> AdvantageEvent | DisadvantageEvent | GatherEvent:
        topic = ev.get("topic")
        self.send_event(AdvantageEvent(topic=topic))
        self.send_event(DisadvantageEvent(topic=topic))
        return GatherEvent(topic=topic)

    @step()
    async def generate_advantage(self, ev: AdvantageEvent) -> AdvantageResultEvent:
        topic = ev.topic
         
        prompt = f"List the benefits about {topic}."
        response = await self.llm.acomplete(prompt)
        return AdvantageResultEvent(result=str(response))

    @step()
    async def generate_disadvantage(self, ev: DisadvantageEvent) -> DisadvantageResultEvent:
        topic = ev.topic

        prompt = f"List the disadvantages about {topic}"
        response = await self.llm.acomplete(prompt)
        return DisadvantageResultEvent(result=str(response))

    @step(pass_context=True)
    async def evaluate(self, ctx: Context, ev: GatherEvent | AdvantageResultEvent | DisadvantageResultEvent) -> StopEvent | None:
        events = ctx.collect_events(ev, [GatherEvent, AdvantageResultEvent, DisadvantageResultEvent])
        if not events:
            return None

        gather_event, advantage_event, disadvantage_event = events
        topic = gather_event.topic
        advantages = advantage_event.result
        disadvantages = disadvantage_event.result

        prompt = f"""\
Based on the advantages and disadvantages about {topic}, decide which is better.

Advantages:
{advantages}

Disadvantages:
{disadvantages}

Decision: 
"""
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=str(response))


w = DecisionFlow(timeout=60, verbose=True)
result = await w.run(topic="investing in NVIDIA")

Running step dispatch_step
Step dispatch_step produced event GatherEvent
Running step evaluate
Step evaluate produced no event
Running step generate_advantage
Running step generate_disadvantage
Step generate_disadvantage produced event DisadvantageResultEvent
Running step evaluate
Step evaluate produced no event
Step generate_advantage produced event AdvantageResultEvent
Running step evaluate
Step evaluate produced event StopEvent