LeetCode智能教練來襲！多智能體LLM協同架構實戰指南 原創 精華
在人工智能飛速發展的今天，生成式AI（Generative AI）已經成為了技術領域的一顆明珠。從簡單的文本生成到復雜的多模態應用，生成式AI正在不斷刷新我們對機器智能的認知。然而，隨著技術的不斷進步，我們發現僅僅依靠單一的大型語言模型（LLM）已經無法滿足日益復雜的應用需求。于是，多智能體LLM協同架構應運而生，它通過多個智能體的協作，極大地提升了生成式AI應用的性能和靈活性。今天，就讓我們一起深入探索多智能體LLM協同的魅力，并通過一個實際案例——打造一個LeetCode智能教練，來感受它的強大能力。
一、什么是智能體（Agents）？
在生成式AI的世界里，智能體（Agents）是一種能夠利用LLM進行決策、控制應用流程、規劃復雜推理步驟、利用工具以及自主與其他智能體交互的系統。簡單來說，智能體就像是AI應用中的“小助手”，它們可以根據用戶的輸入和任務需求，自主地選擇行動路徑、調用工具或者與其他智能體合作，從而完成復雜的任務。
舉個例子，想象一下你正在開發一個智能客服系統。傳統的做法是將所有的邏輯和決策都集中在一個大型的語言模型中，但這種方法往往會導致系統在處理復雜問題時顯得力不從心。而如果你引入了智能體，就可以讓不同的智能體分別負責不同的任務，比如一個智能體專門負責理解用戶的問題，另一個智能體負責從數據庫中檢索信息，還有一個智能體負責生成最終的回答。這樣一來，整個系統不僅更加高效，而且能夠更好地應對復雜的用戶需求。
二、多智能體架構的魅力
多智能體系統的核心在于多個獨立的智能體之間的協作。每個智能體都有自己的角色和上下文，它們可以根據自己的任務需求調用工具并與其他智能體互動。這種架構的好處是顯而易見的：
（一）降低偏見和幻覺
在單智能體系統中，模型可能會因為自身的局限性而產生偏見或者生成不準確的內容（我們通常稱之為“幻覺”）。而在多智能體系統中，由于每個智能體都可以獨立地對任務進行推理，因此可以通過多個智能體的協作來減少這種偏見和幻覺。例如，當一個智能體生成了一個可能存在問題的回答時，其他智能體可以通過自己的判斷和驗證來糾正這個錯誤。
（二）簡化復雜任務
復雜的任務往往需要多個步驟來完成，而每個步驟都可能需要不同的技能和知識。在多智能體系統中，每個智能體都可以專注于自己擅長的任務，從而將復雜的任務分解為多個簡單的子任務。這樣不僅可以提高系統的效率，還可以讓每個智能體都能夠在自己的領域內發揮最大的優勢。
（三）提升系統的靈活性和可擴展性
多智能體架構使得系統可以根據不同的需求靈活地調整智能體的數量和角色。例如，當你需要為系統增加一個新的功能時，你只需要添加一個新的智能體并定義它的角色和任務，而不需要對整個系統進行大規模的修改。這種靈活性和可擴展性使得多智能體系統在面對復雜多變的應用場景時具有巨大的優勢。
三、多智能體架構的常見類型
多智能體系統可以根據不同的需求和應用場景采用不同的架構。以下是最常見的三種架構類型：
（一）網絡架構（Network）
在網絡架構中，每個智能體都可以與其他所有智能體直接交互。這種架構的優點是靈活性高，每個智能體都可以根據自己的需求選擇與其他智能體的交互方式。然而，這種架構的缺點是可能會導致交互過程變得非常復雜，尤其是在智能體數量較多時。想象一下，如果一個系統中有幾十個智能體，每個智能體都可以與其他智能體交互，那么整個系統的交互關系就會變得像一張錯綜復雜的蜘蛛網。
（二）監督架構（Supervisor）
在監督架構中，有一個專門的“監督者”智能體負責協調其他所有智能體的工作。這個監督者智能體可以根據用戶的輸入和任務需求，決定調用哪些智能體以及它們的調用順序。這種架構的優點是結構清晰，監督者智能體可以對整個系統的流程進行有效的控制。然而，這種架構的缺點是監督者智能體可能會成為系統的瓶頸，如果它的決策能力不足，可能會影響整個系統的性能。
（三）層次架構（Hierarchical）
層次架構是在監督架構的基礎上進行擴展的一種架構。它不僅有一個監督者智能體，還可能有多個“子監督者”智能體，形成一個層次化的結構。這種架構的優點是可以處理更加復雜的任務，因為它可以通過多個層次的智能體來實現更加精細的控制。然而，這種架構的缺點是設計和實現起來相對復雜，需要仔細地規劃每個層次的智能體的角色和任務。
四、用LangGraph打造LeetCode智能教練
了解了多智能體架構的基本概念和優勢之后，讓我們通過一個實際的案例來感受它的強大能力。我們將使用LangGraph框架來打造一個LeetCode智能教練。這個智能教練能夠根據用戶的需求生成定制化的LeetCode風格的編程問題，并從互聯網上搜索相關的資源來幫助用戶學習和解答。
（一）準備工作
在開始之前，我們需要做一些準備工作。首先，我們需要安裝一些必要的Python包，包括LangGraph、LangChain、OpenAI等。這些包將為我們提供構建多智能體系統所需的各種工具和接口。其次，我們需要獲取OpenAI和Tavily的API密鑰，因為我們將使用這些API來調用語言模型和搜索工具。
以下是安裝Python包的命令：
pip install python-dotenv langgraph langchain langchain_openai langchain_community langchain_text_splitters chromadb tavily-python openai typing_extensions termcolor然后，創建一個??.env??文件，并在其中定義API密鑰：
OPENAI_API_KEY={INSERT_YOUR_OPENAI_KEY}
TAVILY_API_KEY={INSERT_YOUR_TAVILY_KEY}（二）定義智能體
在我們的LeetCode智能教練中，我們將定義兩個輔助智能體：資源查找器（Resource-Finder）和問題生成器（Problem-Generator）。
1. 資源查找器（Resource-Finder）
資源查找器的作用是根據用戶的輸入，在互聯網上搜索相關的資源。我們將使用Tavily工具來實現這個功能。通過定義一個合適的提示（prompt），我們可以告訴智能體它的角色和任務是什么。然后，我們將Tavily工具綁定到語言模型上，并構建一個工具鏈來實現資源的搜索和返回。
以下是資源查找器的代碼實現：
from langchain.prompts import ChatPromptTemplate
from langchain.llms import ChatOpenAI
from langchain.tools import TavilySearchResults
from langchain.chains import LLMChain
from langgraph import Command, State, TypedDict, Annotated, List
def resource_finder(state: State) -> Command[Literal["Supervisor"]]:
"""Finds resources based on the user's query."""
query = state["messages"][-1].content
context = state["messages"]
# Define prompt
prompt = ChatPromptTemplate.from_template(
f"""You are a computer science resource finder.
You will find resources for the user based on their preferences
in the field of computer science and related topics.
Use the context to help you build the response. You may also answer
general questions or miscellaneous queries.
Question: {query}\n"""
)
# Define tool and bind LLM to Tavily tool
tavily_tool = TavilySearchResults(max_results=5, search_depth="advanced", include_answer=True, include_raw_cnotallow=True)
llm = ChatOpenAI(model="gpt-4o").bind_tools([tavily_tool])
# Define LLM chain
llm_chain = (
prompt
| llm
)
@chain
def tool_chain(user_input: str, config: RunnableConfig):
input_ = {"user_input": user_input, "context": context}
ai_msg = llm_chain.invoke(input_, cnotallow=config)
tool_msgs = tavily_tool.batch(ai_msg.tool_calls, cnotallow=config)
return llm_chain.invoke({**input_, "messages": [ai_msg, *tool_msgs]}, cnotallow=config)
response = tool_chain.invoke(query)
return Command(
update={
"messages": [
AIMessage(cnotallow=response.content, name="Resource-Finder")
]
},
goto="Supervisor",
)2. 問題生成器（Problem-Generator）
問題生成器的作用是根據用戶的輸入生成LeetCode風格的編程問題。我們將使用一種流行的生成式AI技術——檢索增強生成（RAG）。具體來說，我們將使用一個包含近2000個LeetCode問題的Kaggle數據集作為知識庫。通過檢索與用戶輸入相關的數據，并結合語言模型的生成能力，我們可以生成定制化的編程問題。
以下是問題生成器的代碼實現：
from langchain.loaders import CSVLoader
from langchain.text_splitters import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.retrievers import VectorStoreRetriever
from langchain.prompts import ChatPromptTemplate
from langchain.llms import ChatOpenAI
from langchain.chains import LLMChain
from langgraph import Command, State, TypedDict, Annotated, List
def coding_problem_generator(state: State) -> Command[Literal["Supervisor"]]:
"""Generates a coding problem based on the user's query."""
query = state["messages"][-1].content
# Using LeetCode dataset for RAG to generate coding problems
coding_dataset = "leetcode_dataset - lc.csv"
# Load CSV file as input document
loader = CSVLoader(file_path=coding_dataset)
docs = loader.load()
# Split documents into smaller chunks and store in Chroma vectorstore
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits = text_splitter.split_documents(docs)
vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())
# Define retriever using vector store
retriever = vectorstore.as_retriever()
# Define prompt
prompt = ChatPromptTemplate.from_template(
""""You are a LeetCode-style coding problem generator.
You will generate a coding problem for the user to solve
based on their preferences. Use the context to help you build the problem.
Do not provide a solution to the problem unless asked for.
Context: {context}\n
Answer:"""
)
def format_docs(docs):
return"\n\n".join(doc.page_content for doc in docs)
# Define LLM model
llm = ChatOpenAI(model="gpt-4o")
# Build RAG chain using retriever, prompt, and LLM
rag_chain = (
{"context": retriever | format_docs}
| prompt
| llm
| StrOutputParser()
)
# Invoke RAG chain to generate coding problem
response = rag_chain.invoke(query)
return Command(
update={
"messages": [
AIMessage(cnotallow=response, name="Problem-Generator")
]
},
goto="Supervisor",
)（三）定義監督者智能體
監督者智能體的作用是協調整個系統的流程。它將根據用戶的輸入和當前的交互狀態，決定調用哪個智能體或者結束當前的任務。我們將定義一個合適的提示來告訴監督者智能體它的角色和任務，并使用語言模型來生成結構化的輸出，從而決定下一步的行動。
以下是監督者智能體的代碼實現：
from langgraph import Command, State, TypedDict, Annotated, List
from langchain.llms import ChatOpenAI
class Router(TypedDict):
next: Literal["Resource-Finder", "Problem-Generator", "FINISH"]
def supervisor_agent(state: State) -> Command[Literal["Resource-Finder", "Problem-Generator", "__end__"]]:
"""Supervisor agent that manages the conversation between workers."""
question = state["messages"][-1].content
# Include the system prompt and the current conversation state in the messages
members = ["Resource-Finder", "Problem-Generator"]
system_prompt = (
"You are a supervisor tasked with managing a conversation between the"
f" following workers: {members}. Given the following user request {question},"
" respond with the worker to act next. Each worker will perform a"
" task and respond with their results and status. When you determine a task to be finished,"
" respond with FINISH."
" Here are the uses of each worker:\n"
"1. Resource-Finder: Find resources based on the user's query and handles any general or miscellaneous user queries.\n"
"2. Problem-Generator: Generate a coding problem based on the user's query.\n"
)
messages = [
{"role": "system", "content": system_prompt},
] + state["messages"]
llm = ChatOpenAI(model="gpt-4o")
# Use the LLM to decide the next step
response = llm.with_structured_output(Router).invoke(messages)
# Extract the next node from the response
next_node = response.get("next", None)
ifnot next_node:
raise ValueError("Supervisor failed to determine the next step.")
if next_node == "FINISH":
next_node = "__end__"
# Return a Command with the target node in the goto field.
return Command(goto=next_node, update={"next": next_node})（四）構建和運行多智能體系統
在定義了所有的智能體之后，我們需要將它們組合成一個多智能體系統。我們將使用LangGraph框架來構建這個系統，并定義智能體之間的交互關系。最后，我們將運行這個系統，并通過與用戶的交互來測試它的性能。
以下是構建和運行系統的代碼實現：
from langgraph import StateGraph, START, END
import os
from dotenv import load_dotenv
from termcolor import colored
# Load environment variables
load_dotenv()
# Define the graph state
class State(TypedDict):
messages: Annotated[List, add_messages] = []
next: str = ""
# Create the state graph
def create_graph():
workflow = StateGraph(State)
workflow.add_node("Resource-Finder", resource_finder)
workflow.add_node("Problem-Generator", coding_problem_generator)
workflow.add_node("Supervisor", supervisor_agent)
workflow.add_edge(START, "Supervisor")
graph = workflow.compile()
return graph
# Main function to run the graph
def main():
# Create the state graph
graph = create_graph()
# Print out the LangGraph as ASCII
graph.get_graph().print_ascii()
# Continuous input and LLM interaction
print(colored("You can start interacting with the coding assistant. Type 'exit' to end the conversation.", "blue"))
whileTrue:
user_message = input("> ")
if user_message == "exit":
print(colored("Goodbye!", "blue"))
break
input_state = {"messages": [{"role": "user", "content": user_message}]}
# Verbose output
for event in graph.stream(input_state):
print(colored(event, "red"))
print("------------------------------------")
# # Concise output
# final_state = graph.invoke(input_state, config)
# print(colored(final_state["messages"][-1].content, "red"))
# print("------------------------------------")
if __name__ == "__main__":
main()五、實際運行效果
通過實際運行我們的LeetCode智能教練，我們可以看到多智能體系統在處理復雜任務時的強大能力。無論是生成定制化的編程問題，還是從互聯網上搜索相關的資源，我們的系統都能夠快速、準確地完成任務。而且，由于采用了多智能體架構，我們的系統在處理復雜的多步驟任務時也表現出了良好的靈活性和適應性。
六、總結與展望
通過這個案例，我們深刻體會到了多智能體LLM協同架構的強大優勢。它不僅能夠提升系統的性能和靈活性，還能夠降低偏見和幻覺，簡化復雜任務的處理過程。在未來，隨著生成式AI技術的不斷發展，多智能體架構必將在更多的領域發揮重要作用。我們期待看到更多基于多智能體架構的創新應用，為我們的生活和工作帶來更多的便利和驚喜。
本文轉載自公眾號Halo咯咯 作者：基咯咯
