整體入門demo

教程概覽

歡迎來到LangGraph教程！ 這些筆記本通過構(gòu)建各種語言代理和應(yīng)用程序，介紹了如何使用LangGraph。

快速入門（Quick Start）

快速入門部分通過一個全面的入門教程，幫助您從零開始構(gòu)建一個代理，學(xué)習(xí)LangGraph的基礎(chǔ)知識。

1. 快速入門：

   • 內(nèi)容簡介： 在本教程中，您將使用LangGraph構(gòu)建一個支持型聊天機器人（support chatbot）。
   • 目標： 學(xué)習(xí)LangGraph的基本概念和操作，通過實際構(gòu)建一個聊天機器人，掌握創(chuàng)建和管理語言代理的基本技能。

2. LangGraph Cloud 快速入門：

   • 內(nèi)容簡介： 在本教程中，您將構(gòu)建并部署一個代理到LangGraph Cloud。
   • 目標： 學(xué)習(xí)如何在云端環(huán)境中部署和管理LangGraph代理，了解云部署的優(yōu)勢和操作步驟。

用例（Use cases）

用例部分通過具體的示例實現(xiàn)，展示了為特定場景設(shè)計的圖結(jié)構(gòu)和常見的設(shè)計模式。

聊天機器人（Chatbots）

1. 客戶支持（Customer Support）：

   • 內(nèi)容簡介： 構(gòu)建一個客戶支持聊天機器人，用于管理航班、酒店預(yù)訂、租車等任務(wù)。
   • 目標： 學(xué)習(xí)如何設(shè)計和實現(xiàn)一個多功能的客戶支持系統(tǒng)，提高用戶滿意度和自動化水平。

2. 根據(jù)用戶需求生成提示（Prompt Generation from User Requirements）：

   • 內(nèi)容簡介： 構(gòu)建一個信息收集型聊天機器人。
   • 目標： 學(xué)習(xí)如何設(shè)計一個能夠根據(jù)用戶需求生成相關(guān)提示和問題的聊天機器人，提升信息收集效率。

3. 代碼助手（Code Assistant）：

   • 內(nèi)容簡介： 構(gòu)建一個代碼分析和生成助手。
   • 目標： 學(xué)習(xí)如何設(shè)計一個能夠分析、調(diào)試和生成代碼的智能助手，輔助開發(fā)者提高工作效率。

RAG（Retrieval-Augmented Generation）

1. Agentic RAG：

   • 內(nèi)容簡介： 使用代理在回答用戶問題之前，檢索最相關(guān)的信息。
   • 目標： 學(xué)習(xí)如何結(jié)合信息檢索和生成模型，提高回答的準確性和相關(guān)性。

2. 自適應(yīng) RAG（Adaptive RAG）：

   • 內(nèi)容簡介： 自適應(yīng)RAG是一種將查詢分析與主動/自我糾正RAG相結(jié)合的策略。
   • 實現(xiàn)鏈接： 論文鏈接
   • 本地LLM版本： 使用本地LLM實現(xiàn)的自適應(yīng)RAG。

3. 糾正性 RAG（Corrective RAG）：

   • 內(nèi)容簡介： 使用LLM評估檢索信息的質(zhì)量，如果質(zhì)量低，則嘗試從其他來源檢索信息。
   • 實現(xiàn)鏈接： 論文鏈接
   • 本地LLM版本： 使用本地LLM實現(xiàn)的糾正性RAG。

4. 自我反思 RAG（Self-RAG）：

   • 內(nèi)容簡介： 自我RAG是一種結(jié)合自我反思/自我評估對檢索文檔和生成內(nèi)容進行優(yōu)化的策略。
   • 實現(xiàn)鏈接： 論文鏈接
   • 本地LLM版本： 使用本地LLM實現(xiàn)的自我RAG。

5. SQL代理（SQL Agent）：

   • 內(nèi)容簡介： 構(gòu)建一個能夠回答關(guān)于SQL數(shù)據(jù)庫問題的SQL代理。
   • 目標： 學(xué)習(xí)如何設(shè)計一個能夠與SQL數(shù)據(jù)庫交互，執(zhí)行查詢并返回結(jié)果的智能代理。

代理架構(gòu)（Agent Architectures）

1. 多代理系統(tǒng)（Multi-Agent Systems）：

   • 網(wǎng)絡(luò)（Network）：
     • 內(nèi)容簡介： 使兩個或更多代理協(xié)作完成任務(wù)。
     • 目標： 學(xué)習(xí)如何設(shè)計和實現(xiàn)多個代理之間的協(xié)作機制，提高任務(wù)執(zhí)行的效率和復(fù)雜性處理能力。
   • 監(jiān)督者（Supervisor）：
     • 內(nèi)容簡介： 使用LLM來協(xié)調(diào)和委派任務(wù)給各個獨立的代理。
     • 目標： 學(xué)習(xí)如何設(shè)計一個監(jiān)督者代理，負責(zé)管理和分配任務(wù)，提高系統(tǒng)的整體協(xié)調(diào)性。
   • 層級團隊（Hierarchical Teams）：
     • 內(nèi)容簡介： 協(xié)調(diào)嵌套團隊的代理以解決問題。
     • 目標： 學(xué)習(xí)如何設(shè)計層級化的代理團隊，處理復(fù)雜問題，提升系統(tǒng)的組織和管理能力。

2. 規(guī)劃代理（Planning Agents）：

   • 計劃與執(zhí)行（Plan-and-Execute）：
     • 內(nèi)容簡介： 實現(xiàn)一個基本的規(guī)劃與執(zhí)行代理。
     • 目標： 學(xué)習(xí)如何設(shè)計一個能夠制定計劃并執(zhí)行任務(wù)的代理，提高任務(wù)管理的系統(tǒng)性和效率。
   • 無需觀察的推理（Reasoning without Observation）：
     • 內(nèi)容簡介： 通過將觀察結(jié)果保存為變量，減少重新規(guī)劃的需求。
     • 目標： 學(xué)習(xí)如何優(yōu)化代理的推理過程，減少不必要的重新規(guī)劃，提高系統(tǒng)的響應(yīng)速度。
   • LLMCompiler：
     • 內(nèi)容簡介： 從計劃器流式傳輸并提前執(zhí)行DAG任務(wù)。
     • 目標： 學(xué)習(xí)如何使用LLMCompiler，通過流式和提前執(zhí)行任務(wù)來優(yōu)化任務(wù)執(zhí)行流程。

反思與批評（Reflection & Critique）

1. 基本反思（Basic Reflection）：

   • 內(nèi)容簡介： 提示代理對其輸出進行反思和修正。
   • 目標： 學(xué)習(xí)如何設(shè)計一個能夠自我反思并改進輸出的代理，提升系統(tǒng)的智能化水平。

2. 反思（Reflexion）：

   • 內(nèi)容簡介： 批評缺失和多余的細節(jié)，以指導(dǎo)下一步行動。
   • 目標： 學(xué)習(xí)如何通過自我評估，優(yōu)化代理的決策過程和輸出質(zhì)量。

3. 思維樹（Tree of Thoughts）：

   • 內(nèi)容簡介： 使用評分樹搜索候選解決方案。
   • 目標： 學(xué)習(xí)如何設(shè)計一個通過評分和搜索優(yōu)化思維過程的代理，提高問題解決的準確性和效率。

4. 語言代理樹搜索（Language Agent Tree Search）：

   • 內(nèi)容簡介： 使用反思和獎勵驅(qū)動蒙特卡洛樹搜索。
   • 目標： 學(xué)習(xí)如何結(jié)合反思機制和獎勵系統(tǒng)，優(yōu)化代理的決策路徑，提高問題解決能力。

5. 自我發(fā)現(xiàn)代理（Self-Discover Agent）：

   • 內(nèi)容簡介： 分析一個能夠?qū)W習(xí)自身能力的代理。
   • 目標： 學(xué)習(xí)如何設(shè)計一個具備自我學(xué)習(xí)和能力提升的智能代理，提高系統(tǒng)的自適應(yīng)性和智能化水平。

評估（Evaluation）

1. 基于代理的評估（Agent-based Evaluation）：

   • 內(nèi)容簡介： 通過模擬用戶交互來評估聊天機器人。
   • 目標： 學(xué)習(xí)如何設(shè)計和實施代理系統(tǒng)的評估方法，確保系統(tǒng)的有效性和用戶滿意度。

2. 在LangSmith中的評估（In LangSmith Evaluation）：

   • 內(nèi)容簡介： 在LangSmith中使用對話數(shù)據(jù)集評估聊天機器人。
   • 目標： 學(xué)習(xí)如何利用LangSmith平臺進行系統(tǒng)化的評估和監(jiān)控，提高系統(tǒng)的性能和穩(wěn)定性。

實驗性（Experimental）

1. 網(wǎng)絡(luò)研究（STORM）：

   • 內(nèi)容簡介： 通過研究和多視角問答生成類似維基百科的文章。
   • 目標： 學(xué)習(xí)如何設(shè)計一個能夠進行深入研究和多角度問答的代理系統(tǒng)，提升信息生成的多樣性和準確性。

2. TNT-LLM：

   • 內(nèi)容簡介： 構(gòu)建豐富、可解釋的用戶意圖分類系統(tǒng)，使用微軟為其Bing Copilot應(yīng)用開發(fā)的分類系統(tǒng)。
   • 目標： 學(xué)習(xí)如何設(shè)計一個能夠精確分類和解釋用戶意圖的代理系統(tǒng)，提高用戶交互的智能化水平。

3. 網(wǎng)頁導(dǎo)航（Web Navigation）：

   • 內(nèi)容簡介： 構(gòu)建一個能夠?qū)Ш胶徒换ゾW(wǎng)站的代理。
   • 目標： 學(xué)習(xí)如何設(shè)計一個具備網(wǎng)頁瀏覽和互動能力的代理系統(tǒng)，提高系統(tǒng)的實用性和用戶體驗。

4. 競賽編程（Competitive Programming）：

   • 內(nèi)容簡介： 構(gòu)建一個具備少量“情景記憶”和人類參與協(xié)作的代理，解決美國計算機奧林匹克競賽的問題；改編自Shi, Tang, Narasimhan和Yao的“語言模型能否解決奧林匹克編程問題？”論文。
   • 目標： 學(xué)習(xí)如何設(shè)計一個具備記憶和協(xié)作能力的高級編程代理，提高系統(tǒng)在復(fù)雜問題解決中的表現(xiàn)。

5. 復(fù)雜數(shù)據(jù)提取（Complex Data Extraction）：

   • 內(nèi)容簡介： 構(gòu)建一個能夠使用函數(shù)調(diào)用執(zhí)行復(fù)雜提取任務(wù)的代理。
   • 目標： 學(xué)習(xí)如何設(shè)計一個能夠處理復(fù)雜數(shù)據(jù)提取需求的代理系統(tǒng)，提升系統(tǒng)的數(shù)據(jù)處理能力和靈活性。

plan_and_execute

當(dāng)然！我很樂意為您詳細講解這個“Plan-and-Execute”示例代碼。這個案例展示了如何使用LangGraph構(gòu)建一個“計劃和執(zhí)行”風(fēng)格的代理（Agent），該代理首先生成一個多步驟計劃，然后逐步執(zhí)行計劃中的每個步驟。在執(zhí)行過程中，如果有必要，可以重新規(guī)劃。

這個示例受到了“Plan-and-Solve”論文和“Baby-AGI”項目的啟發(fā)，其主要優(yōu)勢包括：

• 顯式的長期規(guī)劃：即使是強大的大型語言模型（LLMs）在長遠規(guī)劃上也可能遇到困難，通過顯式的規(guī)劃可以彌補這一不足。
• 模型效率：可以使用較小或較弱的模型來執(zhí)行步驟，只在規(guī)劃階段使用更大或更好的模型。

---

目錄

1. 環(huán)境設(shè)置
2. 設(shè)置API密鑰
3. 定義工具
4. 定義執(zhí)行代理
5. 定義狀態(tài)
6. 規(guī)劃步驟
7. 重新規(guī)劃步驟
8. 創(chuàng)建計算圖
9. 運行代理
10. 總結(jié)

---

1. 環(huán)境設(shè)置

首先，需要安裝所需的Python包：

%pip install --quiet -U langgraph langchain-community langchain-openai tavily-python

• langgraph：用于構(gòu)建可組合的代理系統(tǒng)。
• langchain-community：LangChain的社區(qū)擴展包，包含一些社區(qū)貢獻的工具和功能。
• langchain-openai：用于與OpenAI的API交互。
• tavily-python：Tavily的Python客戶端，用于搜索功能。

---

2. 設(shè)置API密鑰

為了使用OpenAI的語言模型和Tavily的搜索工具，需要設(shè)置相應(yīng)的API密鑰。

import getpass
import osdef _set_env(var: str):if not os.environ.get(var):os.environ[var] = getpass.getpass(f"{var}: ")_set_env("OPENAI_API_KEY")
_set_env("TAVILY_API_KEY")

• getpass.getpass：用于安全地輸入密鑰，防止在終端中顯示。
• os.environ：將API密鑰設(shè)置為環(huán)境變量，供后續(xù)代碼使用。

---

3. 定義工具

在這個示例中，我們使用了一個內(nèi)置的搜索工具 TavilySearchResults。您也可以根據(jù)需要創(chuàng)建自己的工具。

from langchain_community.tools.tavily_search import TavilySearchResultstools = [TavilySearchResults(max_results=3)]

• TavilySearchResults：一個搜索工具，用于獲取搜索結(jié)果。
  • max_results=3：指定返回的最大搜索結(jié)果數(shù)量。

---

4. 定義執(zhí)行代理

現(xiàn)在，我們創(chuàng)建一個執(zhí)行代理（Execution Agent），用于執(zhí)行規(guī)劃好的任務(wù)。在這個示例中，我們?yōu)槊總€任務(wù)使用相同的執(zhí)行代理，但實際上您可以根據(jù)需要為不同的任務(wù)使用不同的代理。

from langchain import hub
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent# 獲取要使用的提示模板 - 您可以根據(jù)需要修改它！
prompt = hub.pull("ih/ih-react-agent-executor")
prompt.pretty_print()# 選擇驅(qū)動代理的LLM
llm = ChatOpenAI(model="gpt-4-turbo-preview")agent_executor = create_react_agent(llm, tools, state_modifier=prompt)

• hub.pull("ih/ih-react-agent-executor")：從LangChain Hub中獲取預(yù)定義的提示模板。
• prompt.pretty_print()：打印提示模板的內(nèi)容，便于查看和修改。
• ChatOpenAI：初始化OpenAI的聊天模型。
  • model="gpt-4-turbo-preview"：指定使用的模型版本。
• create_react_agent：創(chuàng)建一個基于ReAct框架的代理。
  • llm：語言模型，用于驅(qū)動代理的決策和生成。
  • tools：代理可用的工具列表。
  • state_modifier=prompt：使用自定義的提示模板。

---

5. 定義狀態(tài)

接下來，我們定義代理需要跟蹤的狀態(tài)。這包括：

• input：用戶的初始輸入。
• plan：當(dāng)前的計劃，表示為字符串列表。
• past_steps：之前執(zhí)行的步驟，表示為元組列表，每個元組包含步驟和結(jié)果。
• response：最終的響應(yīng)。

import operator
from typing import Annotated, List, Tuple
from typing_extensions import TypedDictclass PlanExecute(TypedDict):input: strplan: List[str]past_steps: Annotated[List[Tuple], operator.add]response: str

• PlanExecute：定義了代理的狀態(tài)結(jié)構(gòu)，繼承自 TypedDict，用于類型提示。
• past_steps：使用 Annotated 和 operator.add，表示在狀態(tài)更新時，新的步驟會被添加到過去的步驟中。

---

6. 規(guī)劃步驟

現(xiàn)在，我們創(chuàng)建規(guī)劃步驟，用于生成一個多步驟的計劃。這里，我們使用了函數(shù)調(diào)用（Function Calling）和 pydantic 的模型來定義結(jié)構(gòu)化的輸出。

from pydantic import BaseModel, Fieldclass Plan(BaseModel):"""Plan to follow in future"""steps: List[str] = Field(description="different steps to follow, should be in sorted order")from langchain_core.prompts import ChatPromptTemplateplanner_prompt = ChatPromptTemplate.from_messages([("system","""For the given objective, come up with a simple step by step plan. \
This plan should involve individual tasks, that if executed correctly will yield the correct answer. Do not add any superfluous steps. \
The result of the final step should be the final answer. Make sure that each step has all the information needed - do not skip steps.""",),("placeholder", "{messages}"),]
)planner = planner_prompt | ChatOpenAI(model="gpt-4o", temperature=0
).with_structured_output(Plan)

• Plan：一個 pydantic 模型，定義了計劃的結(jié)構(gòu)，包含一個字符串列表 steps。
• planner_prompt：定義了規(guī)劃的提示模板，包含系統(tǒng)消息和占位符。
• ChatPromptTemplate.from_messages：創(chuàng)建一個聊天提示模板，基于消息列表。
• planner：將提示模板與LLM組合，并指定輸出的結(jié)構(gòu)為 Plan。
• with_structured_output(Plan)：將llm的輸出結(jié)構(gòu)化為Plan類定義的格式

示例調(diào)用：

planner.invoke({"messages": [("user", "what is the hometown of the current Australia open winner?")]}
)

輸出：

Plan(steps=['Identify the current winner of the Australia Open.', 'Find the hometown of the identified winner.'])

---

7. 重新規(guī)劃步驟

在執(zhí)行過程中，我們可能需要根據(jù)之前的執(zhí)行結(jié)果重新規(guī)劃。為此，我們定義了重新規(guī)劃的步驟。

from typing import Union# 用于定義一個類型可以是多種類型中的任意一種# 定義agent對于用戶的最終響應(yīng)
class Response(BaseModel):"""Response to user."""# 存儲agent最終要返回給用戶的回答response: str# 用于定義agent在重新規(guī)劃步驟中可能采取的行動
class Act(BaseModel):"""Action to perform."""# Union[Response, Plan] 表示 action 字段可以是 Response 或 Plan 類型之一action: Union[Response, Plan] = Field(# 提供字段的描述信息，解釋何時使用 Response，何時使用 Plandescription="Action to perform. If you want to respond to user, use Response. ""If you need to further use tools to get the answer, use Plan.")
# Act 類用于在重新規(guī)劃時決定智能體的下一步行動，既可以是直接回應(yīng)用戶，也可以是生成新的計劃以繼續(xù)執(zhí)行任務(wù)
replanner_prompt = ChatPromptTemplate.from_template("""For the given objective, come up with a simple step by step plan. \
This plan should involve individual tasks, that if executed correctly will yield the correct answer. Do not add any superfluous steps. \
The result of the final step should be the final answer. Make sure that each step has all the information needed - do not skip steps.Your objective was this:
{input}Your original plan was this:
{plan}You have currently done the follow steps:
{past_steps}Update your plan accordingly. If no more steps are needed and you can return to the user, then respond with that. Otherwise, fill out the plan. Only add steps to the plan that still NEED to be done. Do not return previously done steps as part of the plan."""
)replanner = replanner_prompt | ChatOpenAI(model="gpt-4o", temperature=0
).with_structured_output(Act)

• Response：定義了向用戶響應(yīng)的模型，包含一個字符串 response。
• Act：定義了行動模型，包含一個 action，可以是 Response 或 Plan。
• replanner_prompt：定義了重新規(guī)劃的提示模板，包含了當(dāng)前的目標、原始計劃和已執(zhí)行的步驟。
• replanner：將重新規(guī)劃的提示模板與LLM組合，并指定輸出的結(jié)構(gòu)為 Act。

---

8. 創(chuàng)建計算圖

現(xiàn)在，我們創(chuàng)建代理的計算圖，定義了代理的執(zhí)行流程。

from typing import Literal
from langgraph.graph import ENDasync def execute_step(state: PlanExecute):plan = state["plan"]plan_str = "\n".join(f"{i+1}. {step}" for i, step in enumerate(plan))task = plan[0]task_formatted = f"""For the following plan:
{plan_str}\n\nYou are tasked with executing step {1}, {task}."""agent_response = await agent_executor.ainvoke({"messages": [("user", task_formatted)]})return {"past_steps": [(task, agent_response["messages"][-1].content)],}async def plan_step(state: PlanExecute):plan = await planner.ainvoke({"messages": [("user", state["input"])]})return {"plan": plan.steps}async def replan_step(state: PlanExecute):output = await replanner.ainvoke(state)if isinstance(output.action, Response):return {"response": output.action.response}else:return {"plan": output.action.steps}def should_end(state: PlanExecute):if "response" in state and state["response"]:return ENDelse:return "agent"

• execute_step：執(zhí)行計劃中的一個步驟。
  • 從 state 中獲取當(dāng)前的計劃，執(zhí)行第一個步驟，并將結(jié)果添加到 past_steps。
• plan_step：初始規(guī)劃步驟。
  • 使用 planner 根據(jù)用戶輸入生成計劃。
• replan_step：重新規(guī)劃步驟。
  • 使用 replanner 根據(jù)當(dāng)前狀態(tài)（包括已執(zhí)行的步驟）更新計劃或生成最終響應(yīng)。
• should_end：判斷是否結(jié)束執(zhí)行。
  • 如果 response 存在且非空，則返回 END，否則繼續(xù)執(zhí)行。

創(chuàng)建計算圖：

from langgraph.graph import StateGraph, STARTworkflow = StateGraph(PlanExecute)# 添加節(jié)點
workflow.add_node("planner", plan_step)
workflow.add_node("agent", execute_step)
workflow.add_node("replan", replan_step)# 定義邊
workflow.add_edge(START, "planner")
workflow.add_edge("planner", "agent")
workflow.add_edge("agent", "replan")workflow.add_conditional_edges("replan",should_end,["agent", END],
)# 編譯計算圖
app = workflow.compile()

• StateGraph：定義狀態(tài)圖，描述代理的執(zhí)行流程。
• add_node：添加節(jié)點，關(guān)聯(lián)到對應(yīng)的函數(shù)。
• add_edge：添加邊，定義節(jié)點之間的順序。
• add_conditional_edges：添加條件邊，根據(jù)狀態(tài)決定下一個節(jié)點。
• compile：編譯狀態(tài)圖，生成可執(zhí)行的應(yīng)用程序。

---

9. 運行代理

現(xiàn)在，我們可以運行代理，并觀察其執(zhí)行過程。

config = {"recursion_limit": 50}
inputs = {"input": "what is the hometown of the mens 2024 Australia open winner?"}async for event in app.astream(inputs, config=config):for k, v in event.items():if k != "__end__":print(v)

輸出示例：

{'plan': ["Identify the winner of the men's 2024 Australian Open.", 'Research the hometown of the identified winner.']}
{'past_steps': [("Identify the winner of the men's 2024 Australian Open.", "The winner of the men's singles tennis title at the 2024 Australian Open was Jannik Sinner. He defeated Daniil Medvedev in the final with scores of 3-6, 3-6, 6-4, 6-4, 6-3 to win his first major singles title.")]}
{'plan': ['Research the hometown of Jannik Sinner.']}
{'past_steps': [('Research the hometown of Jannik Sinner.', "Jannik Sinner's hometown is Sexten, which is located in northern Italy.")]}
{'response': "The hometown of the men's 2024 Australian Open winner, Jannik Sinner, is Sexten, located in northern Italy."}

• 第一步：生成計劃，包含兩個步驟：
  1. 確定2024年澳大利亞網(wǎng)球公開賽男子冠軍。
  2. 調(diào)查該冠軍的家鄉(xiāng)。
• 第二步：執(zhí)行第一個步驟，獲取冠軍的信息。
• 第三步：重新規(guī)劃，因為第一個步驟已完成，更新計劃，只剩下第二個步驟。
• 第四步：執(zhí)行第二個步驟，獲取冠軍的家鄉(xiāng)信息。
• 第五步：完成所有步驟，生成最終的響應(yīng)。

---

10. 總結(jié)

通過這個示例，我們創(chuàng)建了一個“計劃和執(zhí)行”風(fēng)格的代理，能夠：

• 生成多步驟的計劃：利用強大的語言模型來規(guī)劃任務(wù)。
• 逐步執(zhí)行計劃：按順序執(zhí)行每個步驟，并跟蹤執(zhí)行結(jié)果。
• 動態(tài)重新規(guī)劃：根據(jù)執(zhí)行結(jié)果，動態(tài)更新計劃，確保任務(wù)能夠順利完成。
• 結(jié)構(gòu)化的狀態(tài)管理：利用 TypedDict 和 pydantic 模型，清晰地定義和管理代理的狀態(tài)。

優(yōu)點

• 顯式的長期規(guī)劃：能夠有效處理需要多步驟解決的問題。
• 模型效率：在執(zhí)行階段，可以使用較小的模型，只在規(guī)劃階段使用更強大的模型。

可能的改進

• 并行執(zhí)行：當(dāng)前的設(shè)計是順序執(zhí)行，如果任務(wù)之間沒有依賴關(guān)系，可以考慮并行執(zhí)行以提高效率。
• 更復(fù)雜的規(guī)劃：可以將計劃表示為有向無環(huán)圖（DAG），以處理更復(fù)雜的任務(wù)依賴關(guān)系。

---

附加說明

• LangGraph：一個用于構(gòu)建可組合代理系統(tǒng)的框架，允許您定義代理的執(zhí)行流程和狀態(tài)管理。
• LangChain：一個用于構(gòu)建語言模型應(yīng)用的庫，提供了工具、鏈和代理等組件。

---

匯總

當(dāng)然！以下是將上述LangGraph“計劃并執(zhí)行”智能體入門案例匯總到一個完整的Python腳本中的示例。所有的提示詞（prompts）都已翻譯成中文，以便您更好地理解和學(xué)習(xí)。

請按照以下步驟操作：

1. 安裝必要的包：確保您已經(jīng)安裝了所需的Python包。您可以在命令行中運行以下命令來安裝：

   pip install -U langgraph langchain-community langchain-openai tavily-python pydantic typing_extensions
   

2. 設(shè)置API密鑰：在運行腳本之前，請確保您已經(jīng)擁有OpenAI和Tavily的API密鑰。

3. 運行腳本：將以下代碼保存為plan_and_execute_agent.py，然后在終端中運行：

   python plan_and_execute_agent.py
   

以下是完整的Python腳本內(nèi)容：

# plan_and_execute_agent.pyimport os
import getpass
import operator
import asyncio
from typing import Annotated, List, Tuple, Union
from typing_extensions import TypedDictfrom pydantic import BaseModel, Field
from langchain import hub
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate
from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph, START, END
from dotenv import load_dotenv
# 導(dǎo)入 ChatTongyi 模型
from langchain_community.chat_models.tongyi import ChatTongyi# 加載 .env 文件
load_dotenv()# 讀取環(huán)境變量
api_key = os.getenv("DASHSCOPE_API_KEY")if not api_key:raise ValueError("API Key not found in .env file.")# 初始化 LLM（ChatTongyi）
llm = ChatTongyi(model='qwen-plus')# ----------------------------------
# 定義工具
# ----------------------------------# 使用Tavily的內(nèi)置搜索工具，最多返回3個結(jié)果
tools = [TavilySearchResults(max_results=3)]# ----------------------------------
# 定義執(zhí)行智能體
# ----------------------------------# 獲取執(zhí)行智能體的提示模板
prompt = hub.pull("ih/ih-react-agent-executor")
print("預(yù)定義prompt:\n"+str(prompt))
prompt.pretty_print()# 創(chuàng)建ReAct風(fēng)格的智能體執(zhí)行器
agent_executor = create_react_agent(llm, tools, state_modifier=prompt)# ----------------------------------
# 定義狀態(tài)
# ----------------------------------class PlanExecute(TypedDict):input: str # 用戶輸入的原始請求plan: List[str] # 一個列表，包含生成的計劃步驟past_steps: Annotated[List[Tuple], operator.add]# 一個列表，記錄已執(zhí)行的任務(wù)步驟response: str# 最終的回復(fù)結(jié)果# ----------------------------------
# 規(guī)劃步驟
# ----------------------------------class Plan(BaseModel):"""計劃結(jié)構(gòu)，包含多個步驟"""steps: List[str] = Field(description="需要遵循的不同步驟，應(yīng)按順序排列")# 創(chuàng)建規(guī)劃提示模板
planner_prompt = ChatPromptTemplate.from_messages([("system","""對于給定的目標，制定一個簡單的逐步計劃。\
這個計劃應(yīng)該包含單獨的任務(wù)，如果正確執(zhí)行，將產(chǎn)生正確的答案。不要添加任何多余的步驟。\
最后一步的結(jié)果應(yīng)該是最終答案。確保每個步驟都包含所有必要的信息——不要跳過步驟。""",),("placeholder", "{messages}"),]
)# 創(chuàng)建規(guī)劃器，指定輸出結(jié)構(gòu)為Plan
planner = (planner_prompt |ChatTongyi(model='qwen-plus',temperature=0).with_structured_output(Plan))# ----------------------------------
# 重新規(guī)劃步驟
# ----------------------------------class Response(BaseModel):"""對用戶的最終響應(yīng)"""response: strclass Act(BaseModel):"""要執(zhí)行的行動"""# 可以是Response類型或者是Plan類型action: Union[Response, Plan] = Field(description="要執(zhí)行的行動。如果您想回復(fù)用戶，請使用Response。如果您需要進一步使用工具來獲取答案，請使用Plan。")# 創(chuàng)建重新規(guī)劃提示模板
replanner_prompt = ChatPromptTemplate.from_template("""對于給定的目標，制定一個簡單的逐步計劃。\
這個計劃應(yīng)該包含單獨的任務(wù)，如果正確執(zhí)行，將產(chǎn)生正確的答案。不要添加任何多余的步驟。\
最后一步的結(jié)果應(yīng)該是最終答案。確保每個步驟都包含所有必要的信息——不要跳過步驟。您的目標是：
{input}您原來的計劃是：
{plan}您目前已經(jīng)完成的步驟是：
{past_steps}請相應(yīng)地更新您的計劃。如果不再需要更多步驟，可以返回給用戶。否則，填寫計劃。只添加仍需完成的步驟。不要將已完成的步驟作為計劃的一部分。"""
)# 創(chuàng)建重新規(guī)劃器，指定輸出結(jié)構(gòu)為Act
replanner = (replanner_prompt |ChatTongyi(model='qwen-plus',temperature=0).with_structured_output(Act))# ----------------------------------
# 創(chuàng)建計算圖
# ----------------------------------async def execute_step(state: PlanExecute):"""執(zhí)行當(dāng)前計劃中的第一個任務(wù)，并將結(jié)果記錄在past_steps中。"""print("execute step:\n"+str(state))plan = state["plan"] # 獲取當(dāng)前的執(zhí)行計劃if not plan:return {} # 如果沒有計劃，直接返回空字典plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan)) # 格式化計劃位字符串task = plan[0] # 獲取當(dāng)前計劃的第一個任務(wù)task_formatted = f"""對于以下計劃：
{plan_str}您需要執(zhí)行第1步，{task}.""" # 構(gòu)建用于執(zhí)行的任務(wù)信息# 調(diào)用agent執(zhí)行器，執(zhí)行第一個任務(wù)agent_response = await agent_executor.ainvoke({"messages": [("user", task_formatted)]})# 獲取執(zhí)行結(jié)果result = agent_response["messages"][-1].content# 返回狀態(tài)，記錄執(zhí)行的任務(wù)和結(jié)果return {"past_steps": [(task, result)],}async def plan_step(state: PlanExecute):"""根據(jù)用戶輸入生成初始計劃。"""# 從state中提取用戶的輸入，input是用戶提供的目標或問題# 調(diào)用planner，他是基于ChatTongyi模型的規(guī)劃器，負責(zé)生成任務(wù)計劃plan = await planner.ainvoke({"messages": [("user", state["input"])]})# 返回生成的計劃，結(jié)構(gòu)是一個包含多個步驟plan.staps的字典return {"plan": plan.steps}async def replan_step(state: PlanExecute):"""根據(jù)當(dāng)前狀態(tài)，決定是生成新的計劃還是直接給出響應(yīng)。"""print(f"Replan input state: {state}")  # 調(diào)試打印狀態(tài)output = await replanner.ainvoke(state)# 調(diào)試：打印輸出內(nèi)容print(f"Replanner output: {output}")if output is None:raise ValueError("Replanner returned None, expected an Act with action field.")if isinstance(output.action, Response):return {"response": output.action.response}else:return {"plan": output.action.steps}def should_end(state: PlanExecute):"""檢查是否有最終的響應(yīng)，如果有，則結(jié)束，否則繼續(xù)執(zhí)行。"""# 檢查當(dāng)前狀態(tài)中是否有非空的響應(yīng)（response）# 如果有相應(yīng)，表示任務(wù)已經(jīng)完成，應(yīng)該結(jié)束if "response" in state and state["response"]:return ENDelse:# 任務(wù)還未結(jié)束，返回agent，表示繼續(xù)執(zhí)行return "agent"# 構(gòu)建狀態(tài)圖
workflow = StateGraph(PlanExecute)# 添加節(jié)點
workflow.add_node("planner", plan_step)
workflow.add_node("agent", execute_step)
workflow.add_node("replan", replan_step)# 添加邊
workflow.add_edge(START, "planner")
workflow.add_edge("planner", "agent")
workflow.add_edge("agent", "replan")# 添加條件邊
workflow.add_conditional_edges("replan",should_end,["agent", END],
)# 編譯狀態(tài)圖
app = workflow.compile()# 可選：可視化狀態(tài)圖（需在Jupyter環(huán)境中運行）
# display(Image(app.get_graph(xray=True).draw_mermaid_png()))# ----------------------------------
# 運行智能體
# ----------------------------------async def run_agent():"""運行計劃并執(zhí)行智能體，處理輸入并輸出結(jié)果。"""config = {"recursion_limit": 50}inputs = {"input": "2024年澳大利亞網(wǎng)球公開賽男子冠軍的家鄉(xiāng)是哪里？"}async for event in app.astream(inputs, config=config):for k, v in event.items():if k != "__end__":print(v)# 主函數(shù)
if __name__ == "__main__":asyncio.run(run_agent())

腳本說明

1. 環(huán)境設(shè)置：

   • 使用getpass模塊安全地獲取用戶輸入的API密鑰，并將其設(shè)置為環(huán)境變量OPENAI_API_KEY和TAVILY_API_KEY。

2. 定義工具：

   • 使用Tavily的內(nèi)置搜索工具TavilySearchResults，配置為最多返回3個搜索結(jié)果。

3. 定義執(zhí)行智能體：

   • 從LangChain的Hub中拉取預(yù)定義的提示模板ih/ih-react-agent-executor。
   • 使用OpenAI的gpt-4-turbo-preview模型作為LLM。
   • 創(chuàng)建ReAct風(fēng)格的智能體執(zhí)行器agent_executor。

4. 定義狀態(tài)：

   • 使用TypedDict定義智能體的狀態(tài)結(jié)構(gòu)，包括用戶輸入、計劃、已完成的步驟和最終響應(yīng)。

5. 規(guī)劃步驟：

   • 使用Pydantic的BaseModel定義計劃的結(jié)構(gòu)Plan。
   • 創(chuàng)建規(guī)劃提示模板planner_prompt，并將其與LLM連接，指定輸出結(jié)構(gòu)為Plan。

6. 重新規(guī)劃步驟：

   • 定義Response和Act模型，用于處理智能體的響應(yīng)和行動。
   • 創(chuàng)建重新規(guī)劃提示模板replanner_prompt，并將其與LLM連接，指定輸出結(jié)構(gòu)為Act。

7. 創(chuàng)建計算圖：

   • 定義三個主要的異步函數(shù)：
     • execute_step: 執(zhí)行當(dāng)前計劃中的第一個任務(wù)，并記錄結(jié)果。
     • plan_step: 根據(jù)用戶輸入生成初始計劃。
     • replan_step: 根據(jù)當(dāng)前狀態(tài)決定是否需要重新規(guī)劃或給出最終響應(yīng)。
   • 定義結(jié)束條件函數(shù)should_end，用于判斷智能體是否完成所有任務(wù)。
   • 使用StateGraph構(gòu)建狀態(tài)圖，連接各個節(jié)點并添加條件邊。
   • 編譯狀態(tài)圖生成可運行的應(yīng)用app。

8. 運行智能體：

   • 定義異步函數(shù)run_agent，設(shè)置遞歸限制和輸入問題，然后運行智能體并打印輸出結(jié)果。
   • 在腳本的主入口處調(diào)用run_agent函數(shù)。

示例輸出

運行腳本后，您應(yīng)該會看到類似以下的輸出：

{'plan': ["確定2024年澳大利亞網(wǎng)球公開賽的男子冠軍。", "查找確定冠軍的家鄉(xiāng)。"]}
{'past_steps': [("確定2024年澳大利亞網(wǎng)球公開賽的男子冠軍。", "2024年澳大利亞網(wǎng)球公開賽男子單打冠軍是雅尼克·辛納（Jannik Sinner）。他在決賽中擊敗了達尼爾·梅德韋杰夫（Daniil Medvedev），以3-6、3-6、6-4、6-4、6-3的比分贏得了他的第一個大滿貫單打冠軍。")]}
{'plan': ['查找雅尼克·辛納的家鄉(xiāng)。']}
{'past_steps': [('查找雅尼克·辛納的家鄉(xiāng)。', "雅尼克·辛納的家鄉(xiāng)是意大利北部的塞克斯滕（Sexten）。")]}
{'response': "2024年澳大利亞網(wǎng)球公開賽男子冠軍雅尼克·辛納的家鄉(xiāng)是意大利北部的塞克斯滕。"}

最終，智能體將輸出：

2024年澳大利亞網(wǎng)球公開賽男子冠軍雅尼克·辛納的家鄉(xiāng)是意大利北部的塞克斯滕。

注意事項

• API密鑰：確保您的OpenAI和Tavily API密鑰正確設(shè)置，否則腳本將無法正常運行。
• 異步運行：腳本使用了asyncio庫來處理異步任務(wù)，確保您的Python環(huán)境支持異步編程。
• 可視化：腳本中包含了可視化狀態(tài)圖的代碼，但它在Jupyter環(huán)境中運行效果最佳。如果您在終端中運行，可以忽略相關(guān)代碼。
• 錯誤處理：為了簡化示例，腳本中未包含詳細的錯誤處理機制。在實際應(yīng)用中，建議添加適當(dāng)?shù)腻e誤處理代碼以增強健壯性。

擴展閱讀

• LangGraph文檔：LangGraph GitHub倉庫
• Plan-and-Solve論文：ArXiv鏈接
• Baby-AGI項目：GitHub鏈接

希望這個完整的Python腳本能幫助您更好地理解和學(xué)習(xí)LangGraph的“計劃并執(zhí)行”智能體架構(gòu)！

Reasoning without Observation

LangGraph 實現(xiàn) ReWOO 示例教程講解

在本教程中，我們將探討如何使用 LangGraph 實現(xiàn) ReWOO（Reasoning without Observation） 代理架構(gòu)。ReWOO 架構(gòu)旨在通過優(yōu)化 token 消耗和簡化微調(diào)過程來改進 ReACT 風(fēng)格的代理。下面是對提供的示例的逐步詳細解釋。

---

ReWOO 概述

ReWOO 是由 Xu 等人提出的一種代理架構(gòu)，旨在通過以下方式增強工具的使用：

1. 減少 token 消耗和執(zhí)行時間：在一次傳遞中生成完整的工具使用鏈，避免重復(fù)的 LLM 調(diào)用和冗余的前綴。
2. 簡化微調(diào)過程：規(guī)劃數(shù)據(jù)不依賴于工具的輸出，因此模型可以在不實際調(diào)用工具的情況下進行微調(diào)。

ReWOO 架構(gòu)包含三個主要模塊：

• 🧠 Planner（規(guī)劃器）：生成計劃，格式為一系列推理步驟，指定要使用的工具及其輸入。
• Worker（執(zhí)行者）：根據(jù)提供的參數(shù)執(zhí)行指定的工具。
• 🧠 Solver（求解器）：基于工具的觀察結(jié)果生成初始任務(wù)的答案。

注：標有 🧠 的模塊涉及 LLM 調(diào)用。

---

環(huán)境設(shè)置

首先，我們需要安裝必要的包，并設(shè)置 Tavily（一個搜索引擎工具）和 OpenAI 的 API 密鑰。

%%capture --no-stderr
%pip install -U langgraph langchain_community langchain_openai tavily-pythonimport getpass
import osdef _set_if_undefined(var: str):if not os.environ.get(var):os.environ[var] = getpass.getpass(f"{var}=")_set_if_undefined("TAVILY_API_KEY")
_set_if_undefined("OPENAI_API_KEY")

---

定義圖狀態(tài)

在 LangGraph 中，每個節(jié)點都會更新共享的圖狀態(tài)。狀態(tài)是一個字典，包含了節(jié)點執(zhí)行所需的所有信息。

from typing import List
from typing_extensions import TypedDictclass ReWOO(TypedDict):task: strplan_string: strsteps: Listresults: dictresult: str

---

實現(xiàn) Planner（規(guī)劃器）

規(guī)劃器通過提示 LLM 來生成一個格式化的計劃，包括推理步驟和工具使用，工具的輸入可能包含特殊變量（如 #E1、#E2）用于變量替換。

規(guī)劃器提示：

prompt = """對于以下任務(wù)，請制定能夠逐步解決問題的計劃。對于每個計劃，指明要使用的外部工具以及工具輸入以獲取證據(jù)。你可以將證據(jù)存儲在變量 #E 中，供后續(xù)工具調(diào)用。（Plan, #E1, Plan, #E2, Plan, ...）工具可以是以下之一：
(1) Google[input]：使用 Google 搜索結(jié)果的工具。適用于需要查找特定主題的簡短答案時。輸入應(yīng)為搜索查詢。
(2) LLM[input]：像你自己一樣的預(yù)訓(xùn)練 LLM。當(dāng)你有信心自己解決問題時，可以使用它。輸入可以是任何指令。例如，
任務(wù)：Thomas、Toby 和 Rebecca 一周共工作了 157 小時。Thomas 工作了 x 小時。Toby 工作了比 Thomas 兩倍少 10 小時，Rebecca 工作了比 Toby 少 8 小時。Rebecca 工作了多少小時？
計劃：給定 Thomas 工作了 x 小時，將問題翻譯成代數(shù)表達式并用 Wolfram Alpha 求解。#E1 = WolframAlpha[求解 x + (2x ? 10) + ((2x ? 10) ? 8) = 157]
計劃：找出 Thomas 工作的小時數(shù)。#E2 = LLM[給定 #E1，x 等于多少]
計劃：計算 Rebecca 工作的小時數(shù)。#E3 = Calculator[(2 ? #E2 ? 10) ? 8]開始！
請用豐富的細節(jié)描述你的計劃。每個計劃后面只應(yīng)跟隨一個 #E。任務(wù)：{task}"""

示例任務(wù)：

task = "2024 年澳大利亞網(wǎng)球公開賽男子冠軍的確切家鄉(xiāng)是哪里"

調(diào)用 LLM 生成計劃：

from langchain_openai import ChatOpenAImodel = ChatOpenAI(model="gpt-4o")
result = model.invoke(prompt.format(task=task))
print(result.content)

示例輸出：

計劃：使用 Google 搜索 2024 年澳大利亞網(wǎng)球公開賽男子冠軍。
#E1 = Google[2024 年澳大利亞網(wǎng)球公開賽男子冠軍]計劃：從搜索結(jié)果中獲取冠軍的姓名。
#E2 = LLM[根據(jù) #E1，2024 年澳大利亞網(wǎng)球公開賽男子冠軍是誰？]計劃：使用 Google 搜索該冠軍的確切家鄉(xiāng)。
#E3 = Google[2024 年澳大利亞網(wǎng)球公開賽男子冠軍的家鄉(xiāng)，基于 #E2]計劃：從搜索結(jié)果中獲取冠軍的家鄉(xiāng)。
#E4 = LLM[根據(jù) #E3，2024 年澳大利亞網(wǎng)球公開賽男子冠軍的家鄉(xiāng)是哪里？]

關(guān)鍵點：

• 計劃格式化：規(guī)劃器生成的計劃遵循特定格式，便于后續(xù)解析和執(zhí)行。
• 變量替換：使用 #E1、#E2 等變量，使得后續(xù)工具輸入可以引用前一步的結(jié)果。

定義 Planner 節(jié)點：

import re
from langchain_core.prompts import ChatPromptTemplate# 正則表達式匹配計劃格式
regex_pattern = r"計劃：\s*(.+)\s*(#E\d+)\s*=\s*(\w+)\s*\[([^\]]+)\]"
prompt_template = ChatPromptTemplate.from_messages([("user", prompt)])
planner = prompt_template | modeldef get_plan(state: ReWOO):task = state["task"]result = planner.invoke({"task": task})# 使用正則表達式解析計劃matches = re.findall(regex_pattern, result.content)return {"steps": matches, "plan_string": result.content}

---

實現(xiàn) Worker（執(zhí)行者）

執(zhí)行者根據(jù)規(guī)劃器生成的計劃，逐步執(zhí)行工具，獲取結(jié)果。

設(shè)置搜索引擎工具：

from langchain_community.tools.tavily_search import TavilySearchResultssearch = TavilySearchResults()

工具執(zhí)行函數(shù)：

def _get_current_task(state: ReWOO):if "results" not in state or state["results"] is None:return 1if len(state["results"]) == len(state["steps"]):return Noneelse:return len(state["results"]) + 1def tool_execution(state: ReWOO):"""執(zhí)行者節(jié)點，執(zhí)行計劃中的工具。"""_step = _get_current_task(state)_, step_name, tool, tool_input = state["steps"][_step - 1]_results = state.get("results", {})# 變量替換for k, v in _results.items():tool_input = tool_input.replace(k, v)# 根據(jù)工具類型執(zhí)行if tool == "Google":result = search.invoke(tool_input)elif tool == "LLM":result = model.invoke(tool_input)else:raise ValueError("未知的工具")_results[step_name] = str(result)return {"results": _results}

關(guān)鍵點：

• 順序執(zhí)行：工具按照計劃的順序執(zhí)行，每次執(zhí)行更新狀態(tài)。
• 變量替換：在工具輸入中，替換之前步驟的結(jié)果，確保工具接收到正確的輸入。

---

實現(xiàn) Solver（求解器）

求解器使用規(guī)劃器的計劃和執(zhí)行者的結(jié)果，生成最終的答案。

求解器提示：

solve_prompt = """解決以下任務(wù)或問題。為了解決問題，我們已經(jīng)制定了逐步的計劃，并為每個計劃檢索了相應(yīng)的證據(jù)。請謹慎使用它們，因為長證據(jù)可能包含不相關(guān)的信息。{plan}現(xiàn)在根據(jù)上述提供的證據(jù)解決問題或任務(wù)。直接給出答案，不要添加額外的詞語。任務(wù)：{task}
回答："""

求解器函數(shù)：

def solve(state: ReWOO):plan = ""for _plan, step_name, tool, tool_input in state["steps"]:_results = state.get("results", {})# 變量替換for k, v in _results.items():tool_input = tool_input.replace(k, v)step_name = step_name.replace(k, v)plan += f"計劃：{_plan}\n{step_name} = {tool}[{tool_input}]"prompt = solve_prompt.format(plan=plan, task=state["task"])result = model.invoke(prompt)return {"result": result.content}

關(guān)鍵點：

• 綜合計劃和結(jié)果：求解器將計劃和工具結(jié)果結(jié)合起來，生成最終的提示。
• 直接回答：要求模型直接給出答案，避免冗余信息。

---

定義圖（Graph）

使用 LangGraph 的 StateGraph 來定義代理的工作流程。

路由函數(shù)：

def _route(state):_step = _get_current_task(state)if _step is None:# 所有任務(wù)已執(zhí)行return "solve"else:# 繼續(xù)執(zhí)行工具return "tool"

構(gòu)建圖：

from langgraph.graph import END, StateGraph, STARTgraph = StateGraph(ReWOO)
graph.add_node("plan", get_plan)
graph.add_node("tool", tool_execution)
graph.add_node("solve", solve)
graph.add_edge("plan", "tool")
graph.add_edge("solve", END)
graph.add_conditional_edges("tool", _route)
graph.add_edge(START, "plan")app = graph.compile()

關(guān)鍵點：

• 節(jié)點定義：圖包含三個主要節(jié)點：plan、tool、solve。
• 條件路由：根據(jù)當(dāng)前狀態(tài)，決定下一個節(jié)點是繼續(xù)執(zhí)行工具還是求解。

---

運行圖

執(zhí)行編譯后的圖，傳入任務(wù)，觀察每個步驟的輸出。

執(zhí)行代碼：

for s in app.stream({"task": task}):print(s)print("---")

示例輸出：

{'plan': {'plan_string': "...", 'steps': [...]}}
---
{'tool': {'results': {'#E1': '...'}}}
---
{'tool': {'results': {'#E1': '...', '#E2': '...'}}}
---
{'solve': {'result': 'San Candido, Italy'}}
---

獲取最終結(jié)果：

print(s["solve"]["result"])

輸出：

San Candido, Italy

---

結(jié)論

通過本教程，我們成功地使用 LangGraph 實現(xiàn)了 ReWOO 代理架構(gòu)。該代理：

• 規(guī)劃 了解決任務(wù)的詳細步驟。
• 執(zhí)行 了指定的工具，獲取了必要的信息。
• 求解 了初始任務(wù)，給出了最終答案。

---

限制

盡管此實現(xiàn)展示了 ReWOO 的有效性，但仍存在一些限制：

1. 上下文缺乏時的工具使用：如果缺乏環(huán)境的上下文，規(guī)劃器在使用工具時可能會效率低下。這通?？梢酝ㄟ^少樣本提示或微調(diào)來改善。
2. 順序執(zhí)行時間較長：任務(wù)仍然是順序執(zhí)行的，因此總執(zhí)行時間受所有工具調(diào)用的影響，而不僅僅是最長的那一步。

---

重點總結(jié)：

• ReWOO 架構(gòu)：通過一次性生成完整的計劃，減少了重復(fù)的 LLM 調(diào)用，提高了效率。
• 模塊化設(shè)計：將代理劃分為規(guī)劃器、執(zhí)行者和求解器三個模塊，便于理解和維護。
• LangGraph 應(yīng)用：使用 LangGraph 定義了代理的工作流程，實現(xiàn)了復(fù)雜的任務(wù)執(zhí)行。

---

通過本次詳細講解，希望您對如何使用 LangGraph 實現(xiàn) ReWOO 代理有了深入的理解。

匯總

完整的 ReWOO 示例代碼匯總

以下是將之前講解中的所有代碼整合到一個 rew00_example.py 文件中的完整代碼。請按照以下步驟操作，以便順利執(zhí)行和理解整個流程。

步驟概述

1. 安裝必要的包：確保安裝所有需要的 Python 包。
2. 設(shè)置 API 密鑰：在運行腳本前，準備好 Tavily 和 OpenAI 的 API 密鑰。
3. 運行腳本：執(zhí)行腳本以運行 ReWOO 代理，并獲取最終結(jié)果。

1. 安裝必要的包

在運行腳本之前，確保安裝以下 Python 包?？梢允褂?pip 命令進行安裝：

pip install -U langgraph langchain_community langchain_openai tavily-python

2. 完整的 rew00_example.py 代碼

將以下代碼復(fù)制到一個名為 rew00_example.py 的文件中：

# rew00_example.pyimport getpass
import os
import re
from typing import List
from typing_extensions import TypedDictfrom langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.graph import END, StateGraph, START# 定義 ReWOO 的狀態(tài)字典
class ReWOO(TypedDict):task: strplan_string: strsteps: Listresults: dictresult: str# 設(shè)置環(huán)境變量，如果未定義則提示輸入
def _set_if_undefined(var: str):if not os.environ.get(var):os.environ[var] = getpass.getpass(f"{var}=")# 設(shè)置 API 密鑰
_set_if_undefined("TAVILY_API_KEY")
_set_if_undefined("OPENAI_API_KEY")# 初始化 LLM 模型
model = ChatOpenAI(model="gpt-4o")# 定義規(guī)劃器的提示模板
prompt = """對于以下任務(wù)，請制定能夠逐步解決問題的計劃。對于每個計劃，指明要使用的外部工具以及工具輸入以獲取證據(jù)。你可以將證據(jù)存儲在變量 #E 中，供后續(xù)工具調(diào)用。（Plan, #E1, Plan, #E2, Plan, ...）工具可以是以下之一：
(1) Google[input]：使用 Google 搜索結(jié)果的工具。適用于需要查找特定主題的簡短答案時。輸入應(yīng)為搜索查詢。
(2) LLM[input]：像你自己一樣的預(yù)訓(xùn)練 LLM。當(dāng)你有信心自己解決問題時，可以使用它。輸入可以是任何指令。例如，
任務(wù)：Thomas、Toby 和 Rebecca 一周共工作了 157 小時。Thomas 工作了 x 小時。Toby 工作了比 Thomas 兩倍少 10 小時，Rebecca 工作了比 Toby 少 8 小時。Rebecca 工作了多少小時？
計劃：給定 Thomas 工作了 x 小時，將問題翻譯成代數(shù)表達式并用 Wolfram Alpha 求解。#E1 = WolframAlpha[求解 x + (2x ? 10) + ((2x ? 10) ? 8) = 157]
計劃：找出 Thomas 工作的小時數(shù)。#E2 = LLM[給定 #E1，x 等于多少]
計劃：計算 Rebecca 工作的小時數(shù)。#E3 = Calculator[(2 ? #E2 ? 10) ? 8]開始！
請用豐富的細節(jié)描述你的計劃。每個計劃后面只應(yīng)跟隨一個 #E。任務(wù)：{task}"""# 創(chuàng)建規(guī)劃器模板
prompt_template = ChatPromptTemplate.from_messages([("user", prompt)])
planner = prompt_template | model# 定義獲取計劃的函數(shù)
def get_plan(state: ReWOO):task = state["task"]result = planner.invoke({"task": task})# 正則表達式匹配計劃格式regex_pattern = r"計劃：\s*(.+)\s*(#E\d+)\s*=\s*(\w+)\s*\[([^\]]+)\]"matches = re.findall(regex_pattern, result.content)return {"steps": matches, "plan_string": result.content}# 初始化搜索工具
search = TavilySearchResults()# 定義獲取當(dāng)前任務(wù)的函數(shù)
def _get_current_task(state: ReWOO):if "results" not in state or state["results"] is None:return 1if len(state["results"]) == len(state["steps"]):return Noneelse:return len(state["results"]) + 1# 定義工具執(zhí)行函數(shù)
def tool_execution(state: ReWOO):"""執(zhí)行者節(jié)點，執(zhí)行計劃中的工具。"""_step = _get_current_task(state)if _step is None:return {}_, step_name, tool, tool_input = state["steps"][_step - 1]_results = state.get("results", {})# 變量替換for k, v in _results.items():tool_input = tool_input.replace(k, v)# 根據(jù)工具類型執(zhí)行if tool == "Google":result = search.invoke(tool_input)elif tool == "LLM":result = model.invoke(tool_input)else:raise ValueError("未知的工具")_results[step_name] = str(result)return {"results": _results}# 定義求解器提示
solve_prompt = """解決以下任務(wù)或問題。為了解決問題，我們已經(jīng)制定了逐步的計劃，并為每個計劃檢索了相應(yīng)的證據(jù)。請謹慎使用它們，因為長證據(jù)可能包含不相關(guān)的信息。{plan}現(xiàn)在根據(jù)上述提供的證據(jù)解決問題或任務(wù)。直接給出答案，不要添加額外的詞語。任務(wù)：{task}
回答："""# 定義求解器函數(shù)
def solve(state: ReWOO):plan = ""for _plan, step_name, tool, tool_input in state["steps"]:_results = state.get("results", {})# 變量替換for k, v in _results.items():tool_input = tool_input.replace(k, v)step_name = step_name.replace(k, v)plan += f"計劃：{_plan}\n{step_name} = {tool}[{tool_input}]\n"prompt = solve_prompt.format(plan=plan, task=state["task"])result = model.invoke(prompt)return {"result": result.content}# 定義路由函數(shù)
def _route(state):_step = _get_current_task(state)if _step is None:# 所有任務(wù)已執(zhí)行return "solve"else:# 繼續(xù)執(zhí)行工具return "tool"# 構(gòu)建圖
graph = StateGraph(ReWOO)
graph.add_node("plan", get_plan)
graph.add_node("tool", tool_execution)
graph.add_node("solve", solve)
graph.add_edge("plan", "tool")
graph.add_edge("solve", END)
graph.add_conditional_edges("tool", _route)
graph.add_edge(START, "plan")# 編譯圖
app = graph.compile()# 定義任務(wù)
task = "2024 年澳大利亞網(wǎng)球公開賽男子冠軍的確切家鄉(xiāng)是哪里"# 運行圖并打印輸出
if __name__ == "__main__":for s in app.stream({"task": task}):print(s)print("---")# 獲取并打印最終結(jié)果if "solve" in s and "result" in s["solve"]:print("最終結(jié)果：")print(s["solve"]["result"])

3. 代碼說明

以下是對上述代碼的詳細說明，以幫助您更好地理解各部分的功能：

3.1 導(dǎo)入必要的庫

import getpass
import os
import re
from typing import List
from typing_extensions import TypedDictfrom langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.graph import END, StateGraph, START

• getpass 和 os：用于安全地獲取和設(shè)置環(huán)境變量（API 密鑰）。
• re：用于正則表達式匹配，解析規(guī)劃器生成的計劃。
• TypedDict：定義狀態(tài)字典的類型。
• LangChain 和 LangGraph：用于與 OpenAI 的 LLM 交互和構(gòu)建狀態(tài)圖。

3.2 定義狀態(tài)字典

class ReWOO(TypedDict):task: strplan_string: strsteps: Listresults: dictresult: str

• ReWOO：定義了整個流程中共享的狀態(tài)，包括任務(wù)、計劃、步驟、結(jié)果和最終答案。

3.3 設(shè)置環(huán)境變量

def _set_if_undefined(var: str):if not os.environ.get(var):os.environ[var] = getpass.getpass(f"{var}=")_set_if_undefined("TAVILY_API_KEY")
_set_if_undefined("OPENAI_API_KEY")

• _set_if_undefined：檢查環(huán)境變量是否已設(shè)置，如果未設(shè)置，則提示用戶輸入。

3.4 初始化 LLM 模型

model = ChatOpenAI(model="gpt-4o")

• 使用 OpenAI 的 GPT-4 模型進行自然語言處理。

3.5 定義規(guī)劃器的提示模板

prompt = """..."""
prompt_template = ChatPromptTemplate.from_messages([("user", prompt)])
planner = prompt_template | model

• prompt：定義了如何指導(dǎo) LLM 生成解決任務(wù)的計劃。
• prompt_template：將提示模板與 LLM 模型關(guān)聯(lián)。

3.6 定義獲取計劃的函數(shù)

def get_plan(state: ReWOO):task = state["task"]result = planner.invoke({"task": task})regex_pattern = r"計劃：\s*(.+)\s*(#E\d+)\s*=\s*(\w+)\s*\[([^\]]+)\]"matches = re.findall(regex_pattern, result.content)return {"steps": matches, "plan_string": result.content}

• get_plan：調(diào)用規(guī)劃器生成計劃，并使用正則表達式解析計劃中的每一步。

3.7 初始化搜索工具

search = TavilySearchResults()

• TavilySearchResults：用于執(zhí)行 Google 搜索的工具。

3.8 定義獲取當(dāng)前任務(wù)的函數(shù)

def _get_current_task(state: ReWOO):if "results" not in state or state["results"] is None:return 1if len(state["results"]) == len(state["steps"]):return Noneelse:return len(state["results"]) + 1

• _get_current_task：確定當(dāng)前需要執(zhí)行的任務(wù)步驟。

3.9 定義工具執(zhí)行函數(shù)

def tool_execution(state: ReWOO):_step = _get_current_task(state)if _step is None:return {}_, step_name, tool, tool_input = state["steps"][_step - 1]_results = state.get("results", {})for k, v in _results.items():tool_input = tool_input.replace(k, v)if tool == "Google":result = search.invoke(tool_input)elif tool == "LLM":result = model.invoke(tool_input)else:raise ValueError("未知的工具")_results[step_name] = str(result)return {"results": _results}

• tool_execution：根據(jù)規(guī)劃器生成的步驟，執(zhí)行相應(yīng)的工具（Google 搜索或 LLM 調(diào)用），并更新結(jié)果。

3.10 定義求解器提示和函數(shù)

solve_prompt = """..."""def solve(state: ReWOO):plan = ""for _plan, step_name, tool, tool_input in state["steps"]:_results = state.get("results", {})for k, v in _results.items():tool_input = tool_input.replace(k, v)step_name = step_name.replace(k, v)plan += f"計劃：{_plan}\n{step_name} = {tool}[{tool_input}]\n"prompt = solve_prompt.format(plan=plan, task=state["task"])result = model.invoke(prompt)return {"result": result.content}

• solve_prompt：定義了如何指導(dǎo) LLM 使用計劃和工具結(jié)果生成最終答案。
• solve：整合計劃和結(jié)果，調(diào)用 LLM 生成最終答案。

3.11 定義路由函數(shù)和構(gòu)建圖

def _route(state):_step = _get_current_task(state)if _step is None:return "solve"else:return "tool"graph = StateGraph(ReWOO)
graph.add_node("plan", get_plan)
graph.add_node("tool", tool_execution)
graph.add_node("solve", solve)
graph.add_edge("plan", "tool")
graph.add_edge("solve", END)
graph.add_conditional_edges("tool", _route)
graph.add_edge(START, "plan")app = graph.compile()

• _route：決定下一個執(zhí)行的節(jié)點是繼續(xù)執(zhí)行工具還是轉(zhuǎn)向求解。
• StateGraph：定義了整個流程的節(jié)點和邊，包括條件路由。

3.12 定義任務(wù)并運行圖

task = "2024 年澳大利亞網(wǎng)球公開賽男子冠軍的確切家鄉(xiāng)是哪里"if __name__ == "__main__":for s in app.stream({"task": task}):print(s)print("---")# 獲取并打印最終結(jié)果if "solve" in s and "result" in s["solve"]:print("最終結(jié)果：")print(s["solve"]["result"])

• task：定義需要解決的具體任務(wù)。
• app.stream：運行整個圖，并逐步輸出每個節(jié)點的結(jié)果。
• 最終結(jié)果：在所有工具執(zhí)行完畢后，打印最終的答案。

4. 運行腳本

確保已經(jīng)設(shè)置了環(huán)境變量 TAVILY_API_KEY 和 OPENAI_API_KEY。如果尚未設(shè)置，運行腳本時會提示輸入。

在終端中運行以下命令：

python rew00_example.py

5. 示例輸出

運行腳本后，您將看到類似以下的輸出：

{'plan': {'plan_string': "計劃：使用 Google 搜索 2024 年澳大利亞網(wǎng)球公開賽男子冠軍。\n#E1 = Google[2024 年澳大利亞網(wǎng)球公開賽男子冠軍]\n計劃：從搜索結(jié)果中獲取冠軍的姓名。\n#E2 = LLM[根據(jù) #E1，2024 年澳大利亞網(wǎng)球公開賽男子冠軍是誰？]\n計劃：使用 Google 搜索該冠軍的確切家鄉(xiāng)。\n#E3 = Google[2024 年澳大利亞網(wǎng)球公開賽男子冠軍的家鄉(xiāng)，基于 #E2]\n計劃：從搜索結(jié)果中獲取冠軍的家鄉(xiāng)。\n#E4 = LLM[根據(jù) #E3，2024 年澳大利亞網(wǎng)球公開賽男子冠軍的家鄉(xiāng)是哪里？]"}}
---
{'tool': {'results': {'#E1': '...'}}}
---
{'tool': {'results': {'#E1': '...', '#E2': '...'}}}
---
{'solve': {'result': 'San Candido, Italy'}}
---
最終結(jié)果：
San Candido, Italy

6. 重要提示

• API 密鑰：確保您擁有有效的 Tavily 和 OpenAI API 密鑰，并正確設(shè)置環(huán)境變量。
• 網(wǎng)絡(luò)連接：腳本需要訪問外部 API，確保您的網(wǎng)絡(luò)連接正常。
• 錯誤處理：如果在執(zhí)行過程中遇到錯誤，請檢查 API 密鑰是否正確，或者相應(yīng)的工具是否配置正確。

7. 總結(jié)

通過以上步驟，您可以成功運行 ReWOO 代理架構(gòu)，并理解其如何通過規(guī)劃、工具執(zhí)行和求解生成最終答案。該腳本展示了如何使用 LangGraph 構(gòu)建復(fù)雜的 LLM 驅(qū)動的工作流，并有效地結(jié)合多種工具來解決實際問題。

如有任何疑問或需要進一步的幫助，請隨時提問！

LLMCompiler

下面我將為您詳細講解LangGraph官方入門案例——LLMCompiler。我們將逐步解釋其中的各個組件和代碼示例。

LLMCompiler簡介

LLMCompiler是一種代理（agent）架構(gòu)，旨在通過在有向無環(huán)圖（DAG）中提前執(zhí)行任務(wù)來加速代理任務(wù)的執(zhí)行。同時，它通過減少對大型語言模型（LLM）的調(diào)用次數(shù)，節(jié)省了冗余的token使用成本。其計算圖的概覽如下：

LLMCompiler計算圖

它主要包含三個組件：

1. 計劃器（Planner）：生成任務(wù)的DAG。
2. 任務(wù)獲取單元（Task Fetching Unit）：調(diào)度并盡快執(zhí)行可執(zhí)行的任務(wù)。
3. 合并器（Joiner）：向用戶響應(yīng)或觸發(fā)第二次計劃。

該案例將逐步演示如何使用LangGraph實現(xiàn)LLMCompiler。最終，您將看到類似以下的執(zhí)行過程。

環(huán)境設(shè)置

首先，我們需要安裝所需的包并設(shè)置API密鑰。

%pip install -U langchain_openai langsmith langgraph langchain numexprimport getpass
import osdef _get_pass(var: str):if var not in os.environ:os.environ[var] = getpass.getpass(f"{var}: ")_get_pass("OPENAI_API_KEY")

設(shè)置LangSmith用于LangGraph開發(fā)

注冊LangSmith，以便快速發(fā)現(xiàn)問題并提高LangGraph項目的性能。LangSmith允許您使用跟蹤數(shù)據(jù)來調(diào)試、測試和監(jiān)控使用LangGraph構(gòu)建的LLM應(yīng)用程序。

輔助文件

數(shù)學(xué)工具

將以下代碼放入名為math_tools.py的文件中，并確保您可以在筆記本中導(dǎo)入它。

（此處省略具體代碼，請確保您已正確配置math_tools.py文件。）

輸出解析器

（同樣，此處省略具體代碼，請確保您已正確配置輸出解析器。）

定義工具

我們首先為代理定義要使用的工具。在本示例中，我們將使用搜索引擎和計算器的組合。

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_openai import ChatOpenAI
from math_tools import get_math_tool_get_pass("TAVILY_API_KEY")calculate = get_math_tool(ChatOpenAI(model="gpt-4-turbo-preview"))
search = TavilySearchResults(max_results=1,description='tavily_search_results_json(query="the search query") - 一個搜索引擎。',
)tools = [search, calculate]

如果您不想注冊Tavily，可以使用免費的DuckDuckGo替代。

接下來，我們測試一下calculate工具：

calculate.invoke({"problem": "What's the temp of sf + 5?","context": ["The temperature of sf is 32 degrees"],}
)

輸出：

'37'

計劃器（Planner）

計劃器接受輸入問題并生成要執(zhí)行的任務(wù)列表。如果提供了先前的計劃，它會被指示重新計劃，這在第一次批量任務(wù)完成后，代理需要采取更多操作時很有用。

以下代碼構(gòu)建了計劃器的提示模板，并將其與LLM和輸出解析器（在output_parser.py中定義）組合。輸出解析器處理以下格式的任務(wù)列表：

1. tool_1(arg1="arg1", arg2=3.5, ...)
Thought: 接下來我要使用tool_2來找到Y(jié)
2. tool_2(arg1="", arg2="${1}")
3. join()<END_OF_PLAN>

"Thought"行是可選的。${#}占位符是變量，用于將工具（任務(wù)）輸出傳遞給其他工具。

from typing import Sequence
from langchain import hub
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import (BaseMessage,FunctionMessage,HumanMessage,SystemMessage,
)
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAI
from output_parser import LLMCompilerPlanParser, Taskprompt = hub.pull("wfh/llm-compiler")
print(prompt.pretty_print())

此代碼從集線器中拉取了一個預(yù)定義的提示模板，并打印出來。

接下來，我們定義創(chuàng)建計劃器的函數(shù)：

def create_planner(llm: BaseChatModel, tools: Sequence[BaseTool], base_prompt: ChatPromptTemplate
):# （函數(shù)內(nèi)容省略，主要是設(shè)置計劃器的提示和重計劃邏輯）return planner

我們初始化LLM和計劃器：

llm = ChatOpenAI(model="gpt-4-turbo-preview")
planner = create_planner(llm, tools, prompt)

測試計劃器：

example_question = "What's the temperature in SF raised to the 3rd power?"for task in planner.stream([HumanMessage(content=example_question)]):print(task["tool"], task["args"])print("---")

輸出將顯示計劃器生成的任務(wù)列表，包括要使用的工具和參數(shù)。

任務(wù)獲取單元（Task Fetching Unit）

此組件負責(zé)調(diào)度任務(wù)。它接收如下格式的任務(wù)流：

{tool: BaseTool,dependencies: number[],
}

基本思想是，一旦任務(wù)的依賴被滿足，就開始執(zhí)行任務(wù)。這是通過多線程實現(xiàn)的。下面的代碼將任務(wù)獲取單元和執(zhí)行器結(jié)合起來。

首先，我們定義一些輔助函數(shù)和類，用于解析和執(zhí)行任務(wù)。

import re
import time
from concurrent.futures import ThreadPoolExecutor, wait
from typing import Any, Dict, Iterable, List, Unionfrom langchain_core.runnables import chain as as_runnable
from typing_extensions import TypedDict# （省略部分輔助函數(shù)的具體實現(xiàn)）

然后，我們定義調(diào)度任務(wù)的函數(shù)schedule_tasks，它將任務(wù)分組到一個DAG調(diào)度中，并處理任務(wù)的執(zhí)行和依賴關(guān)系。

示例計劃

在此步驟中，我們展示一個示例計劃，說明如何將計劃器和任務(wù)獲取單元結(jié)合使用。

tool_messages = plan_and_schedule.invoke({"messages": [HumanMessage(content=example_question)]}
)["messages"]tool_messages

這將輸出任務(wù)執(zhí)行的結(jié)果，包括每個工具的輸出。

合并器（Joiner）

現(xiàn)在，我們需要處理這些輸出，確定是應(yīng)該給出最終答案，還是需要重新計劃。我們使用另一個LLM調(diào)用來實現(xiàn)這一點，利用函數(shù)調(diào)用來提高解析的可靠性。

首先，我們定義輸出的Pydantic模型：

from langchain_core.messages import AIMessage
from pydantic import BaseModel, Fieldclass FinalResponse(BaseModel):response: strclass Replan(BaseModel):feedback: str = Field(description="對先前嘗試的分析和對需要改進的建議。")class JoinOutputs(BaseModel):thought: str = Field(description="選擇的行動的推理過程。")action: Union[FinalResponse, Replan]

然后，我們定義合并器的提示模板和可運行對象。

joiner_prompt = hub.pull("wfh/llm-compiler-joiner").partial(examples=""
)
llm = ChatOpenAI(model="gpt-4-turbo-preview")runnable = joiner_prompt | llm.with_structured_output(JoinOutputs)

接著，我們定義解析合并器輸出的函數(shù)和選擇最近消息的函數(shù)。

def _parse_joiner_output(decision: JoinOutputs) -> List[BaseMessage]:# （函數(shù)內(nèi)容省略）return responsedef select_recent_messages(state) -> dict:# （函數(shù)內(nèi)容省略）return {"messages": selected[::-1]}joiner = select_recent_messages | runnable | _parse_joiner_output

測試合并器：

input_messages = [HumanMessage(content=example_question)] + tool_messagesjoiner.invoke({"messages": input_messages})

這將輸出合并器的決策，指示是給出最終答案還是需要重新計劃。

使用LangGraph組合

我們將代理定義為一個有狀態(tài)的圖，其中主要節(jié)點是：

1. 計劃和執(zhí)行（來自第一步的DAG）
2. 合并：確定是否完成或需要重新計劃
3. 重新上下文化：根據(jù)合并器的輸出更新圖狀態(tài)

首先，我們定義狀態(tài)圖并添加節(jié)點和邊。

from langgraph.graph import END, StateGraph, START
from langgraph.graph.message import add_messages
from typing import Annotatedclass State(TypedDict):messages: Annotated[list, add_messages]graph_builder = StateGraph(State)# 添加節(jié)點
graph_builder.add_node("plan_and_schedule", plan_and_schedule)
graph_builder.add_node("join", joiner)# 定義邊
graph_builder.add_edge("plan_and_schedule", "join")def should_continue(state):messages = state["messages"]if isinstance(messages[-1], AIMessage):return ENDreturn "plan_and_schedule"graph_builder.add_conditional_edges("join",should_continue,
)
graph_builder.add_edge(START, "plan_and_schedule")
chain = graph_builder.compile()

示例運行

簡單問題

for step in chain.stream({"messages": [HumanMessage(content="What's the GDP of New York?")]}
):print(step)print("---")

這將輸出代理在回答簡單問題時的執(zhí)行步驟。

多跳問題

steps = chain.stream({"messages": [HumanMessage(content="What's the oldest parrot alive, and how much longer is that than the average?")]},{"recursion_limit": 100,},
)
for step in steps:print(step)print("---")

這展示了代理在處理需要多次搜索的問題時的執(zhí)行過程。

多步數(shù)學(xué)計算

for step in chain.stream({"messages": [HumanMessage(content="What's ((3*(4+5)/0.5)+3245) + 8? What's 32/4.23? What's the sum of those two values?")]}
):print(step)

代理將執(zhí)行多個數(shù)學(xué)計算，并合并結(jié)果。

復(fù)雜的重新計劃示例

for step in chain.stream({"messages": [HumanMessage(content="Find the current temperature in Tokyo, then, respond with a flashcard summarizing this information")]}
):print(step)

此示例展示了代理在需要重新計劃時的行為。

結(jié)論

恭喜您構(gòu)建了第一個LLMCompiler代理！以下是上述實現(xiàn)的一些已知限制：

• 計劃器輸出解析的格式脆弱：如果您的函數(shù)需要超過1或2個參數(shù)，解析可能會失敗。可以通過使用流式工具調(diào)用來提高可靠性。
• 變量替換脆弱：在上述示例中，變量替換可能不夠健壯?？梢允褂梦⒄{(diào)模型和更健壯的語法（例如使用Lark或工具調(diào)用模式）來改進。
• 狀態(tài)可能變得很長：如果您需要多次重新計劃，狀態(tài)可能會變得很長。為了解決這個問題，您可以在超過一定的token限制后添加消息壓縮器。

希望這個詳細的講解對您理解LLMCompiler在LangGraph中的實現(xiàn)有所幫助。

匯總

好的，下面我將幫助您將上述LangGraph官方入門案例的代碼匯總到一個可運行的Python文件中。為了便于理解和運行，我們將分為幾個部分：

1. 安裝依賴包并設(shè)置API密鑰
2. 創(chuàng)建輔助文件
   • math_tools.py
   • output_parser.py
3. 編寫主腳本
   • llmcompiler_demo.py

請按照以下步驟操作：

---

1. 安裝依賴包并設(shè)置API密鑰

首先，確保您已經(jīng)安裝了所需的Python包，并設(shè)置了必要的API密鑰。

安裝依賴包

在您的終端或命令提示符中運行以下命令來安裝所需的包：

pip install -U langchain_openai langsmith langgraph langchain numexpr pydantic

設(shè)置API密鑰

您需要設(shè)置以下環(huán)境變量：

• OPENAI_API_KEY：用于OpenAI的API密鑰。
• TAVILY_API_KEY：用于Tavily搜索引擎的API密鑰。

您可以在運行腳本前在終端中設(shè)置這些環(huán)境變量，例如：

export OPENAI_API_KEY='your_openai_api_key'
export TAVILY_API_KEY='your_tavily_api_key'

或者在Windows上：

set OPENAI_API_KEY=your_openai_api_key
set TAVILY_API_KEY=your_tavily_api_key

---

2. 創(chuàng)建輔助文件

a. math_tools.py

創(chuàng)建一個名為 math_tools.py 的文件，并將以下代碼粘貼進去：

# math_tools.pyfrom langchain_core.tools import BaseTool
from typing import Optional, List
import mathdef get_math_tool(llm):class MathTool(BaseTool):name = "math"description = ("math(problem: str, context: Optional[list[str]]) -> float:\n"" - 解決提供的數(shù)學(xué)問題。\n"" - `problem` 可以是簡單的數(shù)學(xué)問題（例如 \"1 + 3\"）或文字題（例如 \"如果有3個蘋果和2個蘋果，總共有多少個蘋果\"）。\n"" - 每次只能計算一個表達式。\n"" - 盡量減少 `math` 操作的次數(shù)。\n"" - 可以選擇性地提供一個字符串列表作為 `context` 來幫助解決問題。\n"" - 如果需要對之前的任務(wù)輸出進行數(shù)學(xué)計算，必須將其作為 `context` 提供。\n")args_schema = {"problem": str,"context": Optional[List[str]]}def invoke(self, args: dict, config: Optional[dict] = None) -> str:problem = args.get("problem")context = args.get("context", [])try:# 這里可以集成更復(fù)雜的數(shù)學(xué)解析器，如SymPyresult = eval(problem, {"__builtins__": None}, {})return str(result)except Exception as e:return f"ERROR(Failed to evaluate the expression. Error: {str(e)})"return MathTool()

b. output_parser.py

創(chuàng)建一個名為 output_parser.py 的文件，并將以下代碼粘貼進去：

# output_parser.pyfrom langchain_core.tools import BaseTool
from langchain_core.messages import FunctionMessage
from typing import List, Dict, Any
from pydantic import BaseModel
from typing import Unionclass Task(BaseModel):idx: inttool: BaseToolargs: Dict[str, Any]dependencies: List[int]class LLMCompilerPlanParser(BaseModel):tools: List[BaseTool]def parse(self, plan_str: str) -> List[Task]:tasks = []lines = plan_str.strip().split('\n')for line in lines:if line.startswith("join"):task = Task(idx=len(tasks)+1, tool="join", args={}, dependencies=[])tasks.append(task)continueif not line.strip():continueparts = line.split(". ", 1)if len(parts) != 2:continueidx_str, rest = partsidx = int(idx_str)if "(" in rest and rest.endswith(")"):tool_name, args_str = rest.split("(", 1)args_str = args_str[:-1]  # Remove the closing parenthesisargs = eval(args_str)  # 注意：在實際應(yīng)用中請避免使用evaldependencies = []  # 這里可以根據(jù)args中的變量來解析依賴關(guān)系task = Task(idx=idx, tool=tool_name, args=args, dependencies=dependencies)tasks.append(task)return tasks

注意：在實際應(yīng)用中，使用 eval 解析字符串可能會帶來安全風(fēng)險。請確保輸入內(nèi)容的安全性，或使用更安全的解析方法。

---

3. 編寫主腳本

創(chuàng)建一個名為 llmcompiler_demo.py 的文件，并將以下代碼粘貼進去：

# llmcompiler_demo.pyimport os
import getpass
import itertools
import re
import time
from concurrent.futures import ThreadPoolExecutor, wait
from typing import Any, Dict, Iterable, List, Union, Sequence
from typing_extensions import TypedDictfrom langchain import hub
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import (BaseMessage,FunctionMessage,HumanMessage,SystemMessage,AIMessage,
)
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAIfrom langgraph.graph import END, StateGraph, START
from langgraph.graph.message import add_messagesfrom output_parser import LLMCompilerPlanParser, Task
from math_tools import get_math_toolfrom pydantic import BaseModel, Field# 設(shè)置API密鑰
def _get_pass(var: str):if var not in os.environ:os.environ[var] = getpass.getpass(f"{var}: ")_get_pass("OPENAI_API_KEY")
_get_pass("TAVILY_API_KEY")# 設(shè)置LangSmith（可選）
# 您可以根據(jù)需要設(shè)置LangSmith，本文檔中不涉及具體配置# 定義工具
from langchain_community.tools.tavily_search import TavilySearchResultscalculate = get_math_tool(ChatOpenAI(model="gpt-4-turbo-preview"))
search = TavilySearchResults(max_results=1,description='tavily_search_results_json(query="the search query") - 一個搜索引擎。',
)tools = [search, calculate]# 計劃器（Planner）
prompt = hub.pull("wfh/llm-compiler")
print("Planner Prompt:")
print(prompt.pretty_print())def create_planner(llm: BaseChatModel, tools: Sequence[BaseTool], base_prompt: ChatPromptTemplate
):tool_descriptions = "\n".join(f"{i+1}. {tool.description}\n"for i, tool in enumerate(tools))planner_prompt = base_prompt.partial(replan="",num_tools=len(tools)+ 1,  # 添加 join() 工具tool_descriptions=tool_descriptions,)replanner_prompt = base_prompt.partial(replan=' - 您將獲得“Previous Plan”，即上一個代理創(chuàng)建的計劃以及每個計劃的執(zhí)行結(jié)果（作為 Observation），以及關(guān)于已執(zhí)行結(jié)果的一般思考（作為 Thought）。''您必須使用這些信息來創(chuàng)建“Current Plan”。\n'' - 在開始 Current Plan 時，您應(yīng)該以概述下一步計劃策略的“Thought”開頭。\n'' - 在 Current Plan 中，您絕不應(yīng)重復(fù) Previous Plan 中已執(zhí)行的操作。\n'' - 您必須從上一個任務(wù)的結(jié)束索引繼續(xù)任務(wù)索引。不要重復(fù)任務(wù)索引。',num_tools=len(tools) + 1,tool_descriptions=tool_descriptions,)def should_replan(state: list):return isinstance(state[-1], SystemMessage)def wrap_messages(state: list):return {"messages": state}def wrap_and_get_last_index(state: list):next_task = 0for message in state[::-1]:if isinstance(message, FunctionMessage):next_task = message.additional_kwargs["idx"] + 1breakstate[-1].content = state[-1].content + f" - Begin counting at : {next_task}"return {"messages": state}return (RunnableBranch((should_replan, wrap_and_get_last_index | replanner_prompt),wrap_messages | planner_prompt,)| llm| LLMCompilerPlanParser(tools=tools))llm = ChatOpenAI(model="gpt-4-turbo-preview")
planner = create_planner(llm, tools, prompt)example_question = "What's the temperature in SF raised to the 3rd power?"print("\nPlanning Tasks:")
for task in planner.stream([HumanMessage(content=example_question)]):print(task["tool"], task["args"])print("---")# 任務(wù)獲取單元（Task Fetching Unit）
def _get_observations(messages: List[BaseMessage]) -> Dict[int, Any]:results = {}for message in messages[::-1]:if isinstance(message, FunctionMessage):results[int(message.additional_kwargs["idx"])] = message.contentreturn resultsclass SchedulerInput(TypedDict):messages: List[BaseMessage]tasks: Iterable[Task]def _execute_task(task, observations, config):tool_to_use = task["tool"]if isinstance(tool_to_use, str):return tool_to_useargs = task["args"]try:if isinstance(args, str):resolved_args = _resolve_arg(args, observations)elif isinstance(args, dict):resolved_args = {key: _resolve_arg(val, observations) for key, val in args.items()}else:resolved_args = argsexcept Exception as e:return (f"ERROR(Failed to call {tool_to_use.name} with args {args}.)"f" Args could not be resolved. Error: {repr(e)}")try:return tool_to_use.invoke(resolved_args, config)except Exception as e:return (f"ERROR(Failed to call {tool_to_use.name} with args {args}."+ f" Args resolved to {resolved_args}. Error: {repr(e)})")def _resolve_arg(arg: Union[str, Any], observations: Dict[int, Any]):ID_PATTERN = r"\$\{?(\d+)\}?"def replace_match(match):idx = int(match.group(1))return str(observations.get(idx, match.group(0)))if isinstance(arg, str):return re.sub(ID_PATTERN, replace_match, arg)elif isinstance(arg, list):return [_resolve_arg(a, observations) for a in arg]else:return str(arg)def schedule_task(task_inputs, config):task: Task = task_inputs["task"]observations: Dict[int, Any] = task_inputs["observations"]try:observation = _execute_task(task, observations, config)except Exception:import tracebackobservation = traceback.format_exc()observations[task["idx"]] = observationdef schedule_pending_task(task: Task, observations: Dict[int, Any], retry_after: float = 0.2
):while True:deps = task["dependencies"]if deps and (any([dep not in observations for dep in deps])):time.sleep(retry_after)continueschedule_task({"task": task, "observations": observations}, config=None)breakdef schedule_tasks(scheduler_input: SchedulerInput) -> List[FunctionMessage]:tasks = scheduler_input["tasks"]args_for_tasks = {}messages = scheduler_input["messages"]observations = _get_observations(messages)task_names = {}originals = set(observations)futures = []retry_after = 0.25with ThreadPoolExecutor() as executor:for task in tasks:deps = task["dependencies"]task_names[task["idx"]] = (task["tool"] if isinstance(task["tool"], str) else task["tool"].name)args_for_tasks[task["idx"]] = task["args"]if (deps and (any([dep not in observations for dep in deps]))):futures.append(executor.submit(schedule_pending_task, task, observations, retry_after))else:schedule_task({"task": task, "observations": observations}, config=None)wait(futures)new_observations = {k: (task_names[k], args_for_tasks[k], observations[k])for k in sorted(set(observations.keys()) - originals)}tool_messages = [FunctionMessage(name=name,content=str(obs),additional_kwargs={"idx": k, "args": task_args},tool_call_id=k,)for k, (name, task_args, obs) in new_observations.items()]return tool_messagesdef plan_and_schedule(state):messages = state["messages"]tasks = planner.stream(messages)try:tasks = itertools.chain([next(tasks)], tasks)except StopIteration:tasks = iter([])scheduled_tasks = schedule_tasks({"messages": messages,"tasks": tasks,})return {"messages": scheduled_tasks}# 合并器（Joiner）
class FinalResponse(BaseModel):response: strclass Replan(BaseModel):feedback: str = Field(description="對先前嘗試的分析和對需要改進的建議。")class JoinOutputs(BaseModel):thought: str = Field(description="選擇的行動的推理過程。")action: Union[FinalResponse, Replan]def _parse_joiner_output(decision: JoinOutputs) -> List[BaseMessage]:response = [AIMessage(content=f"Thought: {decision.thought}")]if isinstance(decision.action, Replan):return response + [SystemMessage(content=f"Context from last attempt: {decision.action.feedback}")]else:return response + [AIMessage(content=decision.action.response)]def select_recent_messages(state) -> dict:messages = state["messages"]selected = []for msg in messages[::-1]:selected.append(msg)if isinstance(msg, HumanMessage):breakreturn {"messages": selected[::-1]}joiner_prompt = hub.pull("wfh/llm-compiler-joiner").partial(examples=""
)
joiner_llm = ChatOpenAI(model="gpt-4-turbo-preview")def with_structured_output(llm, output_model):return llmrunnable = joiner_prompt | with_structured_output(joiner_llm, JoinOutputs)
joiner = RunnableBranch(select_recent_messages,runnable,
) | _parse_joiner_output# 使用LangGraph組合
class State(TypedDict):messages: List[BaseMessage]graph_builder = StateGraph(State)graph_builder.add_node("plan_and_schedule", plan_and_schedule)
graph_builder.add_node("join", joiner)graph_builder.add_edge("plan_and_schedule", "join")def should_continue(state):messages = state["messages"]if isinstance(messages[-1], AIMessage):return ENDreturn "plan_and_schedule"graph_builder.add_conditional_edges("join",should_continue,
)graph_builder.add_edge(START, "plan_and_schedule")
chain = graph_builder.compile()# 示例運行
def run_chain(question: str):print(f"\n詢問: {question}")for step in chain.stream({"messages": [HumanMessage(content=question)]}):if "join" in step:final_message = step["join"]["messages"][-1].contentprint(f"最終回答: {final_message}")if __name__ == "__main__":# 簡單問題示例run_chain("What's the GDP of New York?")# 多跳問題示例run_chain("What's the oldest parrot alive, and how much longer is that than the average?")# 多步數(shù)學(xué)計算示例run_chain("What's ((3*(4+5)/0.5)+3245) + 8? What's 32/4.23? What's the sum of those two values?")# 復(fù)雜的重新計劃示例run_chain("Find the current temperature in Tokyo, then, respond with a flashcard summarizing this information")

---

4. 運行腳本

確保您已經(jīng)按照上述步驟創(chuàng)建了 math_tools.py 和 output_parser.py 文件，并設(shè)置了必要的API密鑰。然后，在終端中運行主腳本：

python llmcompiler_demo.py

腳本將依次處理以下示例問題：

1. 簡單問題：紐約的GDP是多少？
2. 多跳問題：目前世界上最老的鸚鵡是誰，它比平均壽命多活了多久？
3. 多步數(shù)學(xué)計算：計算 ((3*(4+5)/0.5)+3245) + 8，然后計算 32/4.23，最后求這兩個結(jié)果的和。
4. 復(fù)雜的重新計劃示例：查找東京當(dāng)前的溫度，然后用閃卡形式總結(jié)此信息。

每個問題的處理過程和最終回答將打印在終端中。

---

5. 注意事項

• API 密鑰：確保您的 OPENAI_API_KEY 和 TAVILY_API_KEY 正確設(shè)置，并具有相應(yīng)的訪問權(quán)限。
• 安全性：在 output_parser.py 中使用了 eval 函數(shù)來解析參數(shù)，這在生產(chǎn)環(huán)境中可能存在安全風(fēng)險。建議使用更安全的解析方法，例如正則表達式或?qū)Ｓ玫慕馕鰩臁?/li>
• 依賴版本：確保安裝的包版本與代碼兼容，特別是 pydantic 和 langchain_core。如果遇到版本兼容性問題，請參考相應(yīng)包的文檔進行調(diào)整。

---

通過上述步驟，您應(yīng)該能夠成功運行LLMCompiler的示例，并對其整體流程有更深入的理解。如有任何問題，請隨時提問！