模块 agent

Agent

一个 Haystack 组件，它实现了支持提供者无关的聊天模型的工具使用 Agent。

该组件处理消息并执行工具，直到满足退出条件。退出条件可以通过直接文本响应或调用特定的指定工具来触发。可以指定多个退出条件。

当您在不带工具的情况下调用 Agent 时，它将充当 ChatGenerator，生成一个响应，然后退出。

使用示例

from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.tools.tool import Tool

tools = [Tool(name="calculator", description="..."), Tool(name="search", description="...")]

agent = Agent(
    chat_generator=OpenAIChatGenerator(),
    tools=tools,
    exit_conditions=["search"],
)

# Run the agent
result = agent.run(
    messages=[ChatMessage.from_user("Find information about Haystack")]
)

assert "messages" in result  # Contains conversation history

Agent.__init__

def __init__(*,
             chat_generator: ChatGenerator,
             tools: Optional[Union[list[Tool], Toolset]] = None,
             system_prompt: Optional[str] = None,
             exit_conditions: Optional[list[str]] = None,
             state_schema: Optional[dict[str, Any]] = None,
             max_agent_steps: int = 100,
             streaming_callback: Optional[StreamingCallbackT] = None,
             raise_on_tool_invocation_failure: bool = False,
             tool_invoker_kwargs: Optional[dict[str, Any]] = None) -> None

初始化 Agent 组件。

参数:

• chat_generator: Agent 应使用的聊天生成器实例。它必须支持工具。
• tools: Agent 可以使用的 Tool 对象或 Toolset 列表。
• system_prompt: Agent 的系统提示。
• exit_conditions: 将导致 Agent 返回的条件列表。如果 Agent 在生成不带工具调用的消息时应返回，则可以包含“text”，或者包含在工具执行后将导致 Agent 返回的工具名称。默认为 [“text”]。
• state_schema: Agent 及其工具使用的运行时状态的模式。
• max_agent_steps: Agent 在停止前将运行的最大步数。默认为 100。如果 Agent 超过此步数，它将停止并返回当前状态。
• streaming_callback: 当从 LLM 流式传输响应时将调用的回调。可以使用相同回调来发出工具调用的工具结果。
• raise_on_tool_invocation_failure: 工具调用失败时 Agent 是否应引发异常？如果设置为 False，异常将被转换为聊天消息并传递给 LLM。
• tool_invoker_kwargs: 要传递给 ToolInvoker 的其他关键字参数。

引发:

• TypeError: 如果 chat_generator 的 run 方法不支持 tools 参数。
• ValueError: 如果 exit_conditions 无效。

Agent.warm_up

def warm_up() -> None

预热 Agent。

Agent.to_dict

def to_dict() -> dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典

Agent.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "Agent"

从字典反序列化 Agent。

参数:

• data: 要从中反序列化的字典

返回值:

反序列化后的 Agent

Agent.run

def run(messages: list[ChatMessage],
        streaming_callback: Optional[StreamingCallbackT] = None,
        *,
        break_point: Optional[AgentBreakpoint] = None,
        snapshot: Optional[AgentSnapshot] = None,
        system_prompt: Optional[str] = None,
        **kwargs: Any) -> dict[str, Any]

处理消息并执行工具，直到满足退出条件。

参数:

• messages: 要处理的 Haystack ChatMessage 对象列表。
• streaming_callback: 当从 LLM 流式传输响应时将调用的回调。可以使用相同回调来发出工具调用的工具结果。
• break_point: AgentBreakpoint，可以是“chat_generator”的断点，也可以是“tool_invoker”的 ToolBreakpoint。
• snapshot: 包含已保存 Agent 执行快照的字典。快照包含有关从上次中断处重新启动 Agent 执行的相关信息。
• system_prompt: Agent 的系统提示。如果提供，它将覆盖默认系统提示。
• kwargs: 要传递给 Agent 使用的状态模式的其他数据。键必须与 Agent 中定义的模式匹配state_schema.

引发:

• RuntimeError: 如果在调用之前 Agent 组件未预热run().
• BreakpointException: 如果触发了 Agent 断点。

返回值:

包含以下键的字典

• "messages": Agent 运行期间交换的所有消息列表。
• "last_message": Agent 运行期间交换的最后一条消息。
• 在 Agent 的state_schema.

Agent.run_async

async def run_async(messages: list[ChatMessage],
                    streaming_callback: Optional[StreamingCallbackT] = None,
                    *,
                    break_point: Optional[AgentBreakpoint] = None,
                    snapshot: Optional[AgentSnapshot] = None,
                    system_prompt: Optional[str] = None,
                    **kwargs: Any) -> dict[str, Any]

异步处理消息并执行工具，直到满足退出条件。

这是run 方法的异步版本。它遵循相同的逻辑，但尽可能使用异步操作，例如调用 ChatGenerator 的run_async 方法（如果可用）。

参数:

• messages: 要处理的 Haystack ChatMessage 对象列表。
• streaming_callback: 当从 LLM 流式传输响应时将调用的异步回调。可以使用相同回调来发出工具调用的工具结果。
• break_point: AgentBreakpoint，可以是“chat_generator”的断点，也可以是“tool_invoker”的 ToolBreakpoint。
• snapshot: 包含已保存 Agent 执行快照的字典。快照包含有关从上次中断处重新启动 Agent 执行的相关信息。
• system_prompt: Agent 的系统提示。如果提供，它将覆盖默认系统提示。
• kwargs: 要传递给 Agent 使用的状态模式的其他数据。键必须与 Agent 中定义的模式匹配state_schema.

引发:

• RuntimeError: 如果在调用之前 Agent 组件未预热run_async().
• BreakpointException: 如果触发了 Agent 断点。

返回值:

包含以下键的字典

• "messages": Agent 运行期间交换的所有消息列表。
• "last_message": Agent 运行期间交换的最后一条消息。
• 在 Agent 的state_schema.

模块 state/state

状态

State 是一个容器，用于在 Agent 及其工具执行期间存储共享信息。

例如，State 可用于存储文档、上下文和中间结果。

内部它包装了一个由schema 定义的_data 字典。每个模式条目都有

  "parameter_name": {
    "type": SomeType,  # expected type
    "handler": Optional[Callable[[Any, Any], Any]]  # merge/update function
  }

Handlers 控制在使用set() 方法时如何合并值

• 对于列表类型：默认为merge_lists（连接列表）
• 对于其他类型：默认为replace_values（覆盖现有值）

一个类型为list[ChatMessage] 的

`messages` 字段会自动添加到模式中。这使得 Agent 能够读写相同的上下文。

使用示例

from haystack.components.agents.state import State

my_state = State(
    schema={"gh_repo_name": {"type": str}, "user_name": {"type": str}},
    data={"gh_repo_name": "my_repo", "user_name": "my_user_name"}
)

State.__init__

def __init__(schema: dict[str, Any], data: Optional[dict[str, Any]] = None)

使用模式和可选数据初始化 State 对象。

参数:

• schema: 映射参数名称及其类型和处理程序配置的字典。类型必须是有效的 Python 类型，处理程序必须是可调用函数或 None。如果处理程序为 None，则使用该类型的默认处理程序。默认处理程序是： - 对于列表类型haystack.agents.state.state_utils.merge_lists - 对于所有其他类型haystack.agents.state.state_utils.replace_values
• data: 可选的初始数据字典，用于填充状态

State.get

def get(key: str, default: Any = None) -> Any

按键从状态中检索值。

参数:

• key: 在状态中查找的键
• default: 如果找不到键，则返回的值

返回值:

与键关联的值，如果未找到则返回默认值

State.set

def set(key: str,
        value: Any,
        handler_override: Optional[Callable[[Any, Any], Any]] = None) -> None

根据模式规则在状态中设置或合并值。

值根据以下规则合并或覆盖

• 如果提供了 handler_override，则使用它
• 否则使用模式中为“key”定义的处理程序

参数:

• key: 要存储值的键
• value: 要存储或合并的值
• handler_override: 可选函数，用于覆盖默认合并行为

State.data

@property
def data()

状态的所有当前数据。

State.has

def has(key: str) -> bool

检查键是否存在于状态中。

参数:

• key: 要检查其存在性的键

返回值:

如果键存在于状态中，则为 True，否则为 False

State.to_dict

def to_dict()

将 State 对象转换为字典。

State.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any])

将字典转换回 State 对象。