在查询管道周围构建代理¶

在本手册中，我们将向您展示如何在查询管道周围构建代理。

代理提供了在您设置的任何查询DAG之上进行复杂的顺序推理的能力。从概念上讲，这也是您可以向图形添加“循环”的一种方式之一。

我们将向您展示两个您可以实现的代理示例：

• 一个完整的 ReAct 代理，可以进行工具选择
• 一个“简单”的代理，可以在文本到SQL查询引擎周围添加重试层。

注意： 任何文本到SQL应用程序都应该意识到，执行任意SQL查询可能存在安全风险。建议采取必要的预防措施，例如使用受限角色、只读数据库、沙箱等。

In [ ]:

from llama_index.core import SQLDatabase
from sqlalchemy import (
    create_engine,
    MetaData,
    Table,
    Column,
    String,
    Integer,
    select,
    column,
)

engine = create_engine("sqlite:///chinook.db")
sql_database = SQLDatabase(engine)

from llama_index.core import SQLDatabase from sqlalchemy import ( create_engine, MetaData, Table, Column, String, Integer, select, column, ) engine = create_engine("sqlite:///chinook.db") sql_database = SQLDatabase(engine)

In [ ]:

from llama_index.core.query_pipeline import QueryPipeline

from llama_index.core.query_pipeline import QueryPipeline

设置¶

设置数据¶

我们使用chinook数据库作为示例数据。来源。

In [ ]:

%pip install llama-index-llms-openai

%pip install llama-index-llms-openai

In [ ]:

!curl "https://www.sqlitetutorial.net/wp-content/uploads/2018/03/chinook.zip" -O ./chinook.zip
!unzip ./chinook.zip

!curl "https://www.sqlitetutorial.net/wp-content/uploads/2018/03/chinook.zip" -O ./chinook.zip !unzip ./chinook.zip

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100  298k  100  298k    0     0  2327k      0 --:--:-- --:--:-- --:--:-- 2387k
curl: (6) Could not resolve host: .
Archive:  ./chinook.zip
  inflating: chinook.db              

设置可观测性¶

我们设置 Arize Phoenix 以实现可观测性。

In [ ]:

# 定义全局回调设置
from llama_index.core.settings import Settings
from llama_index.core.callbacks import CallbackManager

callback_manager = CallbackManager()
Settings.callback_manager = callback_manager

# 定义全局回调设置 from llama_index.core.settings import Settings from llama_index.core.callbacks import CallbackManager callback_manager = CallbackManager() Settings.callback_manager = callback_manager

In [ ]:

# 设置 Arize Phoenix 用于日志记录/可观测性
import phoenix as px
import llama_index.core

px.launch_app()
llama_index.core.set_global_handler("arize_phoenix")

# 设置 Arize Phoenix 用于日志记录/可观测性 import phoenix as px import llama_index.core px.launch_app() llama_index.core.set_global_handler("arize_phoenix")

设置文本到SQL查询引擎/工具¶

现在我们设置一个简单的文本到SQL工具：给定一个查询，将文本翻译成SQL，执行在数据库中，并得到结果。

In [ ]:

from llama_index.core.query_engine import NLSQLTableQueryEngine
from llama_index.core.tools import QueryEngineTool

sql_query_engine = NLSQLTableQueryEngine(
    sql_database=sql_database,
    tables=["albums", "tracks", "artists"],
    verbose=True,
)
sql_tool = QueryEngineTool.from_defaults(
    query_engine=sql_query_engine,
    name="sql_tool",
    description=(
        "Useful for translating a natural language query into a SQL query"
    ),
)

from llama_index.core.query_engine import NLSQLTableQueryEngine from llama_index.core.tools import QueryEngineTool sql_query_engine = NLSQLTableQueryEngine( sql_database=sql_database, tables=["albums", "tracks", "artists"], verbose=True, ) sql_tool = QueryEngineTool.from_defaults( query_engine=sql_query_engine, name="sql_tool", description=( "Useful for translating a natural language query into a SQL query" ), )

设置ReAct Agent Pipeline¶

现在，我们将使用我们的查询管道语法为单个步骤设置一个ReAct管道。这是一个多部分的过程，包括以下内容：

1. 接收agent输入
2. 使用LLM调用ReAct提示生成下一个动作/工具（或返回一个响应）。
3. 如果选择工具/动作，则调用工具管道来执行工具+收集响应。
4. 如果生成响应，则获取响应。

在整个过程中，我们将使用各种特定于agent的查询组件。与普通的查询管道不同，这些组件专门设计用于在QueryPipelineAgentWorker中使用的查询管道：

• 一个AgentInputComponent，允许您将agent输入（任务、状态字典）转换为查询管道的一组输入。
• 一个AgentFnComponent：一个通用处理器，允许您接收当前任务、状态以及任意输入，并返回一个输出。在本手册中，我们定义了一个函数组件来格式化ReAct提示。但是，您可以将其放在任何地方。
• [本笔记本中未使用] 一个CustomAgentComponent：类似于AgentFnComponent，您可以实现_run_component来定义自己的逻辑，同时可以访问任务和状态。它比AgentFnComponent更冗长，但比AgentFnComponent更灵活（例如，您可以定义初始化变量，并且回调在基类中）。

请注意，传递给AgentFnComponent和AgentInputComponent的任何函数都必须包括task和state作为输入变量，因为这些是从agent传递的输入。

请注意，agentic查询管道的输出必须是Tuple[AgentChatResponse, bool]。您将在下面看到这一点。

In [ ]:

from llama_index.core.query_pipeline import QueryPipeline as QP

qp = QP(verbose=True)

from llama_index.core.query_pipeline import QueryPipeline as QP qp = QP(verbose=True)

定义Agent输入组件¶

在这里，我们定义了Agent输入组件，它在每个Agent步骤的开始时被调用。除了传递输入之外，我们还进行初始化和状态修改。

In [ ]:

from llama_index.core.agent.react.types import (
    ActionReasoningStep,  # 行动推理步骤
    ObservationReasoningStep,  # 观察推理步骤
    ResponseReasoningStep,  # 响应推理步骤
)
from llama_index.core.agent import Task, AgentChatResponse  # 任务，代理聊天响应
from llama_index.core.query_pipeline import (
    AgentInputComponent,  # 代理输入组件
    AgentFnComponent,  # 代理函数组件
    CustomAgentComponent,  # 自定义代理组件
    QueryComponent,  # 查询组件
    ToolRunnerComponent,  # 工具运行组件
)
from llama_index.core.llms import MessageRole  # 消息角色
from typing import Dict, Any, Optional, Tuple, List, cast  # 导入类型提示


## 代理输入组件
## 这是产生代理输入的组件，供其他组件使用
## 也可以在这里放置初始化逻辑。
def agent_input_fn(task: Task, state: Dict[str, Any]) -> Dict[str, Any]:
    """代理输入函数。

    返回:
        一个包含输出键和值的字典。如果在定义该组件与其他组件之间的链接时指定了src_key，
        请确保src_key与指定的output_key匹配。

    """
    # 初始化当前推理
    if "current_reasoning" not in state:
        state["current_reasoning"] = []
    reasoning_step = ObservationReasoningStep(observation=task.input)
    state["current_reasoning"].append(reasoning_step)
    return {"input": task.input}


agent_input_component = AgentInputComponent(fn=agent_input_fn)

from llama_index.core.agent.react.types import ( ActionReasoningStep, # 行动推理步骤 ObservationReasoningStep, # 观察推理步骤 ResponseReasoningStep, # 响应推理步骤 ) from llama_index.core.agent import Task, AgentChatResponse # 任务，代理聊天响应 from llama_index.core.query_pipeline import ( AgentInputComponent, # 代理输入组件 AgentFnComponent, # 代理函数组件 CustomAgentComponent, # 自定义代理组件 QueryComponent, # 查询组件 ToolRunnerComponent, # 工具运行组件 ) from llama_index.core.llms import MessageRole # 消息角色 from typing import Dict, Any, Optional, Tuple, List, cast # 导入类型提示 ## 代理输入组件 ## 这是产生代理输入的组件，供其他组件使用 ## 也可以在这里放置初始化逻辑。 def agent_input_fn(task: Task, state: Dict[str, Any]) -> Dict[str, Any]: """代理输入函数。 返回: 一个包含输出键和值的字典。如果在定义该组件与其他组件之间的链接时指定了src_key， 请确保src_key与指定的output_key匹配。 """ # 初始化当前推理 if "current_reasoning" not in state: state["current_reasoning"] = [] reasoning_step = ObservationReasoningStep(observation=task.input) state["current_reasoning"].append(reasoning_step) return {"input": task.input} agent_input_component = AgentInputComponent(fn=agent_input_fn)

定义代理提示¶

在这里，我们定义了生成 ReAct 提示的代理组件，并且在从LLM生成输出之后，将其解析为一个结构化对象。

In [ ]:

from llama_index.core.agent import ReActChatFormatter
from llama_index.core.query_pipeline import InputComponent, Link
from llama_index.core.llms import ChatMessage
from llama_index.core.tools import BaseTool


## 定义提示函数
def react_prompt_fn(
    task: Task, state: Dict[str, Any], input: str, tools: List[BaseTool]
) -> List[ChatMessage]:
    # 将输入添加到推理中
    chat_formatter = ReActChatFormatter()
    return chat_formatter.format(
        tools,
        chat_history=task.memory.get() + state["memory"].get_all(),
        current_reasoning=state["current_reasoning"],
    )


react_prompt_component = AgentFnComponent(
    fn=react_prompt_fn, partial_dict={"tools": [sql_tool]}
)

from llama_index.core.agent import ReActChatFormatter from llama_index.core.query_pipeline import InputComponent, Link from llama_index.core.llms import ChatMessage from llama_index.core.tools import BaseTool ## 定义提示函数 def react_prompt_fn( task: Task, state: Dict[str, Any], input: str, tools: List[BaseTool] ) -> List[ChatMessage]: # 将输入添加到推理中 chat_formatter = ReActChatFormatter() return chat_formatter.format( tools, chat_history=task.memory.get() + state["memory"].get_all(), current_reasoning=state["current_reasoning"], ) react_prompt_component = AgentFnComponent( fn=react_prompt_fn, partial_dict={"tools": [sql_tool]} )

定义代理输出解析器 + 工具流水线¶

一旦LLM给出一个输出，我们就有一个决策树：

1. 如果给出了一个答案，那么我们就完成了。处理输出。
2. 如果给出了一个动作，我们需要使用指定的工具和参数执行指定的工具，然后处理输出。

工具调用可以通过ToolRunnerComponent模块来实现。这是一个简单的包装模块，它接受一个工具列表，并可以使用指定的工具名称（每个工具都有一个名称）和工具动作来“执行”。

我们实现这个整体模块OutputAgentComponent，它是CustomAgentComponent的子类。

注意：我们还实现了sub_query_components来将高级回调管理器传递到工具运行子模块。

In [ ]:

from typing import Set, Optional
from llama_index.core.agent.react.output_parser import ReActOutputParser
from llama_index.core.llms import ChatResponse
from llama_index.core.agent.types import Task


def parse_react_output_fn(
    task: Task, state: Dict[str, Any], chat_response: ChatResponse
):
    """将ReAct输出解析为推理步骤。"""
    output_parser = ReActOutputParser()
    reasoning_step = output_parser.parse(chat_response.message.content)
    return {"done": reasoning_step.is_done, "reasoning_step": reasoning_step}


parse_react_output = AgentFnComponent(fn=parse_react_output_fn)


def run_tool_fn(
    task: Task, state: Dict[str, Any], reasoning_step: ActionReasoningStep
):
    """运行工具并处理工具输出。"""
    tool_runner_component = ToolRunnerComponent(
        [sql_tool], callback_manager=task.callback_manager
    )
    tool_output = tool_runner_component.run_component(
        tool_name=reasoning_step.action,
        tool_input=reasoning_step.action_input,
    )
    observation_step = ObservationReasoningStep(observation=str(tool_output))
    state["current_reasoning"].append(observation_step)
    # TODO: 获取输出

    return {"response_str": observation_step.get_content(), "is_done": False}


run_tool = AgentFnComponent(fn=run_tool_fn)


def process_response_fn(
    task: Task, state: Dict[str, Any], response_step: ResponseReasoningStep
):
    """处理响应。"""
    state["current_reasoning"].append(response_step)
    response_str = response_step.response
    # 现在我们完成了这一步，将其放入内存
    state["memory"].put(ChatMessage(content=task.input, role=MessageRole.USER))
    state["memory"].put(
        ChatMessage(content=response_str, role=MessageRole.ASSISTANT)
    )

    return {"response_str": response_str, "is_done": True}


process_response = AgentFnComponent(fn=process_response_fn)


def process_agent_response_fn(
    task: Task, state: Dict[str, Any], response_dict: dict
):
    """处理代理响应。"""
    return (
        AgentChatResponse(response_dict["response_str"]),
        response_dict["is_done"],
    )


process_agent_response = AgentFnComponent(fn=process_agent_response_fn)

from typing import Set, Optional from llama_index.core.agent.react.output_parser import ReActOutputParser from llama_index.core.llms import ChatResponse from llama_index.core.agent.types import Task def parse_react_output_fn( task: Task, state: Dict[str, Any], chat_response: ChatResponse ): """将ReAct输出解析为推理步骤。""" output_parser = ReActOutputParser() reasoning_step = output_parser.parse(chat_response.message.content) return {"done": reasoning_step.is_done, "reasoning_step": reasoning_step} parse_react_output = AgentFnComponent(fn=parse_react_output_fn) def run_tool_fn( task: Task, state: Dict[str, Any], reasoning_step: ActionReasoningStep ): """运行工具并处理工具输出。""" tool_runner_component = ToolRunnerComponent( [sql_tool], callback_manager=task.callback_manager ) tool_output = tool_runner_component.run_component( tool_name=reasoning_step.action, tool_input=reasoning_step.action_input, ) observation_step = ObservationReasoningStep(observation=str(tool_output)) state["current_reasoning"].append(observation_step) # TODO: 获取输出 return {"response_str": observation_step.get_content(), "is_done": False} run_tool = AgentFnComponent(fn=run_tool_fn) def process_response_fn( task: Task, state: Dict[str, Any], response_step: ResponseReasoningStep ): """处理响应。""" state["current_reasoning"].append(response_step) response_str = response_step.response # 现在我们完成了这一步，将其放入内存 state["memory"].put(ChatMessage(content=task.input, role=MessageRole.USER)) state["memory"].put( ChatMessage(content=response_str, role=MessageRole.ASSISTANT) ) return {"response_str": response_str, "is_done": True} process_response = AgentFnComponent(fn=process_response_fn) def process_agent_response_fn( task: Task, state: Dict[str, Any], response_dict: dict ): """处理代理响应。""" return ( AgentChatResponse(response_dict["response_str"]), response_dict["is_done"], ) process_agent_response = AgentFnComponent(fn=process_agent_response_fn)

组合代理查询管道¶

现在我们可以将顶层代理管道组合在一起：agent_input -> react_prompt -> llm -> react_output。

最后一个组件是调用子组件的if-else组件。

In [ ]:

from llama_index.core.query_pipeline import QueryPipeline as QP
from llama_index.llms.openai import OpenAI

qp.add_modules(
    {
        "agent_input": agent_input_component,
        "react_prompt": react_prompt_component,
        "llm": OpenAI(model="gpt-4-1106-preview"),
        "react_output_parser": parse_react_output,
        "run_tool": run_tool,
        "process_response": process_response,
        "process_agent_response": process_agent_response,
    }
)

from llama_index.core.query_pipeline import QueryPipeline as QP from llama_index.llms.openai import OpenAI qp.add_modules( { "agent_input": agent_input_component, "react_prompt": react_prompt_component, "llm": OpenAI(model="gpt-4-1106-preview"), "react_output_parser": parse_react_output, "run_tool": run_tool, "process_response": process_response, "process_agent_response": process_agent_response, } )

In [ ]:

# 将输入链接到反应提示以解析出响应（工具操作/输入或观察）
qp.add_chain(["agent_input", "react_prompt", "llm", "react_output_parser"])

# 从反应输出到工具调用添加条件链接（如果尚未完成）
qp.add_link(
    "react_output_parser",
    "run_tool",
    condition_fn=lambda x: not x["done"],
    input_fn=lambda x: x["reasoning_step"],
)
# 从反应输出到最终响应处理添加条件链接（如果已完成）
qp.add_link(
    "react_output_parser",
    "process_response",
    condition_fn=lambda x: x["done"],
    input_fn=lambda x: x["reasoning_step"],
)

# 无论是响应处理还是工具输出处理，都添加链接到最终代理响应
qp.add_link("process_response", "process_agent_response")
qp.add_link("run_tool", "process_agent_response")

# 将输入链接到反应提示以解析出响应（工具操作/输入或观察） qp.add_chain(["agent_input", "react_prompt", "llm", "react_output_parser"]) # 从反应输出到工具调用添加条件链接（如果尚未完成） qp.add_link( "react_output_parser", "run_tool", condition_fn=lambda x: not x["done"], input_fn=lambda x: x["reasoning_step"], ) # 从反应输出到最终响应处理添加条件链接（如果已完成） qp.add_link( "react_output_parser", "process_response", condition_fn=lambda x: x["done"], input_fn=lambda x: x["reasoning_step"], ) # 无论是响应处理还是工具输出处理，都添加链接到最终代理响应 qp.add_link("process_response", "process_agent_response") qp.add_link("run_tool", "process_agent_response")

可视化查询管道¶

In [ ]:

from pyvis.network import Network

net = Network(notebook=True, cdn_resources="in_line", directed=True)
net.from_nx(qp.clean_dag)
net.show("agent_dag.html")

from pyvis.network import Network net = Network(notebook=True, cdn_resources="in_line", directed=True) net.from_nx(qp.clean_dag) net.show("agent_dag.html")

agent_dag.html

Out[ ]:

在文本到SQL查询管道周围设置代理工作程序¶

这是我们设置代理工作程序以围绕文本到SQL查询管道的方式。

In [ ]:

from llama_index.core.agent import QueryPipelineAgentWorker
from llama_index.core.callbacks import CallbackManager

agent_worker = QueryPipelineAgentWorker(qp)
agent = agent_worker.as_agent(
    callback_manager=CallbackManager([]), verbose=True
)

from llama_index.core.agent import QueryPipelineAgentWorker from llama_index.core.callbacks import CallbackManager agent_worker = QueryPipelineAgentWorker(qp) agent = agent_worker.as_agent( callback_manager=CallbackManager([]), verbose=True )

运行代理程序¶

让我们尝试在一些示例查询上运行代理程序。

In [ ]:

# 开始任务
task = agent.create_task(
    "AC/DC乐队有哪些歌曲？限制在3首内"
)

# 开始任务 task = agent.create_task( "AC/DC乐队有哪些歌曲？限制在3首内" )

In [ ]:

step_output = agent.run_step(task.task_id)

step_output = agent.run_step(task.task_id)

> Running step edb9926c-7290-4b8d-ac80-1421432a0ea6. Step input: What are some tracks from the artist AC/DC? Limit it to 3
> Running module agent_input with input: 
state: {'sources': [], 'memory': ChatMemoryBuffer(token_limit=3000, tokenizer_fn=functools.partial(<bound method Encoding.encode of <Encoding 'cl100k_base'>>, allowed_special='all'), chat_store=SimpleChatSto...
task: task_id='b9b747a7-880f-4e91-9eed-b64574cbb6d0' input='What are some tracks from the artist AC/DC? Limit it to 3' memory=ChatMemoryBuffer(token_limit=3000, tokenizer_fn=functools.partial(<bound method ...

> Running module react_prompt with input: 
input: What are some tracks from the artist AC/DC? Limit it to 3

> Running module llm with input: 
messages: [ChatMessage(role=<MessageRole.SYSTEM: 'system'>, content='\nYou are designed to help with a variety of tasks, from answering questions     to providing summaries to other types of analyses.\n\n## Too...

> Running module react_output_parser with input: 
chat_response: assistant: Thought: I need to use a tool to help me answer the question.
Action: sql_tool
Action Input: {"input": "Select track_name from tracks where artist_name = 'AC/DC' limit 3"}

> Running module run_tool with input: 
reasoning_step: thought='I need to use a tool to help me answer the question.' action='sql_tool' action_input={'input': "Select track_name from tracks where artist_name = 'AC/DC' limit 3"}

> Running module process_agent_response with input: 
response_dict: {'response_str': 'Observation: {\'output\': ToolOutput(content=\'The top 3 tracks by AC/DC are "For Those About To Rock (We Salute You)", "Put The Finger On You", and "Let\\\'s Get It Up".\', tool_nam...

In [ ]:

step_output = agent.run_step(task.task_id)

step_output = agent.run_step(task.task_id)

> Running step 37e2312b-540b-4c79-9261-15318d4796d9. Step input: None
> Running module agent_input with input: 
state: {'sources': [], 'memory': ChatMemoryBuffer(token_limit=3000, tokenizer_fn=functools.partial(<bound method Encoding.encode of <Encoding 'cl100k_base'>>, allowed_special='all'), chat_store=SimpleChatSto...
task: task_id='b9b747a7-880f-4e91-9eed-b64574cbb6d0' input='What are some tracks from the artist AC/DC? Limit it to 3' memory=ChatMemoryBuffer(token_limit=3000, tokenizer_fn=functools.partial(<bound method ...

> Running module react_prompt with input: 
input: What are some tracks from the artist AC/DC? Limit it to 3

> Running module llm with input: 
messages: [ChatMessage(role=<MessageRole.SYSTEM: 'system'>, content='\nYou are designed to help with a variety of tasks, from answering questions     to providing summaries to other types of analyses.\n\n## Too...

> Running module react_output_parser with input: 
chat_response: assistant: Thought: The user has repeated the request, possibly due to not noticing the previous response. I will provide the information again.

Answer: The top 3 tracks by AC/DC are "For Those About...

> Running module process_response with input: 
response_step: thought='The user has repeated the request, possibly due to not noticing the previous response. I will provide the information again.' response='The top 3 tracks by AC/DC are "For Those About To Rock ...

> Running module process_agent_response with input: 
response_dict: {'response_str': 'The top 3 tracks by AC/DC are "For Those About To Rock (We Salute You)", "Put The Finger On You", and "Let\'s Get It Up".', 'is_done': True}

In [ ]:

step_output.is_last

step_output.is_last

Out[ ]:

True

In [ ]:

response = agent.finalize_response(task.task_id)

response = agent.finalize_response(task.task_id)

In [ ]:

print(str(response))

print(str(response))

The top 3 tracks by AC/DC are "For Those About To Rock (We Salute You)", "Put The Finger On You", and "Let's Get It Up".

In [ ]:

# 运行这个端到端测试
agent.reset()
response = agent.chat(
    "AC/DC这个艺术家有哪些歌曲？限制在3首内"
)

# 运行这个端到端测试 agent.reset() response = agent.chat( "AC/DC这个艺术家有哪些歌曲？限制在3首内" )

> Running step 781d6e78-5bfe-4330-b8fc-3242deb6f64a. Step input: What are some tracks from the artist AC/DC? Limit it to 3
> Running module agent_input with input: 
state: {'sources': [], 'memory': ChatMemoryBuffer(token_limit=3000, tokenizer_fn=functools.partial(<bound method Encoding.encode of <Encoding 'cl100k_base'>>, allowed_special='all'), chat_store=SimpleChatSto...
task: task_id='c09dd358-19e8-4fcc-8b82-326783ba4af2' input='What are some tracks from the artist AC/DC? Limit it to 3' memory=ChatMemoryBuffer(token_limit=3000, tokenizer_fn=functools.partial(<bound method ...

> Running module react_prompt with input: 
input: What are some tracks from the artist AC/DC? Limit it to 3

> Running module llm with input: 
messages: [ChatMessage(role=<MessageRole.SYSTEM: 'system'>, content='\nYou are designed to help with a variety of tasks, from answering questions     to providing summaries to other types of analyses.\n\n## Too...

> Running module react_output_parser with input: 
chat_response: assistant: Thought: I need to use a tool to help me answer the question.
Action: sql_tool
Action Input: {"input": "SELECT track_name FROM tracks WHERE artist_name = 'AC/DC' LIMIT 3"}

> Running module run_tool with input: 
reasoning_step: thought='I need to use a tool to help me answer the question.' action='sql_tool' action_input={'input': "SELECT track_name FROM tracks WHERE artist_name = 'AC/DC' LIMIT 3"}

> Running module process_agent_response with input: 
response_dict: {'response_str': 'Observation: {\'output\': ToolOutput(content=\'The top three tracks by AC/DC are "For Those About To Rock (We Salute You)", "Put The Finger On You", and "Let\\\'s Get It Up".\', tool...

> Running step a65d44a6-7a98-49ec-86ce-eb4b3bcd6a48. Step input: None
> Running module agent_input with input: 
state: {'sources': [], 'memory': ChatMemoryBuffer(token_limit=3000, tokenizer_fn=functools.partial(<bound method Encoding.encode of <Encoding 'cl100k_base'>>, allowed_special='all'), chat_store=SimpleChatSto...
task: task_id='c09dd358-19e8-4fcc-8b82-326783ba4af2' input='What are some tracks from the artist AC/DC? Limit it to 3' memory=ChatMemoryBuffer(token_limit=3000, tokenizer_fn=functools.partial(<bound method ...

> Running module react_prompt with input: 
input: What are some tracks from the artist AC/DC? Limit it to 3

> Running module llm with input: 
messages: [ChatMessage(role=<MessageRole.SYSTEM: 'system'>, content='\nYou are designed to help with a variety of tasks, from answering questions     to providing summaries to other types of analyses.\n\n## Too...

> Running module react_output_parser with input: 
chat_response: assistant: Thought: The user has repeated the request for tracks from the artist AC/DC, limited to 3, despite having already received an answer. It's possible that the user did not see the previous re...

> Running module process_response with input: 
response_step: thought="The user has repeated the request for tracks from the artist AC/DC, limited to 3, despite having already received an answer. It's possible that the user did not see the previous response, or ...

> Running module process_agent_response with input: 
response_dict: {'response_str': 'The top three tracks by AC/DC are "For Those About To Rock (We Salute You)", "Put The Finger On You", and "Let\'s Get It Up".', 'is_done': True}

In [ ]:

print(str(response))

print(str(response))

The top three tracks by AC/DC are "For Those About To Rock (We Salute You)", "Put The Finger On You", and "Let's Get It Up".

设置简单的文本到SQL重试代理管道¶

与进行工具选择的完整ReAct管道不同，让我们尝试一个更简单的代理管道，只进行文本到SQL的转换，并带有重试逻辑。

我们尝试一个简单的基于文本的“重试”提示，根据用户输入和先前的对话历史，可以生成一个修改后的查询，以输出正确的结果。

定义核心模块¶

• 代理输入
• 重试提示
• 输出处理器（包括验证提示）

In [ ]:

来自llama_index.llms.openai的OpenAI

# llm = OpenAI(model="gpt-3.5-turbo")
llm = OpenAI(model="gpt-4-1106-preview")

来自llama_index.llms.openai的OpenAI # llm = OpenAI(model="gpt-3.5-turbo") llm = OpenAI(model="gpt-4-1106-preview")

In [ ]:

from llama_index.core.agent import Task, AgentChatResponse
from typing import Dict, Any
from llama_index.core.query_pipeline import (
    AgentInputComponent,
    AgentFnComponent,
)


def agent_input_fn(task: Task, state: Dict[str, Any]) -> Dict:
    """代理输入函数。"""
    # 初始化当前推理
    if "convo_history" not in state:
        state["convo_history"] = []
        state["count"] = 0
    state["convo_history"].append(f"用户: {task.input}")
    convo_history_str = "\n".join(state["convo_history"]) or "无"
    return {"input": task.input, "convo_history": convo_history_str}


agent_input_component = AgentInputComponent(fn=agent_input_fn)

from llama_index.core.agent import Task, AgentChatResponse from typing import Dict, Any from llama_index.core.query_pipeline import ( AgentInputComponent, AgentFnComponent, ) def agent_input_fn(task: Task, state: Dict[str, Any]) -> Dict: """代理输入函数。""" # 初始化当前推理 if "convo_history" not in state: state["convo_history"] = [] state["count"] = 0 state["convo_history"].append(f"用户: {task.input}") convo_history_str = "\n".join(state["convo_history"]) or "无" return {"input": task.input, "convo_history": convo_history_str} agent_input_component = AgentInputComponent(fn=agent_input_fn)

In [ ]:

from llama_index.core import PromptTemplate

retry_prompt_str = """\
您正在尝试根据用户输入生成一个合适的自然语言查询。

然后，这个查询将被下游的文本到SQL代理解释，该代理将把查询转换为SQL语句。如果代理触发错误，那么这将反映在当前的对话历史中（见下文）。

如果对话历史为None，请使用用户输入。如果不为None，则生成一个新的SQL查询，避免前一个SQL查询的问题。

输入：{input}
对话历史（失败的尝试）：
{convo_history}

新输入："""
retry_prompt = PromptTemplate(retry_prompt_str)

from llama_index.core import PromptTemplate retry_prompt_str = """\ 您正在尝试根据用户输入生成一个合适的自然语言查询。 然后，这个查询将被下游的文本到SQL代理解释，该代理将把查询转换为SQL语句。如果代理触发错误，那么这将反映在当前的对话历史中（见下文）。 如果对话历史为None，请使用用户输入。如果不为None，则生成一个新的SQL查询，避免前一个SQL查询的问题。 输入：{input} 对话历史（失败的尝试）： {convo_history} 新输入：""" retry_prompt = PromptTemplate(retry_prompt_str)

In [ ]:

from llama_index.core import Response
from typing import Tuple

validate_prompt_str = """\
给定用户查询，验证推断的SQL查询和执行查询后的响应是否正确并回答了查询。

用YES或NO回答。

查询: {input}
推断的SQL查询: {sql_query}
SQL响应: {sql_response}

结果: """
validate_prompt = PromptTemplate(validate_prompt_str)

MAX_ITER = 3


def agent_output_fn(
    task: Task, state: Dict[str, Any], output: Response
) -> Tuple[AgentChatResponse, bool]:
    """代理输出组件。"""
    print(f"> 推断的SQL查询: {output.metadata['sql_query']}")
    print(f"> SQL响应: {str(output)}")
    state["convo_history"].append(
        f"助手（推断的SQL查询）: {output.metadata['sql_query']}"
    )
    state["convo_history"].append(f"助手（响应）: {str(output)}")

    # 运行一个小型链来获取响应
    validate_prompt_partial = validate_prompt.as_query_component(
        partial={
            "sql_query": output.metadata["sql_query"],
            "sql_response": str(output),
        }
    )
    qp = QP(chain=[validate_prompt_partial, llm])
    validate_output = qp.run(input=task.input)

    state["count"] += 1
    is_done = False
    if state["count"] >= MAX_ITER:
        is_done = True
    if "YES" in validate_output.message.content:
        is_done = True

    return AgentChatResponse(response=str(output)), is_done


agent_output_component = AgentFnComponent(fn=agent_output_fn)

from llama_index.core import Response from typing import Tuple validate_prompt_str = """\ 给定用户查询，验证推断的SQL查询和执行查询后的响应是否正确并回答了查询。 用YES或NO回答。 查询: {input} 推断的SQL查询: {sql_query} SQL响应: {sql_response} 结果: """ validate_prompt = PromptTemplate(validate_prompt_str) MAX_ITER = 3 def agent_output_fn( task: Task, state: Dict[str, Any], output: Response ) -> Tuple[AgentChatResponse, bool]: """代理输出组件。""" print(f"> 推断的SQL查询: {output.metadata['sql_query']}") print(f"> SQL响应: {str(output)}") state["convo_history"].append( f"助手（推断的SQL查询）: {output.metadata['sql_query']}" ) state["convo_history"].append(f"助手（响应）: {str(output)}") # 运行一个小型链来获取响应 validate_prompt_partial = validate_prompt.as_query_component( partial={ "sql_query": output.metadata["sql_query"], "sql_response": str(output), } ) qp = QP(chain=[validate_prompt_partial, llm]) validate_output = qp.run(input=task.input) state["count"] += 1 is_done = False if state["count"] >= MAX_ITER: is_done = True if "YES" in validate_output.message.content: is_done = True return AgentChatResponse(response=str(output)), is_done agent_output_component = AgentFnComponent(fn=agent_output_fn)

In [ ]:

from llama_index.core.query_pipeline import (
    QueryPipeline as QP,
    Link,
    InputComponent,
)

qp = QP(
    modules={
        "input": agent_input_component,
        "retry_prompt": retry_prompt,
        "llm": llm,
        "sql_query_engine": sql_query_engine,
        "output_component": agent_output_component,
    },
    verbose=True,
)
qp.add_link("input", "retry_prompt", src_key="input", dest_key="input")
qp.add_link(
    "input", "retry_prompt", src_key="convo_history", dest_key="convo_history"
)
qp.add_chain(["retry_prompt", "llm", "sql_query_engine", "output_component"])

from llama_index.core.query_pipeline import ( QueryPipeline as QP, Link, InputComponent, ) qp = QP( modules={ "input": agent_input_component, "retry_prompt": retry_prompt, "llm": llm, "sql_query_engine": sql_query_engine, "output_component": agent_output_component, }, verbose=True, ) qp.add_link("input", "retry_prompt", src_key="input", dest_key="input") qp.add_link( "input", "retry_prompt", src_key="convo_history", dest_key="convo_history" ) qp.add_chain(["retry_prompt", "llm", "sql_query_engine", "output_component"])

可视化查询管道¶

In [ ]:

from pyvis.network import Network

net = Network(notebook=True, cdn_resources="in_line", directed=True)
net.from_nx(qp.dag)
net.show("agent_dag.html")

from pyvis.network import Network net = Network(notebook=True, cdn_resources="in_line", directed=True) net.from_nx(qp.dag) net.show("agent_dag.html")

agent_dag.html

Out[ ]:

定义代理工作者¶

In [ ]:

from llama_index.core.agent import QueryPipelineAgentWorker
from llama_index.core.callbacks import CallbackManager

# 回调管理器从查询管道传递给代理工作器/代理
agent_worker = QueryPipelineAgentWorker(qp)
agent = agent_worker.as_agent(
    callback_manager=CallbackManager(), verbose=False
)

from llama_index.core.agent import QueryPipelineAgentWorker from llama_index.core.callbacks import CallbackManager # 回调管理器从查询管道传递给代理工作器/代理 agent_worker = QueryPipelineAgentWorker(qp) agent = agent_worker.as_agent( callback_manager=CallbackManager(), verbose=False )

In [ ]:

response = agent.chat(
    "How many albums did the artist who wrote 'Restless and Wild' release? (answer should be non-zero)?"
)
print(str(response))

response = agent.chat( "How many albums did the artist who wrote 'Restless and Wild' release? (answer should be non-zero)?" ) print(str(response))

> Running module input with input: 
state: {'sources': [], 'memory': ChatMemoryBuffer(token_limit=3000, tokenizer_fn=functools.partial(<bound method Encoding.encode of <Encoding 'cl100k_base'>>, allowed_special='all'), chat_store=SimpleChatSto...
task: task_id='2d8a63de-7410-4422-98f3-f0ca41884f58' input="How many albums did the artist who wrote 'Restless and Wild' release? (answer should be non-zero)?" memory=ChatMemoryBuffer(token_limit=3000, toke...

> Running module retry_prompt with input: 
input: How many albums did the artist who wrote 'Restless and Wild' release? (answer should be non-zero)?
convo_history: User: How many albums did the artist who wrote 'Restless and Wild' release? (answer should be non-zero)?

> Running module llm with input: 
messages: You are trying to generate a proper natural language query given a user input.

This query will then be interpreted by a downstream text-to-SQL agent which
will convert the query to a SQL statement. I...

> Running module sql_query_engine with input: 
input: assistant: Given the conversation history, it seems that the previous attempt to generate a SQL query from the user's question may have resulted in an error. To avoid the same problem, we need to reph...

> Running module output_component with input: 
output: I'm sorry, but there seems to be an error in the SQL query. The query you provided is invalid SQL. Please double-check the syntax and try again.

> Inferred SQL Query: SELECT COUNT(albums.AlbumId) 
FROM albums 
JOIN tracks ON albums.AlbumId = tracks.AlbumId 
WHERE tracks.Name = 'Restless and Wild' 
AND albums.ArtistId = tracks.Composer 
AND COUNT(albums.AlbumId) > 0
> SQL Response: I'm sorry, but there seems to be an error in the SQL query. The query you provided is invalid SQL. Please double-check the syntax and try again.
> Running module input with input: 
state: {'sources': [], 'memory': ChatMemoryBuffer(token_limit=3000, tokenizer_fn=functools.partial(<bound method Encoding.encode of <Encoding 'cl100k_base'>>, allowed_special='all'), chat_store=SimpleChatSto...
task: task_id='2d8a63de-7410-4422-98f3-f0ca41884f58' input="How many albums did the artist who wrote 'Restless and Wild' release? (answer should be non-zero)?" memory=ChatMemoryBuffer(token_limit=3000, toke...

> Running module retry_prompt with input: 
input: How many albums did the artist who wrote 'Restless and Wild' release? (answer should be non-zero)?
convo_history: User: How many albums did the artist who wrote 'Restless and Wild' release? (answer should be non-zero)?
Assistant (inferred SQL query): SELECT COUNT(albums.AlbumId) 
FROM albums 
JOIN tracks ON album...

> Running module llm with input: 
messages: You are trying to generate a proper natural language query given a user input.

This query will then be interpreted by a downstream text-to-SQL agent which
will convert the query to a SQL statement. I...

> Running module sql_query_engine with input: 
input: assistant: Given the previous error, it seems that the SQL query was incorrect because it attempted to use an aggregate function (`COUNT`) in the `WHERE` clause, which is not allowed. Additionally, th...

> Running module output_component with input: 
output: The number of albums released by the artist who composed the track 'Restless and Wild' is 1.

> Inferred SQL Query: SELECT COUNT(DISTINCT albums.AlbumId) AS NumAlbums
FROM tracks
JOIN albums ON tracks.AlbumId = albums.AlbumId
JOIN artists ON albums.ArtistId = artists.ArtistId
WHERE tracks.Name = 'Restless and Wild'
GROUP BY artists.ArtistId
HAVING NumAlbums > 0;
> SQL Response: The number of albums released by the artist who composed the track 'Restless and Wild' is 1.
> Running module input with input: 
state: {'sources': [], 'memory': ChatMemoryBuffer(token_limit=3000, tokenizer_fn=functools.partial(<bound method Encoding.encode of <Encoding 'cl100k_base'>>, allowed_special='all'), chat_store=SimpleChatSto...
task: task_id='2d8a63de-7410-4422-98f3-f0ca41884f58' input="How many albums did the artist who wrote 'Restless and Wild' release? (answer should be non-zero)?" memory=ChatMemoryBuffer(token_limit=3000, toke...

> Running module retry_prompt with input: 
input: How many albums did the artist who wrote 'Restless and Wild' release? (answer should be non-zero)?
convo_history: User: How many albums did the artist who wrote 'Restless and Wild' release? (answer should be non-zero)?
Assistant (inferred SQL query): SELECT COUNT(albums.AlbumId) 
FROM albums 
JOIN tracks ON album...

> Running module llm with input: 
messages: You are trying to generate a proper natural language query given a user input.

This query will then be interpreted by a downstream text-to-SQL agent which
will convert the query to a SQL statement. I...

> Running module sql_query_engine with input: 
input: assistant: Given the conversation history, it seems that the previous SQL query was successful in retrieving the number of albums released by the artist who composed the track 'Restless and Wild'. How...

> Running module output_component with input: 
output: I apologize, but there seems to be an error in the SQL query provided. Please double-check the syntax and try again.

> Inferred SQL Query: SELECT COUNT(DISTINCT albums.AlbumId) AS TotalAlbums
FROM albums
JOIN artists ON albums.ArtistId = artists.ArtistId
WHERE artists.ArtistId = (
    SELECT artists.ArtistId
    FROM tracks
    JOIN albums ON tracks.AlbumId = albums.AlbumId
    JOIN artists ON albums.ArtistId = artists.ArtistId
    WHERE tracks.Name = 'Restless and Wild'
    LIMIT 1
)
AND TotalAlbums > 0;
> SQL Response: I apologize, but there seems to be an error in the SQL query provided. Please double-check the syntax and try again.
I apologize, but there seems to be an error in the SQL query provided. Please double-check the syntax and try again.