LlamaIndex实现 JSON结构化输出的反射工作流程

🕗 发布于 2024-10-16 21:58 json llamaindex python llm DashScope

结构化输出的 Reflection 工作流程

本笔记本将介绍如何设置一个LlamaIndex的工作流，从用户的提问中，提取结构化对象的关键信息，以便通过重试和对错误进行反思来提供可靠的JSON结构化文本输出的。这在用户希望LLM能按照用户意愿生成标准的Json格式文本场景中，非常有用！

本文中使用的是国内阿里云的DashScope中 LLM

pip install -U llama-index-llms-dashscope

由于工作流首先是异步的，因此这一切都在 Notebook 中运行良好。如果您在自己的代码中运行，则希望使用它来启动异步事件循环（如果尚未运行）。asyncio.run()

async def main():
    <async code>

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

设计工作流程

要验证 LLM 的结构化输出，我们只需要两个步骤：

生成结构化输出
验证输出是否为正确的 JSON

这里的关键是，如果 output 无效，我们就会循环直到它无效，并将错误反馈提供给下一代。

工作流事件

要处理这些步骤，我们需要定义一些事件：

要在生成的提取中传递的事件
在提取无效时提供反馈的事件

其他步骤将使用内置的StartEvent和StopEvent事件。

from llama_index.core.workflow import Event


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


class ValidationErrorEvent(Event):
    error: str
    wrong_output: str
    passage: str

要提取的项目

为了提示我们的模型，让我们定义一个我们想要提取的 pydantic 模型。

from pydantic import BaseModel


class Car(BaseModel):
    brand: str
    model: str
    power: int


class CarCollection(BaseModel):
    cars: list[Car]

工作流本身

定义事件后，我们可以构建工作流和步骤。

请注意，工作流会使用类型注释自动验证自身，因此我们步骤中的类型注释非常有用！

import json

from llama_index.core.workflow import (
    Workflow,
    StartEvent,
    StopEvent,
    Context,
    step,
)
from llama_index.llms.ollama import Ollama


EXTRACTION_PROMPT = """
Context information is below:
---------------------
{passage}
---------------------

Given the context information and not prior knowledge, create a JSON object from the information in the context.
The JSON object must follow the JSON schema:
{schema}

"""

REFLECTION_PROMPT = """
You already created this output previously:
---------------------
{wrong_answer}
---------------------

This caused the JSON decode error: {error}

Try again, the response must contain only valid JSON code. Do not add any sentence before or after the JSON object.
Do not repeat the schema.
"""


class ReflectionWorkflow(Workflow):
    max_retries: int = 3

    @step
    async def extract(
        self, ctx: Context, ev: StartEvent | ValidationErrorEvent
    ) -> StopEvent | ExtractionDone:
        current_retries = await ctx.get("retries", default=0)
        if current_retries >= self.max_retries:
            return StopEvent(result="Max retries reached")
        else:
            await ctx.set("retries", current_retries + 1)

        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:
            CarCollection.model_validate_json(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)

就是这样！让我们稍微探索一下我们编写的工作流程。

我们有一个入口点（接受extractStartEvent)
完成后，它会发出一个事件extractExtraction
validate运行并确认提取：
- 如果正常，它将发出并停止工作流StopEvent
- 如果不是，它返回一个带有关于ValidationErrorEvent的信息
任何发出ValidationErrorEvent的信息都将触发循环，并再次运行！
这将一直持续到验证结构化输出为止

运行 Workflow！

注意：对于循环，我们需要注意运行时。这里，我们设置了 120 秒的超时时间。

w = ReflectionWorkflow(timeout=120, verbose=True)
# 运行工作流
ret = await w.run(
    passage="I own two cars: a Fiat Panda with 45Hp and a Honda Civic with 330Hp."
)

Running step extract
Step extract produced event ExtractionDone
Running step validate
Validation failed, retrying…
Step validate produced event ValidationErrorEvent
Running step extract
Step extract produced event ExtractionDone
Running step validate
Step validate produced event StopEvent

print(ret)

{ “cars”: [ { “brand”: “Fiat”, “model”: “Panda”, “power”: 45 }, { “brand”: “Honda”, “model”: “Civic”, “power”: 330 } ] }

完整代码

import asyncio

from llama_index.core.workflow import (
    Event,
    Context,
    StartEvent,
    StopEvent,
    Workflow,
    step, )
from llama_index.llms.dashscope import DashScope, DashScopeGenerationModels
from pydantic import BaseModel


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


class ValidationErrorEvent(Event):
    error: str
    wrong_output: str
    passage: str


class Car(BaseModel):
    brand: str
    model: str
    power: int


class CarCollection(BaseModel):
    cars: list[Car]


EXTRACTION_PROMPT = """
Context information is below:
---------------------
{passage}
---------------------

Given the context information and not prior knowledge, create a JSON object from the information in the context.
The JSON object must follow the JSON schema:
{schema}

"""

REFLECTION_PROMPT = """
You already created this output previously:
---------------------
{wrong_answer}
---------------------

This caused the JSON decode error: {error}

Try again, the response must contain only valid JSON code. Do not add any sentence before or after the JSON object.
Do not repeat the schema.
"""


class ReflectionWorkflow(Workflow):
    max_retries: int = 1

    @step
    async def extract(
            self, ctx: Context, ev: StartEvent | ValidationErrorEvent
    ) -> StopEvent | ExtractionDone:
        passage = None
        reflection_prompt = None
        current_retries = await ctx.get("retries", default=0)
        if current_retries >= self.max_retries:
            return StopEvent(result="Max retries reached")
        else:
            await ctx.set("retries", current_retries + 1)

        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 = DashScope(
            model_name=DashScopeGenerationModels.QWEN_PLUS,
            api_key="sk-your-api-key",
            max_tokens=512
        )
        prompt = EXTRACTION_PROMPT.format(
            passage=passage, schema=CarCollection.model_json_schema()
        )
        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:
            CarCollection.model_validate_json(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)


async def main():
    w = ReflectionWorkflow(timeout=120, verbose=True)

    # Run the workflow
    ret = await w.run(
        passage="I own two cars: a Fiat Panda with 45Hp and a Honda Civic with 330Hp."
    )
    print("-------------------------------------------------------------------------")
    print(ret)


asyncio.run(main())

替换代码中的api_key="sk-your-api-key" 为你的阿里云DashScope的key。

原文地址：https://blog.csdn.net/weixin_40986713/article/details/142953472

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：JL-31 管式墒情记录仪一体式水分测量仪土壤墒情监测仪
下一篇：【VUE】Vue中动态组件

水库抽样算法（大数据算法作业）
水库抽样技术归根到底就是在总体容量未知的情况下，仅通过单遍扫描数据集便能生成等概率抽样集合的一种均匀抽样技术。代码或许很简单，但是其中的数学知识以及思想方法是很值得学习的！
阅读更多2024-10-16
【C语言】--数组
😊个人主页: 起名字真南😋个人专栏:【数据结构初阶】【C语言】【C++】数组是一组相同类型元素的集合：一维数组创建的基本语法如下：存放在数组中的值被称为数组元素，数组在创建的时候可以指定数组的大
阅读更多2024-10-16
专题十一_递归_回溯_剪枝_综合练习_算法专题详细总结
这一题相对来说还是十分简单的，就是上一个专题的最后一题求子集，只要递归函数体能写对，那怎么写这题都能过。2.全排列 Ⅱ（medium）题意很简单就是对全排列后的所有数组添加到ret后去掉所有重复的数组
阅读更多2024-10-16
MySQL insert 记录后查询是乱码问题分析
MySQL客户端和服务器通信过程中的字符集处理
阅读更多2024-10-16
java怎么连接数据库sql server
在Java中，可以使用JDBC（Java Database Connectivity）来连接和操作SQL Server数据库。要实现这一点，你需要SQL Server的JDBC驱动程序，并编写Java
阅读更多2024-10-16
常用STL的操作以及特点
STL 提供了强大的数据结构和算法库，使得开发者可以快速、高效地解决许多常见的问题。每种容器和算法。
阅读更多2024-10-16
【优选算法】——双指针(下篇)！
b、如果大于，count+=（right-left ），固定max和right两边后，left是right之后最小的一边。2、排序后，我们定义target在len-1的位置，left=0，right=
阅读更多2024-10-16
python自动化办公实例（使用openpyxl、os处理统计Excel表中的数据并将其合并）
这里的手动处理可以通过优化我们的代码来进行一些简化以减少我们的工作量（如其他列不用删只要把奖项这里的列名进行统一就好了、文件名其实也可以不用进行重命名可以直接用我们之前的名字这样不仅可以减少工作量又可
阅读更多2024-10-16
王道考研视频——操作系统笔记第三章：内存管理
内存管理（ Memory Management）是操作系统设计中最重要和最复杂的内容之一。操作系统对内存的划分和动态分配，就是内存管理的概念
阅读更多2024-10-16
[搜索] 质数
给定n个正整数，将它们分组，使得每组中任意两个数互质。至少要分成多少个组？在满足最少的组数的情况下，使得元素个数最多的那一组的元素个数尽可能的少。
阅读更多2024-10-16