【Agent】【workflow】15.反思工作流- 惊觉

1. 案例目标

本案例展示了如何使用LlamaIndex的Workflow框架构建一个具有反思功能的结构化输出系统。该系统通过重试和错误反思机制，确保大语言模型生成的输出符合预定义的JSON结构。当输出不符合要求时，系统会自动提供错误反馈并重新生成，直到获得有效的结构化输出或达到最大重试次数。

2. 技术栈与核心依赖

llama-index - LlamaIndex核心框架，提供工作流功能
llama-index-core - LlamaIndex核心组件，包含工作流、事件和上下文管理
llama-index-llms-ollama - Ollama本地语言模型集成
pydantic - 数据验证和设置管理
Ollama - 本地运行大语言模型的工具

3. 环境配置

注意： 本案例使用Ollama本地运行大语言模型。在运行前，请确保已安装并启动Ollama，并下载所需模型（如llama3）：

# 安装Ollama并下载模型
# 访问 https://ollama.com 获取安装说明
# 下载模型：ollama pull llama3

安装必要依赖：

!pip install -U llama-index llama-index-llms-ollama

4. 案例实现

4.1 工作流事件定义

定义了两个自定义事件来控制工作流的状态转换：

from llama_index.core.workflow import Event

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

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

4.2 数据模型定义

使用Pydantic定义要提取的数据结构：

from pydantic import BaseModel

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

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

4.3 反思工作流类

创建ReflectionWorkflow类，继承自Workflow，包含以下核心方法：

extract() - 生成结构化输出
validate() - 验证输出是否符合要求

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.store.get("retries", default=0)
        if current_retries >= self.max_retries:
            return StopEvent(result="Max retries reached")
        else:
            await ctx.store.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,
            # 手动设置上下文窗口以限制内存使用
            context_window=8000,
        )
        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)

4.4 工作流运行

初始化并运行反思工作流：

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."
)

print(ret)

5. 案例效果

该反思工作流实现了以下功能：

能够从非结构化文本中提取结构化数据
自动验证输出是否符合预定义的JSON结构
当验证失败时，提供错误反馈并重新生成
限制最大重试次数，防止无限循环
使用本地模型，保护数据隐私

示例输出：

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

6. 案例实现思路

本案例的实现基于以下核心思路：

事件驱动循环：通过事件系统实现生成-验证-重试的循环流程
结构化验证：使用Pydantic模型验证JSON输出的结构和类型
错误反馈：将验证错误信息反馈给模型，指导其生成正确的输出
重试机制：设置最大重试次数，平衡成功率和资源消耗
本地模型：使用Ollama本地运行模型，确保数据隐私和降低延迟

7. 扩展建议

多模型支持：支持切换不同的语言模型，比较不同模型的输出质量
动态重试策略：根据错误类型调整重试策略，如针对特定错误类型提供更精确的反馈
复杂结构支持：支持更复杂的嵌套数据结构和验证规则
批量处理：扩展为批量处理多个文本，提高处理效率
性能监控：添加性能指标收集，如重试次数、处理时间等
人机交互：在多次重试失败后，引入人工干预机制
缓存机制：对相似输入实现缓存，减少重复计算

8. 总结

本案例展示了如何使用LlamaIndex的Workflow框架构建一个具有反思功能的结构化输出系统。通过事件驱动的工作流设计，实现了生成、验证和重试的循环过程，确保输出符合预定义的结构。该系统特别适用于需要高精度结构化数据提取的场景，如信息抽取、数据转换等。通过本地模型和自定义验证逻辑，可以在保证数据隐私的同时，提高输出的准确性和可靠性。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/rengang66/article/details/165630938

【Agent】【workflow】15.反思工作流