ReAct介紹

要介紹ReAct最好要知道它是從哪來的。

ReAct這個概念出自《REACT : SYNERGIZING REASONING AND ACTING IN LANGUAGE MODELS》這篇很牛的論文。

論文地址：https://arxiv.org/abs/2210.03629

我們先來看下這篇論文的摘要：

翻譯：

盡管大型語言模型（LLMs）在語言理解和交互式決策任務中展現了卓越的性能，但其推理能力（如思維鏈提示）與行為能力（如行動規劃生成）以往主要被作為獨立課題進行研究。

在本文中，我們探索了一種將LLMs用于以交錯方式同時生成推理軌跡與任務特定動作的方法，從而實現兩者之間的更強協同效應：推理軌跡幫助模型推導、跟蹤和更新行動方案，并處理異常情況；而動作則使模型能夠與外部源（如知識庫或環境）進行交互并獲取額外信息。我們將這一方法命名為ReAct，并將其應用于一系列語言理解和決策任務中，結果表明其性能超越了當前最先進的基線方法，同時顯著提升了人類可解釋性與可信度。

具體而言，在問答（HotpotQA）和事實驗證（Fever）任務中，ReAct通過與簡易維基百科API交互，有效克服了思維鏈推理中普遍存在的幻覺和錯誤傳播問題，并生成了比無推理軌跡的基線方法更具可解釋性的人類模仿式問題解決軌跡。此外，在兩個交互式決策基準（ALFWorld 和 WebShop）上，ReAct 僅通過一到兩個上下文示例的提示，便分別以34%和10%的絕對成功率超越了模仿學習與強化學習方法。

從摘要中我們大概可以知道ReAct是什么，可以把它理解為一種讓 AI 一邊思考，一邊行動的強大框架。

ReAct 是 Reason 和 Act 兩個詞的縮寫，即“推理”和“行動”。

這個“思考 → 行動 → 觀察”的循環，會不斷重復，直到模型能夠給出一個可靠、準確的最終答案。

再來看下這篇論文的結論：

翻譯：

我們提出了ReAct，一種簡單而有效的用于在大語言模型中協同推理與行動的方法。通過在多跳問答、事實核查和交互式決策任務上的一系列多樣化實驗，我們表明ReAct能夠帶來更優的性能，并產生可解釋的決策軌跡。盡管我們的方法很簡單，但在動作空間較大的復雜任務中，模型需要更多示例才能良好學習，而這不幸很容易超出上下文學習的輸入長度限制。我們在HotpotQA上探索了微調方法，取得了初步的積極結果，但未來仍需依賴更多高質量的人類標注數據來進一步提升性能。通過多任務訓練擴展ReAct，并將其與強化學習等互補范式結合，有望催生更強大的智能體，進一步釋放大語言模型在更多應用場景中的潛力。

在這篇論文中沒有圖示，不過已經有大佬畫了很多圖示了，這里我貼一個大佬畫的圖，結合圖示大家可以更好的理解ReAct是怎么樣的。

圖片來源：https://mlpills.substack.com/p/diy-14-step-by-step-implementation

實現一個簡單的ReAct AI Agent

我使用的是python中的langgraph這個框架。

LangGraph 是一個用于構建有狀態、多智能體（multi-agent）應用程序的庫，它通過將工作流定義為圖（Graph）的形式，讓你能夠用代碼精確地控制應用的執行流程和邏輯。

更多可以從官方上了解。

我學習的那個文章工具是使用維基百科，但我試了一下感覺效果不太行，很多問題找不到，就自己改成了使用duckduckgo_search。

新建一個python環境，安裝包：

pip install -U langgraph langchain-openai wikipedia ddgs

導入必要的類和函數：

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import BaseMessage, SystemMessage, ToolMessage
from typing import TypedDict, Sequence, Annotated
from langgraph.graph.message import add_messages
import json
from ddgs import DDGS

定義狀態模型：

class AgentState(TypedDict):
    """State of the agent for one turn or conversation."""
    messages: Annotated[Sequence[BaseMessage], add_messages]

LangGraph 使用一個狀態對象來跟蹤對話和任何中間數據。對于一個基本的 ReAct 代理，狀態可以簡單到僅包含一個消息列表（聊天歷史）。我們定義一個名為 TypedDict 的狀態類型，其中只有一個鍵 "messages"，用于保存一系列消息。我們還為該字段附加了一個歸約函數 add_messages —— 這樣做可以確保當節點返回新消息時，這些消息會被追加到狀態的消息列表中（而不是覆蓋它）。

設置大語言模型和工具：

# Initialize the chat model (LLM) - make sure your API key is set
model = ChatOpenAI(
    base_url="https://api.siliconflow.cn/v1",
    api_key="sk-xxx",
    model="Qwen/Qwen3-Next-80B-A3B-Instruct",
    temperature=0)

# Define a Wikipedia search tool
#import wikipedia

# @tool
# def wiki_search(query: str) -> str:
#     """Search Wikipedia for the query and return a brief summary of the top result."""
#     try:
#         # Fetch summary of top search result (we set it to 5 sentences)
#         summary = wikipedia.summary(query, sentences=5)
#         return summary
#     except Exception as e:
#         return f"Error: {e}"

@tool
def duckduckgo_search(query: str) -> str:
    """Search DuckDuckGo for the query and return a brief summary of the top result."""
    try:
        # Fetch summary of top search result (we set it to 5 sentences)
        summary = DDGS().text(query, max_results=5)
        return summary
    except Exception as e:
        return f"Error: {e}"
    
# Map tool name to the tool function for easy lookup
tools = [duckduckgo_search]
tools_by_name = {tool.name: tool for tool in tools}

# Give the model access to the tools
model = model.bind_tools(tools) 

接下來，初始化大語言模型并定義我們的工具。我們使用 LangChain 中的 ChatOpenAI 來創建一個聊天模型實例,可以使用任何可用的模型（只要其兼容工具調用功能）。

定義 LangGraph 節點和條件邏輯：

推理節點（LLM 調用）：該節點將調用大語言模型，以生成最終答案或決定執行某個工具操作。我們將其實現為一個函數 call_model(state)。該函數接收當前狀態（包含迄今為止的對話消息），并返回 LLM 生成的新消息。我們還會加入一個系統提示（例如：“你是一個樂于助人的助手…”）來引導 LLM 的行為。用戶的查詢已包含在狀態的消息中，因此我們將系統提示與所有現有消息一并傳遞給模型進行調用。LangChain 的 ChatOpenAI 可以返回一條包含函數調用的消息——如果模型判斷需要使用工具（底層可能利用 OpenAI 的函數調用功能，請求執行 duckduckgo_search 等操作）。

def call_model(state: AgentState):
    """LLM reasoning node: call the chat model with system prompt + conversation."""
    system_prompt = SystemMessage(content="You are a helpful AI assistant. If needed, you can use the duckduckgo_search tool to build your answer.")
    # Call the chat model with system + existing messages (user question is included in state["messages"])
    response = model.invoke([system_prompt] + list(state["messages"]))
    # Return the response as a list (to be appended to state's messages via reducer)
    return {"messages": [response]}

工具節點（執行工具）：此節點用于執行大語言模型所請求的任何工具。我們實現 tool_node(state) 來檢查大語言模型最新消息中是否存在工具調用。如果存在工具調用，我們將調用相應的工具函數，并將其結果封裝為一個特殊的 ToolMessage。該 ToolMessage 將被添加到狀態中，以便大語言模型在下一次迭代中能夠看到工具的輸出。

def tool_node(state: AgentState):
    """Tool execution node: execute any tool calls the LLM asked for."""
    outputs = []
    # Check the last message from the LLM for tool calls
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        # If the model requested one or more tool calls, execute each
        for tool_call in last_message.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            if tool_name in tools_by_name:
                # Invoke the corresponding tool function with provided arguments
                result = tools_by_name[tool_name].invoke(tool_args)
            else:
                result = f"Tool '{tool_name}' not found."
            # Wrap the result in a ToolMessage for the LLM to read
            outputs.append(
                ToolMessage(
                    content=json.dumps(result, ensure_ascii=False),   # tool result as JSON string, preserve Chinese characters
                    name=tool_name,
                    tool_call_id=tool_call.get("id")  # use id if provided
                )
            )
    # Return the tool outputs to be added to messages
    return {"messages": outputs}

條件邊（should_continue）：在每次LLM推理步驟之后，我們需要決定代理是應該以答案結束，還是通過使用工具繼續。我們定義一個函數 should_continue(state)，用于檢查LLM的最后一條消息。如果LLM沒有請求任何工具（即沒有函數調用），則表示它已經生成了最終答案，此時代理可以結束工作；如果請求了工具，則應繼續前往工具節點。該函數將返回一個標志（例如“continue”或“end”），LangGraph將據此選擇下一個節點。

def should_continue(state: AgentState) -> str:
    """Decide whether to continue the ReAct loop or end it, based on last LLM message."""
    last_message = state["messages"][-1]
    print(last_message)
    # If the LLM's last message did not request a tool, we're done
    if not (hasattr(last_message, "tool_calls") and last_message.tool_calls):
        return "end"
    else:
        # There is a tool request, so continue to the tool node
        return "continue"

構建和編譯圖：

現在，我們使用 LangGraph 的 StateGraph 來構建圖。我們添加了兩個節點：“agent”（用于 LLM 推理）和 “tool”（用于工具執行），設置入口點，并定義節點間的轉換關系。關鍵部分是從 LLM 節點添加一條條件邊，根據 should_continue 函數的輸出，決定走向工具節點或圖的結束節點。我們將“continue”信號映射到“tool”節點，將“end”信號映射到 END（一個特殊標記，表示圖應終止）。同時，我們從工具節點添加一條普通邊返回到 LLM 節點，從而形成一個循環：在使用工具后，智能體將返回 LLM 以整合新獲取的信息。

graph_builder = StateGraph(AgentState)

graph_builder.add_node("call_model", call_model)
graph_builder.add_node("tool_node", tool_node)
graph_builder.set_entry_point("call_model")
graph_builder.add_edge("tool_node", "call_model")
graph_builder.add_conditional_edges("call_model", 
                                    should_continue,
                                    {"continue": "tool_node", "end": END})

graph = graph_builder.compile()

如果你使用的是NoteBook，添加from IPython.display import Image, display、display(Image(graph.get_graph().draw_mermaid_png()))，還可以看到這個圖的結構：

使用這個簡單的ReAct AI Agent：

通過上面的幾個步驟，我們就構建了這個簡單的ReAct AI Agent，現在我們來試試它的效果吧?。?/p>

def main():
    inputs = {
        "messages": [("user", "樺加沙是什么？")],
        "intermediate_steps": [],
    }
    print_stream(graph.stream(inputs, stream_mode="values"))

if __name__ == "__main__":
    main()

效果：

這個簡單的ReAct AI Agent調用了工具來回答這個問題。

以上就是本期的全部內容，期待跟你一起進步，每天學習一點，希望對你有所幫助。