Haystack Agent 組件深度解析基于 ChatGenerator 與 State 的工具型 Agent 架構指南【免費下載鏈接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.項目地址: https://gitcode.com/GitHub_Trending/ha/haystack本文是 Haystack 官方參考文檔 agents-api 的技術解讀與實戰手冊。文章以該文檔定義的Agent與State兩大核心類為主線結合當前倉庫源碼haystack/components/agents/agent.py、haystack/components/agents/state/state.py深入展開你將掌握 Agent 的構造參數、退出條件、運行循環、序列化機制以及 State 狀態容器的 schema 合并語義并能在真實項目中用幾十行代碼搭建一個可調工具、可共享上下文的 LLM Agent。一、Agent 是什么一個工具型 LLM 組件的整體定位根據參考文檔的定義Agent是一個實現工具使用型 Agent的 Haystack 組件其核心特性是與聊天模型供應商無關只要 ChatGenerator 的run()方法支持tools參數就可以作為 Agent 的推理引擎例如OpenAIChatGenerator等。循環處理消息并執行工具組件會不斷思考 → 調用工具 → 觀察工具結果 → 再思考直到滿足某個退出條件exit condition才停止。退出條件可配置既可以在模型直接生成文本回復時退出text也可以在指定的某個工具被執行完后退出工具名多個退出條件可同時指定。無工具即退化為純 LLM當不傳入任何工具時Agent 的行為等價于一個 ChatGenerator產生一條回復后立即退出。從當前倉庫的類注釋haystack/components/agents/agent.py可以看到同樣定位的描述A tool-using Agent powered by a large language model. The Agent processes messages and calls tools until it meets an exit condition.該組件在haystack.components.agents包下導出與State一起對外可見見 haystack/components/agents/init.pyfrom haystack.components.agents import Agent from haystack.components.agents.state import State二、Agent 構造參數詳解參考文檔給出了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各參數含義如下表參數與說明以參考文檔為準參數類型默認值說明chat_generatorChatGenerator必填Agent 使用的聊天生成器實例必須支持 tools其run()方法需接受tools參數。toolslist[Tool]/ToolsetNoneAgent 可使用的工具列表或工具集。system_promptstrNone系統提示詞用于約束 Agent 行為。exit_conditionslist[str][text]退出條件列表。可包含text模型生成無工具調用的消息時退出或工具名該工具執行完成后退出。state_schemadict[str, Any]None工具運行期共享狀態的 schema供工具讀寫。max_agent_stepsint100Agent 運行的最大步數上限達到后停止并返回當前狀態。streaming_callbackStreamingCallbackTNoneLLM 流式輸出時的回調同一回調也可配置為在工具被調用時輸出工具結果。raise_on_tool_invocation_failureboolFalse工具調用失敗時是否拋出異常。為False時異常會被轉換為一條聊天消息交給 LLM 繼續處理。tool_invoker_kwargsdict[str, Any]None透傳給 ToolInvoker 的額外關鍵字參數。異常約束TypeError當傳入的chat_generator的run()方法不支持tools參數時拋出。ValueError當exit_conditions不合法時拋出。從當前倉庫源碼看構造時的工具支持校驗發生在__init__中haystack/components/agents/agent.pyAgent通過inspect.signature(chat_generator.run)檢查tools是否在參數列表里若傳入了工具而生成器不支持會立刻拋出TypeError并給出明確提示。2.1 當前倉庫中的參數演進從當前倉庫源碼可以確認Agent.__init__在后續版本中進一步擴展這些屬于文檔之后的演進供參考user_prompt可復用的用戶提示模板支持 Jinja2 模板變量運行時追加到傳入的 messages 之后。required_variables聲明user_prompt/system_prompt中必須由運行期提供的模板變量缺省為*全部必填設為None則全部可選。tool_concurrency_limit并行執行工具調用的最大并發數默認4設為1可關閉并行工具執行。hooks在before_run、before_llm、before_tool、after_tool、on_exit、after_run等鉤子點上注冊的 Hook 列表鉤子接收實時State并可通過修改它影響運行流程。三、快速上手文檔最小示例與完整可運行示例3.1 參考文檔的最小示例參考文檔給出的最小用法如下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(namecalculator, description...), Tool(namesearch, description...)] agent Agent( chat_generatorOpenAIChatGenerator(), toolstools, exit_conditions[search], ) # Run the agent result agent.run( messages[ChatMessage.from_user(Find information about Haystack)] ) assert messages in result # Contains conversation history要點exit_conditions[search]表示當模型調用了名為search的工具且執行成功后Agent 即結束本次運行。返回的result至少包含messages整個對話歷史assert語句驗證了這一點。3.2 完整實戰示例搜索 計算的 Agent參考文檔的最小示例只展示了骨架。結合源碼類注釋中的完整示例haystack/components/agents/agent.py下面是一個查詢法國小費習慣 → 用計算器算小費的端到端可運行示例。它演示了用tool裝飾器定義工具、通過Annotated描述參數、通過system_prompt約束行為以及用streaming_callback打印流式輸出from typing import Annotated, Literal from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.generators.utils import print_streaming_chunk from haystack.dataclasses import ChatMessage from haystack.tools import tool tool def search(query: Annotated[str, The search query]) - str: Search for information on the web. # 實際項目中此處應調用真實搜索 API return In France, a 15% service charge is typically included, but leaving 5-10% extra is appreciated. tool def calculator( operation: Annotated[Literal[multiply, percentage], The mathematical operation to perform], a: Annotated[float, First number], b: Annotated[float, Second number], ) - float: Perform mathematical calculations. if operation multiply: return a * b elif operation percentage: return (a / 100) * b return 0 agent Agent( system_prompt( You are a helpful assistant. Use the search tool to find information about a users question and the calculator tool to perform math. ), chat_generatorOpenAIChatGenerator(), tools[search, calculator], streaming_callbackprint_streaming_chunk, ) result agent.run( messages[ChatMessage.from_user(Calculate the appropriate tip for an €85 meal in France)] ) # 獲取最終回復 print(result[last_message].text)tool裝飾器會把普通函數轉換為Tool對象name、description來自函數名與 docstringparameters則根據帶Annotated類型注解的函數簽名自動生成符合 JSON Schema 的參數定義。Tool數據類的完整字段定義見 haystack/tools/tool.pyname、description、parameters、function同步函數、async_function協程函數、outputs_to_string、inputs_from_state、outputs_to_state。3.3 使用 Toolset 組織工具當工具數量變多時可以用Toolset把相關工具打包成一個集合傳給 Agenthaystack/tools/toolset.pyfrom typing import Annotated from haystack.tools import tool, Toolset from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator tool def add(a: Annotated[int, first number], b: Annotated[int, second number]) - int: Add two numbers. return a b tool def subtract(a: Annotated[int, first number], b: Annotated[int, second number]) - int: Subtract b from a. return a - b math_toolset Toolset([add, subtract]) agent Agent(chat_generatorOpenAIChatGenerator(), toolsmath_toolset)Toolset實現了集合接口__iter__、__contains__、__len__、__getitem__因此可以像工具列表一樣被Agent與聊天生成器消費。同時它也是動態工具加載的基類通過子類化Toolset并覆寫warm_up()在啟動時把工具賦給self.tools與to_dict()/from_dict()序列化端點描述符而非工具實例可以實現從 OpenAPI URL、MCP 服務器等外部來源動態加載工具。四、運行 Agentrun 與 run_async4.1 run 方法參考文檔給出的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, tools: Optional[Union[list[Tool], Toolset, list[str]]] None, **kwargs: Any) - dict[str, Any]參數說明以參考文檔為準參數說明messages要處理的 HaystackChatMessage列表。streaming_callbackLLM 流式輸出時的回調同一回調可配置為在工具被調用時輸出工具結果。break_point一個AgentBreakpoint可以是針對chat_generator的Breakpoint或針對tool_invoker的ToolBreakpoint。snapshot之前保存的 Agent 執行快照包含從上次中斷處恢復執行所需的全部信息。system_prompt本次運行的系統提示詞提供時覆蓋默認系統提示詞。tools本次運行使用的工具Tool列表、Toolset或工具名字符串列表。傳工具名時會從 Agent 構造時配置的工具中按名選取。kwargs傳給 State schema 的額外數據鍵必須與state_schema中定義的一致。異常RuntimeError調用run()前未對 Agent 執行warm_up()。BreakpointException觸發了 agent breakpoint。返回值字典messagesAgent 運行期間交換的全部消息列表。last_message運行期間交換的最后一條消息。以及state_schema中定義的任何額外鍵。4.2 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, tools: Optional[Union[list[Tool], Toolset, list[str]]] None, **kwargs: Any) - dict[str, Any]run_async是run的異步版本遵循相同邏輯但在可能的地方使用異步操作例如若 ChatGenerator 提供了run_async方法則優先調用它。對于同步的生成器當前倉庫通過_execute_component_async將調用派發到工作線程執行asyncio.to_thread并保持 tracing span 上下文工具的invoke_async在無async_function時也會回退到asyncio.to_thread運行同步函數見 haystack/tools/tool.py。4.3 運行結果的運行時元數據當前倉庫參考文檔的返回值說明聚焦于messages、last_message與state_schema鍵。從當前倉庫源碼haystack/components/agents/agent.py可以看到Agent 會在運行期自動向 State 中寫入一批運行元數據鍵并以輸出形式暴露可用于下游路由與監控輸出鍵類型含義step_countint已執行的步數。一步 一次 chat-generator 調用 該次調用中模型請求的所有工具調用若有的執行。token_usagedict本次運行中所有 LLM 調用的 token 用量聚合由每條 LLM 消息的meta[usage]累加而來。tool_call_countsdict[str, int]各工具被調用的次數映射。exit_reasonstrAgent 停止的原因可用于ConditionalRouter之類的下游路由。取值包括text模型返回了無工具調用的完整回復、length/content_filter模型返回了不完整回復、滿足工具退出條件的工具名此時last_message是該工具的結果、max_agent_steps達到步數上限、或鉤子通過stop_run狀態鍵提供的自定義原因。這些鍵由常量_RUN_METADATA_STATE_KEYS定義屬于保留鍵用戶不能在自己的state_schema中重定義它們。Agent 內部還有一組完全私有的_INTERNAL_STATE_KEYS如continue_run、stop_run、tools、hook_context、context_tokens它們不暴露為輸入或輸出。五、退出條件exit_conditions機制詳解參考文檔指出退出條件可以是text模型生成無工具調用的消息時返回也可以是工具名該工具執行完成后返回多個條件可同時指定默認是[text]。從源碼看退出判定有兩套邏輯haystack/components/agents/agent.py1. 模型退出原因判定_get_model_exit_reason當最后一條消息來自 assistant 且不含工具調用時若meta[finish_reason]為length返回length若為content_filter返回content_filter若消息含文本返回text空響應且無終止原因時不退出——這保留了 Agent 對 ChatGenerator 丟棄的畸形工具調用的恢復能力。2. 工具退出條件判定_check_exit_conditions當exit_conditions不只包含text時遍歷 LLM 消息中的每個工具調用只要模型調用了至少一個在exit_conditions中列出的工具且該工具執行未出錯就返回該工具名作為退出原因若被調用的退出條件工具執行出錯則取消退出即使同一步中有其他退出條件工具成功多個退出條件工具在同一步被并行調用時返回第一個遇到的工具名。測試用例test/components/agents/test_agent.py大量驗證了這類組合場景例如exit_conditions[text, weather_tool]、exit_conditions[weather_tool, search]等配置。六、StateAgent 與工具共享的運行時狀態容器參考文檔用專門一節介紹了State。它是在 Agent 及其工具執行期間存儲共享信息的容器例如文檔、上下文和中間結果都可以放入其中使 Agent 與工具能夠讀寫同一份上下文。6.1 schema 結構與合并語義State內部包裝了一個由schema定義的_data字典每個 schema 條目形如parameter_name: { type: SomeType, # 期望的類型 handler: Optional[Callable[[Any, Any], Any]] # 合并/更新函數 }handler 控制set()時的合并方式列表類型默認使用merge_lists拼接列表其他類型默認使用replace_values用新值覆蓋舊值。參考文檔特別強調一個messages字段類型list[ChatMessage]會被自動加入 schema這正是 Agent 能持續讀寫同一對話上下文的機制。6.2 State 使用示例參考文檔示例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} )6.3 State 的完整 API參考文檔給出了以下方法簽名與說明__init__def __init__(schema: dict[str, Any], data: Optional[dict[str, Any]] None)schema參數名到類型與 handler 配置的映射。type必須是合法的 Python 類型handler必須是可調用對象或None。handler 為None時使用類型默認 handler列表類型為haystack.agents.state.state_utils.merge_lists其他類型為haystack.agents.state.state_utils.replace_values。data可選的初始數據字典。get(key, defaultNone)按 key 讀取值未找到時返回default。set(key, value, handler_overrideNone)按 schema 規則寫入或合并值。合并規則為若給了handler_override則使用它否則使用 schema 中該 key 定義的 handler。datapropertyState 當前的全部數據。has(key)判斷 key 是否存在于 state 中返回布爾值。to_dict()將 State 序列化為字典。from_dict(data)classmethod從字典反序列化回 State 對象。6.4 源碼級深入校驗、默認 handler 與讀寫語義從 haystack/components/agents/state/state.py 可以確認幾個重要實現細節schema 校驗構造時_validate_schema會檢查每個條目必須有type字段、type必須是合法 Python 類型支持普通類、list[str]等泛型、Union/Optional聯合類型、handler 必須可調用或為None特別地messages鍵的類型必須是list[ChatMessage]。默認 handler 注入構造時會對未顯式指定 handler 的條目按類型補上默認 handler列表用merge_lists其余用replace_values實現位于 haystack/components/agents/state/state_utils.py。merge_lists會把非列表值包裝成列表后拼接current為None視為空列表replace_values則直接返回新值。set()的健壯性若寫入的 key 不在 schema 中set()會拋出ValueErrorget()返回的是值的深拷貝避免外部意外修改內部數據。Agent 集成在Agent.__init__中用戶提供的state_schema會被淺拷貝進resolved_state_schema若沒有messages鍵則自動補充{type: list[ChatMessage], handler: merge_lists}再疊加上運行元數據鍵與內部鍵最終據此生成組件的輸入/輸出 socket見 haystack/components/agents/agent.py。序列化to_dict()序列化 schema 中的類型serialize_type與 handler 可調用serialize_callable并用字段級回退保證單個不可序列化的值不會拖垮整個 State 的序列化from_dict()相應地進行反序列化見 haystack/components/agents/state/state.py。6.5 工具與 State 的雙向數據流State 之所以重要是因為工具可以通過Tool的inputs_from_state與outputs_to_state字段直接與 Agent 共享數據haystack/tools/tool.pyinputs_from_state把 State 鍵映射到工具參數名。例如{repository: repo}表示把 State 中的repository值傳給工具的repo參數。工具構造時會校驗這些參數名確實存在于工具的函數簽名或 JSON schema 中。outputs_to_state把工具輸出映射回 State 鍵并可附帶 handler。例如{documents: {source: docs, handler: custom_handler}}表示把工具結果字典中docs字段經custom_handler處理后寫入 State 的documents鍵省略source時整個工具結果傳給 handler。在工具執行層haystack/components/agents/tool_calling.py_merge_tool_outputs_into_state負責把工具輸出按outputs_to_state配置寫入 State_build_tool_result_message/_process_tool_output則根據outputs_to_string把工具結果轉換為字符串默認對非字符串結果做 JSON 序列化失敗時回退到str()或按raw_result原樣返回用于圖片等TextContent/ImageContent場景。七、warm_up 與序列化7.1 warm_updef warm_up() - None參考文檔說明其作用為Warm up the Agent。從源碼看haystack/components/agents/agent.pywarm_up()會依次完成三件事預熱全部工具warm_up_tools、預熱鉤子warm_up_hooks、預熱底層 chat generator若其有warm_up方法。run()內部會在正式執行前調用warm_up()。工具與 Toolset 的warm_up()約定是冪等的——例如動態加載工具的 Toolset 子類應通過自身狀態做保護if self._client is not None: return因為預熱可能在每次運行前被調用。7.2 to_dict 與 from_dictdef to_dict() - dict[str, Any] def from_dict(cls, data: dict[str, Any]) - Agent參考文檔說明to_dict將組件序列化為字典from_dict從字典反序列化。這正是 Haystack Pipeline 的 YAML 序列化機制的基礎——Agent 可以被嵌入 Pipeline并整體導出/導入。從源碼看haystack/components/agents/agent.pyto_dict()會序列化chat_generator組件字典、tools工具或 Toolset 的序列化形式、system_prompt、exit_conditions、state_schema類型與 handler 可調用均被序列化、max_agent_steps、streaming_callback可調用序列化、raise_on_tool_invocation_failure等from_dict()則按需反序列化 chat generator、state_schema、streaming_callback、tools、hooks。State自身的to_dict()/from_dict()與此同理見 haystack/components/agents/state/state.py。八、運行循環的源碼級剖析雖然參考文檔沒有畫出運行循環細節但從當前倉庫源碼haystack/components/agents/agent.py可以完整還原 Agent 的一次運行過程run()首先調用warm_up()然后通過_initialize_fresh_execution構建執行上下文把kwargs中與state_schema匹配的鍵填充進State初始化messages、step_count、token_usage、tool_call_counts、exit_reason等狀態解析流式回調并組裝 chat generator 與工具執行的輸入。進入while exe_context.counter self.max_agent_steps主循環每輪執行_run_step先重新展平工具讓SearchableToolset這類動態工具集能持續暴露新發現的工具并做重名檢查運行before_llm鉤子檢查stop_run狀態鍵是否要求提前終止調用chat_generator.run(messages..., tools...)把回復寫入 State并記錄 token 用量與上下文 token 數若無工具或模型產出了無工具調用的終止性回復則按退出原因結束本步走on_exit鉤子判定是否繼續否則運行before_tool鉤子重新讀取 State 中最后一條消息的待執行工具調用鉤子可以改寫、拒絕這些調用調用_run_tool并發執行工具并發上限為tool_concurrency_limit把工具結果消息寫回 State記錄各工具調用次數運行after_tool鉤子最后檢查工具退出條件滿足則設置exit_reason并結束。若循環因max_agent_steps耗盡而自然結束未break則設置exit_reason max_agent_steps并記錄警告日志。運行after_run鉤子從 State 中剔除內部鍵后構建返回值messages、last_message取消息列表最后一條、運行元數據以及state_schema中定義的所有鍵。整個循環還被 Haystack 的 tracing 體系包裹每次運行會產生haystack.agent.runspan每步產生haystack.agent.stepspanLLM 調用與工具執行分別有各自的子 span便于在監控系統中觀測 Agent 的逐步行為。九、注意事項與邊界條件聊天生成器必須支持工具為 Agent 配置工具時chat_generator.run()必須接受tools參數否則構造時即拋出TypeError。若在運行期傳入工具而生成器不支持同樣會拋錯。無工具即純 LLMtools為空時Agent 生成一條回復后立即退出行為等價于 ChatGenerator。步數上限是硬邊界max_agent_steps默認 100保證即使模型反復請求工具調用也不會無限循環達到上限時返回當前狀態并以max_agent_steps作為exit_reason。一步的定義是一次 LLM 調用 該次調用請求的全部工具調用執行。工具失敗的可恢復性raise_on_tool_invocation_failureFalse默認時工具調用失敗不會中斷運行而是把異常轉換為聊天消息交回給 LLM讓模型可以換一種方式繼續設為True則直接拋出。State 的鍵約束寫入 State 的 key 必須存在于 schema 中否則set()拋ValueErrormessages類型必須是list[ChatMessage]運行元數據鍵與內部鍵為保留鍵不可在state_schema中重定義。快照與斷點通過run(snapshot...)可以從保存的執行快照恢復運行break_point可在 chat_generator 或 tool_invoker 處暫停執行這為調試與人工介入Human-in-the-Loop提供了基礎。Warm-up 前置條件雖然run()內部會自動預熱但若繞過run()直接依賴組件狀態如把 Agent 嵌入 Pipeline應保證 Pipeline 在運行前完成 warm-up。十、總結參考文檔圍繞兩個核心類勾勒了 Haystack Agent 的完整能力Agent工具使用、退出條件、同步/異步運行、序列化與Stateschema 化共享狀態、類型化合并語義。結合當前倉庫源碼可以看到Agent 本質上是一個LLM 循環控制器通過 State 把對話消息、工具調用、運行元數據與用戶自定義上下文統一管理通過exit_conditions精確控制終止時機并通過Tool/Toolset機制把任意函數、組件乃至外部服務接入模型推理回路。無論是構建 RAG Agent、多工具編排 Agent還是將 Agent 嵌入 Haystack PipelineAgentState都是值得首先掌握的組件組合。相關測試test/components/agents/test_agent.py、test/components/agents/test_state_class.py提供了大量可對照的行為示例可作為進一步學習的起點。【免費下載鏈接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.項目地址: https://gitcode.com/GitHub_Trending/ha/haystack創作聲明：本文部分內容由AI輔助生成（AIGC），僅供參考