LlamainDex進行研究和寫作的多代理工作流程

大語言模型代理是自動化搜索，內容生成和質量審查等任務的強大工具。但是，單個代理通常無法有效地做所有事情，尤其是當您需要集成外部資源（例如Web搜索）和多個專業步驟（例如，起草與審查）時。多代理工作流程使您可以將這些任務分配給不同的代理，每個代理都有自己的工具，約束和職責。在本文中，我們將研究如何構建一個三個代理系統（研究，寫作和審查），每個代理都處理在Internet上創建簡潔歷史報告的特定部分。我們還將確保系統不會陷入搜索循環，這可能會浪費時間和學分。

學習目標

了解多代理工作流程如何通過LLMS增強任務自動化。 >
• 學習建立一個三個代理系統，以進行研究，寫作和審查任務。
實施保障措施，以防止自動化工作流程中的無限搜索循環。
探索諸如DuckDuckgo之類的外部工具的集成以進行有效的數據檢索。
• 開發一個由LLM驅動的工作流程，可確保結構化和高質量的內容生成。
• >本文是

> > data Science Blogathon的一部分。 內容表 語言模型（LLM） - OpenAI GPT-4

>工作流的基本工具

>定義任務執行的AI代理
• >
• 代理工作流 - 與避免進行任務執行
結論
• 常見問題
語言模型（LLM） - OpenAI GPT-4
>我們將使用Llama-Index的OpenAI（Model =“ GPT-4O”）。如果您願意，您可以將其交換，但是GPT-4通常是多步推理任務的強烈選擇。

工作流的基本工具

>工具是代理可以打電話以在其語言建模之外執行操作的功能。典型的工具包括：

###############################################################################
# 1. INSTALLATION
###############################################################################
# Make sure you have the following installed:
#   pip install llama-index langchain duckduckgo-search

###############################################################################
# 2. IMPORTS
###############################################################################
%pip install llama-index langchain duckduckgo-search

from llama_index.llms.openai import OpenAI

# For DuckDuckGo search via LangChain
from langchain.utilities import DuckDuckGoSearchAPIWrapper

# llama-index workflow classes
from llama_index.core.workflow import Context
from llama_index.core.agent.workflow import (
    FunctionAgent,
    AgentWorkflow,
    AgentInput,
    AgentOutput,
    ToolCall,
    ToolCallResult,
    AgentStream
)

import asyncio

###############################################################################
# 3. CREATE LLM
###############################################################################
# Replace "sk-..." with your actual OpenAI API key
llm = OpenAI(model="gpt-4", api_key="OPENAI_API_KEY")

> Web搜索

• 讀取/寫文件
• 數學計算器 外部服務的
• API>
• 在我們的示例中，關鍵工具是DuckDuckgoSearch，它在引擎蓋下使用Langchain的Duckduckgosearchapiwrapper。我們還擁有輔助工具來記錄筆記，編寫報告並進行審查。 > 定義任務執行的AI代理

每個代理是

> functionagent

的實例。關鍵字段包括：

###############################################################################
# 4. DEFINE DUCKDUCKGO SEARCH TOOL WITH SAFEGUARDS
###############################################################################
# We wrap LangChain's DuckDuckGoSearchAPIWrapper with our own logic
# to prevent repeated or excessive searches.

duckduckgo = DuckDuckGoSearchAPIWrapper()

MAX_SEARCH_CALLS = 2
search_call_count = 0
past_queries = set()

async def safe_duckduckgo_search(query: str) -> str:
    """
    A DuckDuckGo-based search function that:
      1) Prevents more than MAX_SEARCH_CALLS total searches.
      2) Skips duplicate queries.
    """
    global search_call_count, past_queries

    # Check for duplicate queries
    if query in past_queries:
        return f"Already searched for '{query}'. Avoiding duplicate search."

    # Check if we've reached the max search calls
    if search_call_count >= MAX_SEARCH_CALLS:
        return "Search limit reached, no more searches allowed."

    # Otherwise, perform the search
    search_call_count += 1
    past_queries.add(query)

    # DuckDuckGoSearchAPIWrapper.run(...) is synchronous, but we have an async signature
    result = duckduckgo.run(query)
    return str(result)
    
###############################################################################
# 5. OTHER TOOL FUNCTIONS: record_notes, write_report, review_report
###############################################################################
async def record_notes(ctx: Context, notes: str, notes_title: str) -> str:
    """Store research notes under a given title in the shared context."""
    current_state = await ctx.get("state")
    if "research_notes" not in current_state:
        current_state["research_notes"] = {}
    current_state["research_notes"][notes_title] = notes
    await ctx.set("state", current_state)
    return "Notes recorded."

async def write_report(ctx: Context, report_content: str) -> str:
    """Write a report in markdown, storing it in the shared context."""
    current_state = await ctx.get("state")
    current_state["report_content"] = report_content
    await ctx.set("state", current_state)
    return "Report written."

async def review_report(ctx: Context, review: str) -> str:
    """Review the report and store feedback in the shared context."""
    current_state = await ctx.get("state")
    current_state["review"] = review
    await ctx.set("state", current_state)
    return "Report reviewed."    

• > name 和描述
• system_prompt：指示代理有關其角色和約束
• llm：使用的語言模型
>
• 工具：哪個功能代理可以調用​​
>
• can_handoff_to ：該代理可以將控制權交給
>

研究成分

• 搜索Web（最多可查詢的指定限制）
>
• 將相關發現保存為“筆記”
> 收集到足夠的信息後，
• >

writeagent

在Markdown中撰寫該報告，使用任何註釋收集的研究
移交給審查的反饋

評論

審查內容草案以獲取正確性和完整性
如果需要更改，則手控制回到寫入
否則，提供最終批准

###############################################################################
# 1. INSTALLATION
###############################################################################
# Make sure you have the following installed:
#   pip install llama-index langchain duckduckgo-search

###############################################################################
# 2. IMPORTS
###############################################################################
%pip install llama-index langchain duckduckgo-search

from llama_index.llms.openai import OpenAI

# For DuckDuckGo search via LangChain
from langchain.utilities import DuckDuckGoSearchAPIWrapper

# llama-index workflow classes
from llama_index.core.workflow import Context
from llama_index.core.agent.workflow import (
    FunctionAgent,
    AgentWorkflow,
    AgentInput,
    AgentOutput,
    ToolCall,
    ToolCallResult,
    AgentStream
)

import asyncio

###############################################################################
# 3. CREATE LLM
###############################################################################
# Replace "sk-..." with your actual OpenAI API key
llm = OpenAI(model="gpt-4", api_key="OPENAI_API_KEY")

代理工作流 - 協調任務執行

> AgentWorkflow協調消息和狀態在代理之間的移動方式。當用戶啟動請求時（例如，“給我寫一份關於Internet歷史的簡潔報告……”），工作流程：

：

• > researchAgent接收用戶提示，並決定是執行Web搜索還是記錄一些筆記。
• > writeagent>使用註釋創建結構化或樣式的輸出（例如Markdown文檔）。
• >評論檢查最終輸出，然後將其發送回修訂或批准。 > 一旦批准了內容且不要求進一步更改，工作流程結束了。
構建工作流

在此步驟中，我們定義了代理工作流程，其中包括研究，寫作和審查代理。 root_agent設置為Research_agent，這意味著該過程始於收集研究。初始狀態包含用於研究筆記，報告內容和審查狀態的佔位符。

運行工作流程

>使用用戶請求執行工作流程，該請求指定主題和報告中要涵蓋的要點。此示例中的請求要求提供有關互聯網歷史的簡明報告，包括其起源，萬維網的發展以及其現代發展。工作流程通過協調代理來處理此請求。

###############################################################################
# 4. DEFINE DUCKDUCKGO SEARCH TOOL WITH SAFEGUARDS
###############################################################################
# We wrap LangChain's DuckDuckGoSearchAPIWrapper with our own logic
# to prevent repeated or excessive searches.

duckduckgo = DuckDuckGoSearchAPIWrapper()

MAX_SEARCH_CALLS = 2
search_call_count = 0
past_queries = set()

async def safe_duckduckgo_search(query: str) -> str:
    """
    A DuckDuckGo-based search function that:
      1) Prevents more than MAX_SEARCH_CALLS total searches.
      2) Skips duplicate queries.
    """
    global search_call_count, past_queries

    # Check for duplicate queries
    if query in past_queries:
        return f"Already searched for '{query}'. Avoiding duplicate search."

    # Check if we've reached the max search calls
    if search_call_count >= MAX_SEARCH_CALLS:
        return "Search limit reached, no more searches allowed."

    # Otherwise, perform the search
    search_call_count += 1
    past_queries.add(query)

    # DuckDuckGoSearchAPIWrapper.run(...) is synchronous, but we have an async signature
    result = duckduckgo.run(query)
    return str(result)
    
###############################################################################
# 5. OTHER TOOL FUNCTIONS: record_notes, write_report, review_report
###############################################################################
async def record_notes(ctx: Context, notes: str, notes_title: str) -> str:
    """Store research notes under a given title in the shared context."""
    current_state = await ctx.get("state")
    if "research_notes" not in current_state:
        current_state["research_notes"] = {}
    current_state["research_notes"][notes_title] = notes
    await ctx.set("state", current_state)
    return "Notes recorded."

async def write_report(ctx: Context, report_content: str) -> str:
    """Write a report in markdown, storing it in the shared context."""
    current_state = await ctx.get("state")
    current_state["report_content"] = report_content
    await ctx.set("state", current_state)
    return "Report written."

async def review_report(ctx: Context, review: str) -> str:
    """Review the report and store feedback in the shared context."""
    current_state = await ctx.get("state")
    current_state["review"] = review
    await ctx.set("state", current_state)
    return "Report reviewed."    

>用於調試或觀察的流動事件

要監視工作流的執行，我們流式傳輸事件並打印有關代理活動的詳細信息。這使我們能夠跟踪當前正在工作的代理，查看中間輸出並檢查代理進行的工具調用。顯示調試信息，例如工具使用和響應，以提高可見性。

###############################################################################
# 1. INSTALLATION
###############################################################################
# Make sure you have the following installed:
#   pip install llama-index langchain duckduckgo-search

###############################################################################
# 2. IMPORTS
###############################################################################
%pip install llama-index langchain duckduckgo-search

from llama_index.llms.openai import OpenAI

# For DuckDuckGo search via LangChain
from langchain.utilities import DuckDuckGoSearchAPIWrapper

# llama-index workflow classes
from llama_index.core.workflow import Context
from llama_index.core.agent.workflow import (
    FunctionAgent,
    AgentWorkflow,
    AgentInput,
    AgentOutput,
    ToolCall,
    ToolCallResult,
    AgentStream
)

import asyncio

###############################################################################
# 3. CREATE LLM
###############################################################################
# Replace "sk-..." with your actual OpenAI API key
llm = OpenAI(model="gpt-4", api_key="OPENAI_API_KEY")

檢索並打印最終報告

>工作流完成後，我們提取最終狀態，其中包含生成的報告。印刷報告內容，然後是評論代理的任何審核反饋。這樣可以確保輸出完成，並在必要時可以進一步完善。

>

###############################################################################
# 4. DEFINE DUCKDUCKGO SEARCH TOOL WITH SAFEGUARDS
###############################################################################
# We wrap LangChain's DuckDuckGoSearchAPIWrapper with our own logic
# to prevent repeated or excessive searches.

duckduckgo = DuckDuckGoSearchAPIWrapper()

MAX_SEARCH_CALLS = 2
search_call_count = 0
past_queries = set()

async def safe_duckduckgo_search(query: str) -> str:
    """
    A DuckDuckGo-based search function that:
      1) Prevents more than MAX_SEARCH_CALLS total searches.
      2) Skips duplicate queries.
    """
    global search_call_count, past_queries

    # Check for duplicate queries
    if query in past_queries:
        return f"Already searched for '{query}'. Avoiding duplicate search."

    # Check if we've reached the max search calls
    if search_call_count >= MAX_SEARCH_CALLS:
        return "Search limit reached, no more searches allowed."

    # Otherwise, perform the search
    search_call_count += 1
    past_queries.add(query)

    # DuckDuckGoSearchAPIWrapper.run(...) is synchronous, but we have an async signature
    result = duckduckgo.run(query)
    return str(result)
    
###############################################################################
# 5. OTHER TOOL FUNCTIONS: record_notes, write_report, review_report
###############################################################################
async def record_notes(ctx: Context, notes: str, notes_title: str) -> str:
    """Store research notes under a given title in the shared context."""
    current_state = await ctx.get("state")
    if "research_notes" not in current_state:
        current_state["research_notes"] = {}
    current_state["research_notes"][notes_title] = notes
    await ctx.set("state", current_state)
    return "Notes recorded."

async def write_report(ctx: Context, report_content: str) -> str:
    """Write a report in markdown, storing it in the shared context."""
    current_state = await ctx.get("state")
    current_state["report_content"] = report_content
    await ctx.set("state", current_state)
    return "Report written."

async def review_report(ctx: Context, review: str) -> str:
    """Review the report and store feedback in the shared context."""
    current_state = await ctx.get("state")
    current_state["review"] = review
    await ctx.set("state", current_state)
    return "Report reviewed."    

避免無限搜索循環

使用Web搜索工具時，LLM可能會“混淆”並反複調用搜索功能。這可能導致不必要的成本或時間消耗。為了防止這種情況，我們使用了兩種機制：

• Hard Limitwe Setmax_search_calls = 2，因此研究工具只能稱為兩次。
• >重複dentectionwe在集合中存儲過去的查詢（past_queries），以避免多次重複完全相同的搜索。
>

如果滿足了任何一個條件（最大搜索或重複查詢），我們的函數將返回罐頭消息，而不是執行新搜索。 >

期望什麼？

研究

接收用戶請求，以編寫有關Internet歷史記錄的簡明報告。 >
• >可能會執行最多兩個不同的DuckDuckgo搜索（例如，“互聯網的歷史記錄”和“ World Web Web Tim Berners-Lee”等），然後致電record_notes存儲摘要。
>

writeagent

>從共享上下文中讀取“ research_notes”。

>
• 草稿簡短的Markdown報告。
• 移交給審查。
>

評論

評估內容。

• 如果需要更改，則可以將控制權傳遞回寫入。否則，它批准了報告。

工作流結束

最終輸出存儲在

final_state [“ report_content”]。

結論 >通過將工作流程分為不同的代理，以獲取

> search

，

寫作

和>評論>，您可以創建一個功能強大的模塊化系統，該系統： >收集相關信息（以一種受控的方式，防止搜索過多）

產生結構化的高質量輸出

• >自我檢查是否準確性和完整性
>使用Langchain的DuckDuckgo集成為多代理工作流提供了插件的Web搜索解決方案，而無需使用專門的API鍵或憑據。結合內置的保障措施（搜索呼叫限制，重複檢測），該系統穩健，高效且適用於廣泛的研究和內容生成任務。
• >

  鑰匙要點

  通過向LLM代理分配專業角色，
  • 多代理工作流程提高了效率。
  • 使用DuckDuckGo之類的外部工具增強了LLM代理的研究功能。
  • 實施約束，例如搜索限制，可以防止不必要的資源消耗。
  • >協調的代理工作流程確保結構化的高質量內容生成。
  設計精心設計的交接機制有助於避免多餘的任務和無限的循環。
  常見問題

  > Q1。為什麼要使用多個代理代替單一的通用代理？跨代理商（研究，寫作，審查）分配職責可確保每個步驟都明確定義且易於管理。它還減少了模型決策中的混亂，並促進了更準確的結構化輸出。我如何限制Web搜索的數量？在代碼中，我們使用全局計數器（search_call_count）和一個常數（max_search_calls = 2）。每當搜索代理調用safe_duckduckgo_search時，它都會檢查計數器是否達到限制。如果是這樣，它將返回一條消息，而不是執行另一個搜索。

  Q3。如果代理多次重複相同的查詢怎麼辦？我們維護一個名為cast_queries的python集，以檢測重複的查詢。如果查詢已經在該集合中，則該工具將跳過執行實際搜索並返回簡短消息，從而阻止重複查詢運行。

  Q4。我可以更改提示以適應此工作流程的其他主題或樣式嗎？絕對地。您可以編輯每個代理的System_prompt，以根據所需的域或寫作方式量身定制說明。例如，您可以指示寫入列表，敘述性論文或技術摘要。

  Q5。我需要GPT-4，還是可以使用其他型號？您可以將OpenAI（Model =“ GPT-4”）交換為由Llama-Index支持的另一個模型（例如GPT-3.5，甚至是本地模型）。該體系結構保持不變，儘管某些模型可能會產生不同的質量輸出。 >

  >本文所示的媒體不歸Analytics Vidhya擁有，並由作者的酌情決定使用。

以上是LlamainDex進行研究和寫作的多代理工作流程的詳細內容。更多資訊請關注PHP中文網其他相關文章！