萬字長文剖析企業級多智能體案例架構設計與代碼級落地實踐原創

玄姐聊AGI

發布于 2025-10-10 06:42

瀏覽

0收藏

1. 前言

本文會從零構建一個企業級多智能體架構與代碼級落地實踐，不依賴 LangChain 或 CrewAI 這類高層級編排庫。我們會把整套邏輯拆解為三個簡單且符合邏輯的模塊：

萬字長文剖析企業級多智能體案例架構設計與代碼級落地實踐-AI.x社區

一個??Agent?? 類：負責思考、行動，并維護自身的推理循環；
一個??Tool?? 類：智能體可調用的工具，用于與外部世界交互；
一個??Crew?? 類：整合多個智能體，協調它們的工作流程。

通過這種方式，我們能完全掌控智能體的行為，更易優化和排查問題。文中將使用 OpenAI 作為大模型后端，但如果您更傾向于本地部署，也可以用 Ollama（本地運行大模型的開源工具）搭配 Qwen3 等模型。

過程中，我們會講解多智能體模式的原理、設計 “協作與工具調用交織” 的智能體循環，并實現多個可供智能體調用的工具。最終，您將得到一個基于純 Python 和大模型后端的可運行多智能體系統，其中的智能體既能獨立推理、調用工具，又能在受控流程中傳遞結果。

下文我們詳細剖析之。

2. 為什么需要多智能體？

我們已經知道，單個大模型智能體能力很強，但面對復雜的多步驟任務時，它的處理能力會受限。而 “多智能體方案”（用一組專業智能體協作）已成為解決復雜流程的有效方式。

萬字長文剖析企業級多智能體案例架構設計與代碼級落地實踐-AI.x社區

將大問題拆解為多個子任務，分配給專門的智能體處理，這種方式相比 “單體智能體” 有諸多優勢：

萬字長文剖析企業級多智能體案例架構設計與代碼級落地實踐-AI.x社區

2.1 開發與調試更簡單

每個智能體專注于特定子任務或角色，整體系統的邏輯更清晰。比如修改某個智能體的提示詞或邏輯時，不會影響整個系統 —— 其他智能體的角色是獨立的（盡管一個智能體的輸出可能會被另一個使用）。這種 “職責分離” 意味著：升級某個智能體（比如優化 “調研智能體”）時，不會給其他智能體帶來意外影響，就像團隊中替換一位專家無需重新培訓所有人。

萬字長文剖析企業級多智能體案例架構設計與代碼級落地實踐-AI.x社區

2.2 降低復雜性，減少 “幻覺”

如果讓單個智能體用一個超長提示詞兼顧所有工具和指令，它很容易混淆或編造信息；而多個 “專注型” 智能體更能保持方向。比如在 “代碼助手” 工作流中，我們可以設置三個智能體：“規劃智能體” 負責拆解任務、“編碼智能體” 負責寫代碼、“審核智能體” 負責檢查代碼 —— 每個智能體的提示詞和工具都針對自身職責優化，能更高效地運用領域知識或專用工具。

萬字長文剖析企業級多智能體案例架構設計與代碼級落地實踐-AI.x社區

2.3 問題解決更高效

多智能體的 “分布式協作” 能讓復雜任務（比如規劃旅行）拆解為多個簡單子任務：一個智能體查天氣、一個找酒店、一個規劃路線…… 這些專業智能體協作解決整體問題的效率，遠高于單個 “全能型” 智能體。

2.4 推理過程更透明

多智能體系統天然具備 “明確的職責分工”，讓推理過程更易追蹤。每個智能體通常會 “清晰地思考”（通過 “思維鏈提示詞” 輸出推理過程），產生的中間結果也能被其他智能體或開發者查看。甚至，多智能體工作流會鼓勵智能體互相驗證輸出 —— 比如一個智能體生成計劃，另一個智能體評價或優化計劃，形成可追溯的對話記錄。這種 “審計軌跡” 能提升可解釋性：您能看到每個 “專家” 的決策依據，而不是單個黑箱模型的輸出。

2.5 問題定位更快速

當多智能體流程出錯時，更容易定位問題根源。因為任務是分段的，您能通過詳細日志判斷 “哪個智能體”“哪一步” 產生了錯誤輸出或卡住（比如查看 “調研智能體” 的搜索結果、“撰寫智能體” 的草稿摘要）。這比排查 “一個做所有事的巨型智能體” 要簡單得多。

2.6 符合人類團隊協作邏輯

多智能體模式模仿了人類團隊的運作方式，讓 AI 工作流的設計更自然。在企業中，人們會分角色協作（分析師、規劃師、構建者、審核者）；同理，多智能體系統中，每個智能體就像團隊成員，有明確職責，協調者確保它們協同工作。這種映射讓設計復雜 AI 方案變得直觀 —— 只需思考 “我的團隊需要哪些角色的智能體”。比如 “AI 旅行 concierge 服務” 可設計為一組智能體：一個查景點、一個處理交通與天氣、一個優化行程，共同響應用戶需求。

3. 多智能體的內部原理

如前所述，多智能體模式將 AI 工作流設計為 “一組協作的智能體”，每個智能體有明確角色。它不是讓一個智能體從頭處理任務，而是讓多個智能體形成流水線：每個智能體處理任務的一部分，再將結果傳遞給下一個，最終完成目標。

萬字長文剖析企業級多智能體案例架構設計與代碼級落地實踐-AI.x社區

這種設計通常（但不總是）由 “協調器” 把控 —— 確保智能體按正確順序運行、共享信息。這就像流水線或接力賽：智能體 A 完成第一步，智能體 B 用 A 的輸出做第二步，依此類推直到任務完成。

每個智能體在處理子問題時具備自主性，但會通過定義好的工作流協作解決整體問題。從技術角度看，多智能體中的每個智能體，內部仍遵循 “推理 + 行動” 循環（通常基于我們之前學過的 ReAct 范式）：接收輸入（或上下文）→ 思考要做什么（生成 “Thought”）→ 行動（輸出結果或調用工具）→ 觀察結果→ 繼續循環。

關鍵區別在于：多智能體中的每個智能體 “職責范圍有限”—— 它的關注點更窄、工具集更小、目標更明確。比如一個負責 “數據庫查詢” 的智能體，只會思考 “如何查詢數據庫” 并執行查詢操作，而不會操心 “如何向用戶展示最終答案”（這是另一個智能體的工作）。通過限制智能體的職責，我們能讓它的提示詞更簡潔、更有針對性，行為也更可靠。

4. 系統整體架構

為理清多智能體架構的組成部分，我們拆解一下核心模塊：

4.1 智能體（Agent）

智能體是 “自主 AI 單元”（通常由大模型 + 提示詞構成），能感知輸入、推理（通過思維鏈）、執行行動以完成子任務。每個智能體都會配置特定角色，且只能訪問完成該角色所需的工具或信息。

比如，我們可以創建一個 ??ResearchAgent???（調研智能體），它能使用 “網頁搜索工具”；再創建一個 ??SummaryAgent??（摘要智能體），它能使用 “文本生成工具”。智能體會循環 “思考→行動”，直到完成自身負責的子任務。由于職責聚焦，它能遵循嚴格的提示格式或協議（通常由框架定義），確保行為安全、不偏離任務。

4.2 工具（Tool）

工具是智能體可調用的 “外部能力”，用于與現實世界交互或獲取信息，比如網頁搜索 API、計算器、數據庫查詢接口、發郵件函數等。在多智能體模式中，工具是智能體可執行的特定行動，且每個智能體的工具集都與自身角色相關。

例如，調研智能體可能有 “網頁搜索” 工具，摘要智能體可能有 “知識庫查詢” 工具。智能體的 ReAct 循環會包含：選擇工具→ 給工具輸入參數（如搜索關鍵詞）→ 讀取工具輸出（觀察結果）→ 指導下一步思考。這種結構化的工具調用方式，能讓智能體的推理和行動更透明、可日志化（比如我們能看到它搜索了什么關鍵詞、得到了什么結果）。

萬字長文剖析企業級多智能體案例架構設計與代碼級落地實踐-AI.x社區

需要注意的是，LangChain 或 CrewAI 這類框架會定義工具接口，并要求智能體遵守格式（如工具輸入必須是 JSON），以確保交互可靠。

4.3 協作組（Crew）

Crew 本質是 “協調器”—— 一個（代碼中的）類，負責搭建多智能體工作流、管理執行順序。如果說每個智能體是團隊成員，Crew 就是 “團隊經理”，掌握整體計劃。

在 “順序流水線” 中，Crew 會把初始輸入傳給第一個智能體，再把它的輸出傳給第二個，直到最后一個智能體生成答案。它會定義智能體的運行順序（有時還會定義每個智能體對數據的具體處理任務）。比如用 CrewAI 庫時，我們可以實例化一個 ??Crew(agents=[智能體1, 智能體2, ...], tasks=[任務1, 任務2, ...], process=Process.sequential)???，然后調用 ??crew.kickoff()?? 讓智能體按順序運行。

Crew 還會處理協作細節，比如確保每個智能體拿到正確輸入、收集或整合輸出。在更復雜的場景中，協調器還能實現 “循環邏輯”（重復運行智能體或步驟直到滿足條件）或 “分支邏輯”（根據條件選擇調用哪個智能體）。

有些架構中，協調器本身也是一個智能體（常被稱為 “管理智能體” 或 “監督智能體”），職責是給其他專業智能體分配任務。無論它是顯式的類還是管理智能體，這個 “協調角色” 是讓多個智能體形成 “統一系統” 而非 “孤立機器人集合” 的關鍵。

4.4 信息流轉方式

多智能體流水線中，信息通常從一個智能體流向另一個。比如一個簡單流程：智能體 A 收集數據→ 智能體 B 分析數據→ 智能體 C 撰寫報告。Crew 會把初始查詢傳給智能體 A，A 用工具返回原始數據；數據傳給 B，B 生成分析結果；結果再傳給 C，C 輸出最終報告。每個智能體只處理與自身任務相關的輸入，無需關心整體任務 —— 這種 “順序傳遞” 就是多智能體模式的核心：多個 “專精型” AI 智能體串聯，輸出依次傳遞，直至完成目標。

5. 從零實現多智能體系統

前面我們已經了解了多智能體系統在模塊化、清晰度和可解釋性上的優勢。現在，我們動手從零實現一個多智能體系統，不依賴任何外部編排庫（如 LangChain 或 CrewAI）。

如前所述，系統將基于三個核心抽象類構建，下面我們逐一詳解每個類的原理和實現思路（您可直接下載文末代碼使用）：

5.1 工具類（Tool）

Tool 類會封裝實際功能（如新聞搜索、文本摘要、統計計算），并以 “智能體可識別、可驗證、可調用” 的方式暴露這些功能。每個工具包含：

解析后的函數簽名（讓智能體知道工具的輸入要求）；
自動輸入驗證和類型轉換（確保智能體傳入的參數合法）；
標準化調用接口（讓智能體調用工具的方式統一）。

通過這種設計，智能體無需硬編碼函數調用，就能自主判斷 “有哪些工具可用”“如何使用工具”。

5.2 智能體類（Agent）

Agent 類代表單個 AI 智能體，具備以下能力：

通過 ReAct 循環思考任務；
調用工具與環境交互；
將輸出傳遞給下游依賴的智能體。

每個智能體都有獨特的 “背景設定、任務目標、輸出要求和工具集”，模塊化和角色特異性極強。智能體還能定義依賴關系（如 “智能體 A 必須在智能體 B 之前運行”）。內部而言，每個智能體由 ??ReactAgent?? 包裝器驅動 —— 通過語言模型處理推理循環。

5.3 協作組類（Crew）

Crew 類是 “粘合劑”，負責協調所有組件，主要管理：

智能體注冊（將智能體加入系統）；
依賴解析（用拓撲排序確保智能體按依賴順序運行）；
有序執行（根據依賴關系安排智能體的運行順序）。

接下來的章節中，我們會詳細拆解每個類的實現細節，理解它們的工作原理，以及如何組合成完整的多智能體系統。首先，我們從 Tool 類開始。

6. 詳細實現：工具類（Tool）

工具是智能體與外部世界交互的 “手”，我們需要讓工具滿足兩個核心要求：智能體能看懂怎么用、調用后能返回可靠結果。下面從設計思路到代碼實現，一步步拆解 Tool 類。

6.1 設計核心：讓智能體 “清晰認知工具”

智能體沒辦法直接 “讀懂” Python 函數，所以 Tool 類需要把工具的功能、參數格式，轉換成自然語言描述 + 結構化簽名 —— 比如告訴智能體 “這個工具叫‘新聞搜索’，需要傳入‘關鍵詞’和‘時間范圍’兩個參數，能返回最近相關新聞列表”。

同時，為了避免智能體傳錯參數（比如把 “時間范圍” 寫成 “2025” 而不是 “2025-01-01 至 2025-01-10”），Tool 類還要包含輸入驗證邏輯，自動檢查參數合法性，不合法就返回錯誤提示，讓智能體修正后再調用。

6.2 代碼實現：基礎 Tool 類

我們用 Python 實現一個通用 Tool 類，后續所有具體工具（如新聞搜索、文本摘要）都可以繼承它，不用重復寫基礎邏輯。

from pydantic import BaseModel, ValidationError
from typing import Any, Callable, Dict, Optional
class Tool:
    def __init__(
        self,
        name: str,  # 工具名稱（給智能體看的，要簡潔，比如"news_search"）
        description: str,  # 工具功能描述（詳細說明能做什么，比如"根據關鍵詞搜索最近的新聞，返回標題和摘要"）
        func: Callable,  # 工具實際執行的函數（比如調用新聞API的函數）
        args_schema: Optional[BaseModel] = None  # 參數校驗規則（用Pydantic定義，確保參數合法）
    ):
        self.name = name
        self.description = description
        self.func = func
        # 用Pydantic模型定義參數格式，沒傳的話默認不校驗
        self.args_schema = args_schema if args_schema is not None else BaseModel
    def get_tool_info(self) -> Dict[str, str]:
        """生成工具信息，給智能體看（告訴它工具怎么用）"""
        # 提取參數描述（比如"關鍵詞：str，必填，搜索的核心詞；時間范圍：str，可選，格式如'2025-01-01至2025-01-10'）
        args_desc = ""
        if self.args_schema:
            for field_name, field in self.args_schema.model_fields.items():
                # 處理必填/可選、類型、描述
                required = "必填" if field.is_required() else "可選"
                field_type = str(field.annotation).split("[")[-1].split("]")[0] if "[" in str(field.annotation) else str(field.annotation)
                field_desc = field.description or "無描述"
                args_desc += f"- {field_name}：{field_type}，{required}，{field_desc}\n"


        return {
            "工具名稱": self.name,
            "工具功能": self.description,
            "參數說明": args_desc.strip() if args_desc else "無參數"
        }
    def run(self, args: Dict[str, Any]) -> Dict[str, Any]:
        """執行工具：先校驗參數，再調用函數，返回結果"""
        try:
            # 1. 校驗參數（用args_schema檢查格式）
            validated_args = self.args_schema(** args).dict()
        except ValidationError as e:
            # 參數不合法，返回錯誤提示，讓智能體修正
            error_msg = f"參數錯誤：{e.errors()[0]['msg']}，請檢查參數格式（參考：{self.get_tool_info()['參數說明']}）"
            return {"status": "error", "result": error_msg}


        try:
            # 2. 調用工具函數，執行實際操作
            result = self.func(** validated_args)
            # 3. 返回成功結果
            return {"status": "success", "result": result}
        except Exception as e:
            # 執行出錯（比如API調用失敗），返回錯誤信息
            return {"status": "error", "result": f"工具執行失敗：{str(e)}"}

有了基礎 Tool 類，我們可以快速創建實際能用的工具。比如做一個 “新聞搜索工具”，調用公開的新聞 API（這里用模擬函數演示，實際項目可替換成真實 API）。

from pydantic import Field
from datetime import datetime
# 1. 定義新聞搜索工具的參數校驗規則（用Pydantic）
class NewsSearchArgs(BaseModel):
    keyword: str = Field(..., descriptinotallow="搜索關鍵詞，比如'AI智能體'")  # ...表示必填
    time_range: str = Field(
        default=f"{datetime.now().strftime('%Y-%m-%d')}至{datetime.now().strftime('%Y-%m-%d')}",
        descriptinotallow="時間范圍，格式如'2025-01-01至2025-01-10'，默認當天"
    )
    max_results: int = Field(default=5, descriptinotallow="返回新聞數量，默認5條，最多20條")
# 2. 模擬新聞搜索函數（實際項目可替換成百度新聞API、新浪新聞API等）
def mock_news_search(keyword: str, time_range: str, max_results: int) -> list:
    # 模擬返回新聞數據（真實場景會調用API獲取）
    return [
        {
            "標題": f"{keyword}最新進展：多智能體協作效率提升30%",
            "摘要": "近日研究顯示，多智能體系統在復雜任務處理中，效率比單智能體高30%，已應用于金融分析場景。",
            "來源": "科技日報",
            "發布時間": time_range.split("至")[1]
        }
        for _ in range(min(max_results, 20))  # 確保不超過20條
    ]
# 3. 創建新聞搜索工具（繼承基礎Tool類）
news_search_tool = Tool(
    name="news_search",
    descriptinotallow="根據關鍵詞和時間范圍，搜索最新新聞，返回標題、摘要、來源和發布時間",
    func=mock_news_search,
    args_schema=NewsSearchArgs  # 綁定參數校驗規則
)
# 測試工具：調用看看效果
test_args = {"keyword": "AI多智能體", "time_range": "2025-01-01至2025-01-10", "max_results": 3}
test_result = news_search_tool.run(test_args)
print(test_result)
# 輸出（成功案例）：
# {
#     "status": "success",
#     "result": [
#         {"標題": "AI多智能體最新進展：多智能體協作效率提升30%", ...},
#         {"標題": "AI多智能體最新進展：多智能體協作效率提升30%", ...},
#         {"標題": "AI多智能體最新進展：多智能體協作效率提升30%", ...}
#     ]
# }
# 測試參數錯誤：max_results傳25（超過20）
error_args = {"keyword": "AI多智能體", "max_results": 25}
error_result = news_search_tool.run(error_args)
print(error_result)
# 輸出（錯誤案例）：
# {
#     "status": "error",
#     "result": "參數錯誤：確保該值小于等于20，...（參數說明）"
# }

7. 詳細實現：智能體類（Agent）

智能體是 “思考 + 行動” 的主體，核心是ReAct 循環：拿到輸入→思考 “該做什么”→決定 “調用工具還是直接輸出”→執行→根據結果繼續思考，直到完成子任務。

下面我們實現的 Agent 類，會包含 “接收上下文”“生成思考過程”“調用工具”“輸出結果” 四個核心能力，并且嚴格綁定自己的工具集（避免越權調用）。

7.1 設計核心：讓智能體 “有角色、會思考、能行動”

角色綁定：每個智能體有明確的 “角色描述”（比如 “新聞調研智能體，負責收集指定主題的最新新聞”），確保思考不偏離職責；
工具限制：智能體只能調用自己被分配的工具，不能用其他工具（比如 “調研智能體” 不能用 “報告生成工具”）；
思考透明：強制智能體輸出 “思考過程”（比如 “用戶需要 AI 多智能體的最新動態，我應該先調用新聞搜索工具，關鍵詞設為‘AI 多智能體’，時間范圍選最近 10 天”），方便調試和追蹤；
循環終止：智能體要能判斷 “什么時候該停止”（比如 “已經拿到 5 條新聞，足夠生成摘要，不需要再搜索了”），避免無限循環。

7.2 代碼實現：基礎 Agent 類

這里我們用 OpenAI 的 API 作為大模型后端（需要您自己準備 API Key），也可以替換成 Ollama 調用本地模型（修改 ??generate_thought?? 方法即可）。

import openai
from typing import List, Dict, Optional
from tool import Tool  # 導入前面實現的Tool類
# 配置OpenAI API（替換成您自己的Key）
openai.api_key = "your-openai-api-key"
class Agent:
    def __init__(
        self,
        name: str,  # 智能體名稱（比如"NewsResearchAgent"）
        role: str,  # 角色描述（詳細說明職責，比如"新聞調研智能體，負責根據用戶需求，調用新聞搜索工具收集最新相關新聞，確保信息準確、全面"）
        tools: List[Tool],  # 智能體可調用的工具列表
        model: str = "gpt-3.5-turbo"  # 使用的大模型
    ):
        self.name = name
        self.role = role
        self.tools = tools  # 綁定工具集（只能用這些工具）
        self.model = model
        # 記錄歷史交互（思考過程、工具調用、結果），用于上下文連貫
        self.history: List[Dict[str, str]] = []
    def get_available_tools_info(self) -> str:
        """生成智能體可調用的工具列表（給大模型看，告訴它有哪些工具可用）"""
        tools_info = "你可調用的工具如下：\n"
        for tool in self.tools:
            tool_info = tool.get_tool_info()
            tools_info += f"""
工具名稱：{tool_info['工具名稱']}
工具功能：{tool_info['工具功能']}
參數說明：{tool_info['參數說明']}
"""
        return tools_info.strip()
    def generate_thought(self, input_context: str) -> Dict[str, Optional[str]]:
        """生成思考過程，決定下一步行動：調用工具（返回工具名+參數）或直接輸出結果"""
        # 構建提示詞：包含角色、歷史記錄、可用工具、當前輸入
        prompt = f"""
你是{self.name}，你的角色是：{self.role}
【歷史交互記錄】
{[f"類型：{item['type']}，內容：{item['content']}" for item in self.history] if self.history else "無"}
【你可調用的工具】
{self.get_available_tools_info()}
【當前任務】
{input_context}
【思考要求】
1. 先分析：當前任務是否需要調用工具？如果已有足夠信息，直接輸出結果；如果需要更多信息，調用對應的工具。
2. 若調用工具：請嚴格按照格式返回：{{"action": "call_tool", "tool_name": "工具名稱", "tool_args": {{參數名: 參數值, ...}}}}
   - 工具名稱必須是【你可調用的工具】中的名稱，不能編造假工具。
   - 參數必須符合工具的【參數說明】，格式正確、必填參數不遺漏。
3. 若直接輸出：請嚴格按照格式返回：{{"action": "output_result", "result": "你的最終結果"}}
4. 思考過程不用寫出來，直接返回上述格式的JSON即可，不要加其他內容。
"""
        # 調用大模型生成思考結果
        response = openai.ChatCompletion.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3  # 降低隨機性，確保輸出穩定
        )
        thought_json = response.choices[0].message["content"]


        # 解析JSON（處理可能的格式問題）
        try:
            return eval(thought_json)  # 實際項目建議用json.loads，這里為了簡化用eval
        except Exception as e:
            # 解析失敗，返回錯誤提示，讓智能體重新思考
            return {"action": "error", "error_msg": f"思考結果格式錯誤：{str(e)}，請重新按照要求返回JSON"}
    def run(self, input_context: str, max_round: int = 5) -> Dict[str, str]:
        """執行智能體：啟動ReAct循環，最多循環max_round次（避免無限循環）"""
        current_context = input_context
        for round in range(1, max_round + 1):
            print(f"=== 智能體{self.name} - 第{round}輪思考 ===")


            # 1. 生成思考，決定下一步行動
            thought = self.generate_thought(current_context)
            print(f"思考結果：{thought}")


            # 2. 處理行動
            if thought["action"] == "output_result":
                # 直接輸出結果，循環終止
                self.history.append({"type": "output", "content": thought["result"]})
                return {"status": "success", "result": thought["result"], "history": self.history}


            elif thought["action"] == "call_tool":
                # 調用工具
                tool_name = thought["tool_name"]
                tool_args = thought["tool_args"]


                # 找到對應的工具
                target_tool = next((t for t in self.tools if t.name == tool_name), None)
                if not target_tool:
                    error_msg = f"工具不存在：{tool_name}，請檢查工具名稱"
                    self.history.append({"type": "error", "content": error_msg})
                    current_context = f"上一輪調用工具失敗：{error_msg}，請重新處理當前任務：{input_context}"
                    continue


                # 執行工具
                tool_result = target_tool.run(tool_args)
                self.history.append({
                    "type": "tool_call", 
                    "content": f"調用工具{tool_name}，參數：{tool_args}，結果：{tool_result}"
                })
                print(f"工具執行結果：{tool_result}")


                # 處理工具結果
                if tool_result["status"] == "success":
                    # 工具調用成功，更新上下文（把工具結果加入任務）
                    current_context = f"上一輪調用工具{tool_name}獲取到信息：{tool_result['result']}，請繼續處理原任務：{input_context}"
                else:
                    # 工具調用失敗，更新上下文（讓智能體修正）
                    current_context = f"上一輪調用工具{tool_name}失敗：{tool_result['result']}，請修正參數或選擇其他工具，繼續處理原任務：{input_context}"


            elif thought["action"] == "error":
                # 思考錯誤，更新上下文（讓智能體重新思考）
                self.history.append({"type": "error", "content": thought["error_msg"]})
                current_context = f"上一輪思考出錯：{thought['error_msg']}，請重新處理當前任務：{input_context}"


            else:
                # 未知行動，終止循環
                error_msg = f"未知行動類型：{thought['action']}"
                self.history.append({"type": "error", "content": error_msg})
                return {"status": "error", "result": error_msg, "history": self.history}


        # 循環次數用盡，仍未輸出結果
        error_msg = f"超過最大循環次數（{max_round}輪），未完成任務"
        self.history.append({"type": "error", "content": error_msg})
        return {"status": "error", "result": error_msg, "history": self.history}
# 測試智能體：創建新聞調研智能體，調用新聞搜索工具
if __name__ == "__main__":
    # 1. 準備工具（用前面創建的新聞搜索工具）
    tools = [news_search_tool]


    # 2. 創建智能體
    research_agent = Agent(
        name="NewsResearchAgent",
        role="新聞調研智能體：負責根據用戶需求，調用新聞搜索工具收集最新相關新聞，整理成清晰的列表（包含標題、摘要、來源），不需要額外添加分析內容",
        tools=tools
    )


    # 3. 運行智能體：任務是“收集2025年1月1日至2025年1月10日關于‘AI多智能體’的3條最新新聞”
    result = research_agent.run(
        input_cnotallow="收集2025年1月1日至2025年1月10日關于‘AI多智能體’的3條最新新聞，整理成列表",
        max_round=3
    )


    # 4. 輸出結果
    print("\n=== 最終結果 ===")
    print(result)

8. 詳細實現：協作組類（Crew）

有了工具和智能體，還需要一個 “協調者” 來管理它們的依賴關系和運行順序 —— 這就是 Crew 類的作用。比如 “先讓調研智能體收集新聞，再讓摘要智能體生成總結”，Crew 會確保這個順序不混亂，還能處理 “某個智能體失敗后是否重試” 等邏輯。

8.1 設計核心：讓多智能體 “有序協作”

依賴管理：支持智能體間的依賴設置（如 “摘要智能體” 依賴 “調研智能體” 的輸出），用拓撲排序確保先運行被依賴的智能體；
順序執行：按依賴關系依次運行智能體，把前一個智能體的輸出作為后一個的輸入；
錯誤處理：某個智能體失敗時，可選擇 “終止流程” 或 “跳過該智能體”（根據任務重要性配置）；
結果匯總：收集所有智能體的輸出，生成最終的整體結果，方便用戶查看。

8.2 代碼實現：基礎 Crew 類

from typing import List, Dict, Optional, Tuple
from agent import Agent  # 導入前面實現的Agent類
from tool import Tool  # 導入Tool類（用于類型提示）
class Crew:
    def __init__(
        self,
        agents: List[Agent],  # 協作組中的所有智能體
        dependencies: Optional[Dict[str, List[str]]] = None,  # 依賴關系：{智能體名稱: [依賴的智能體名稱列表]}
        max_retries: int = 2  # 智能體失敗后的最大重試次數
    ):
        self.agents = {agent.name: agent for agent in agents}  # 用字典存儲，方便按名稱查找
        self.dependencies = dependencies or {}  # 默認無依賴
        self.max_retries = max_retries
        # 驗證依賴關系（確保依賴的智能體存在）
        self._validate_dependencies()
    def _validate_dependencies(self) -> None:
        """驗證依賴關系：確保所有依賴的智能體都在協作組中"""
        all_agent_names = set(self.agents.keys())
        for agent_name, dep_names in self.dependencies.items():
            # 檢查當前智能體是否存在
            if agent_name not in all_agent_names:
                raise ValueError(f"智能體{agent_name}不存在于協作組中")
            # 檢查依賴的智能體是否存在
            for dep_name in dep_names:
                if dep_name not in all_agent_names:
                    raise ValueError(f"智能體{agent_name}依賴的{dep_name}不存在于協作組中")
    def _topological_sort(self) -> List[str]:
        """拓撲排序：根據依賴關系，生成智能體的運行順序"""
        # 1. 初始化入度（每個智能體依賴的數量）和鄰接表
        in_degree = {agent_name: 0 for agent_name in self.agents.keys()}
        adj = {agent_name: [] for agent_name in self.agents.keys()}


        # 2. 填充入度和鄰接表
        for agent_name, dep_names in self.dependencies.items():
            in_degree[agent_name] = len(dep_names)
            for dep_name in dep_names:
                adj[dep_name].append(agent_name)


        # 3. 拓撲排序（Kahn算法）
        from collections import deque
        queue = deque([name for name, cnt in in_degree.items() if cnt == 0])
        sorted_order = []


        while queue:
            current = queue.popleft()
            sorted_order.append(current)
            # 減少依賴當前智能體的入度
            for neighbor in adj[current]:
                in_degree[neighbor] -= 1
                if in_degree[neighbor] == 0:
                    queue.append(neighbor)


        # 4. 檢查是否有循環依賴（如果排序結果長度不等于智能體數量，說明有循環）
        if len(sorted_order) != len(self.agents):
            raise ValueError("智能體間存在循環依賴，無法生成運行順序")


        return sorted_order
    def run(
        self,
        initial_input: str,  # 整個協作組的初始輸入（如用戶的原始需求）
        skip_failed_agents: bool = False  # 智能體失敗時是否跳過（True=跳過，False=終止）
    ) -> Dict[str, Any]:
        """運行協作組：按拓撲排序執行智能體，傳遞輸入，收集結果"""
        # 1. 生成智能體運行順序
        try:
            run_order = self._topological_sort()
            print(f"=== 協作組運行順序 ===")
            for i, agent_name in enumerate(run_order, 1):
                print(f"{i}. {agent_name}")
        except ValueError as e:
            return {"status": "error", "result": f"協作組初始化失敗：{str(e)}"}


        # 2. 存儲每個智能體的輸出（用于傳遞給依賴它的智能體）
        agent_outputs: Dict[str, str] = {}
        # 存儲整體運行日志
        global_history: List[Dict[str, str]] = []


        # 3. 按順序運行每個智能體
        for agent_name in run_order:
            agent = self.agents[agent_name]
            print(f"\n=== 開始運行智能體：{agent_name} ===")


            # 3.1 準備輸入：如果有依賴，把依賴智能體的輸出加入輸入
            current_input = initial_input
            if agent_name in self.dependencies:
                dep_outputs = "\n".join([
                    f"【{dep_name}的輸出】：{agent_outputs[dep_name]}"
                    for dep_name in self.dependencies[agent_name]
                ])
                current_input = f"原始需求：{initial_input}\n\n依賴智能體的輸出：\n{dep_outputs}\n\n請基于以上信息，完成你的任務：{agent.role.split('：')[-1]}"


            # 3.2 運行智能體（支持重試）
            retry_count = 0
            agent_result = None
            while retry_count < self.max_retries:
                agent_result = agent.run(input_cnotallow=current_input)
                if agent_result["status"] == "success":
                    print(f"智能體{agent_name}運行成功")
                    break
                else:
                    retry_count += 1
                    print(f"智能體{agent_name}運行失敗（第{retry_count}次重試）：{agent_result['result']}")


            # 3.3 處理智能體最終結果
            if agent_result["status"] != "success":
                error_msg = f"智能體{agent_name}運行失敗（已重試{self.max_retries}次）：{agent_result['result']}"
                global_history.append({"type": "agent_failed", "content": error_msg})
                if not skip_failed_agents:
                    return {"status": "error", "result": error_msg, "global_history": global_history}
                else:
                    print(f"跳過失敗的智能體：{agent_name}")
                    agent_outputs[agent_name] = f"該智能體運行失敗，無輸出：{error_msg}"
            else:
                # 運行成功，存儲輸出和日志
                agent_outputs[agent_name] = agent_result["result"]
                global_history.append({
                    "type": "agent_success",
                    "content": f"智能體{agent_name}運行成功，輸出：{agent_result['result']}",
                    "agent_history": agent_result["history"]
                })


        # 4. 生成最終結果（匯總所有智能體輸出）
        final_result = "\n\n".join([
            f"【{agent_name}的輸出】：{output}"
            for agent_name, output in agent_outputs.items()
        ])


        return {
            "status": "success",
            "final_result": final_result,
            "agent_outputs": agent_outputs,
            "global_history": global_history,
            "run_order": run_order
        }
# 測試協作組：創建“調研+摘要”雙智能體協作
if __name__ == "__main__":
    # 1. 先創建所需工具
    ## 1.1 新聞搜索工具（復用前面實現的news_search_tool）
    ## 1.2 文本摘要工具（新創建，用于總結新聞）
    from pydantic import Field
    # 摘要工具的參數校驗
    class SummaryArgs(BaseModel):
        text: str = Field(..., descriptinotallow="需要摘要的文本，比如多條新聞的集合")
        summary_length: str = Field(default="中等", descriptinotallow="摘要長度：短（100字內）、中等（200-300字）、長（500字以上）")
    # 模擬文本摘要函數
    def mock_text_summary(text: str, summary_length: str) -> str:
        length_map = {"短": 100, "中等": 250, "長": 550}
        max_len = length_map.get(summary_length, 250)
        # 模擬摘要邏輯：提取關鍵詞+簡化句子
        summary = f"【{summary_length}摘要】：基于文本內容，核心信息包括：AI多智能體系統效率提升、金融場景應用、新聞來源為科技日報等。{text[:max_len-100]}..."
        return summary[:max_len]  # 確保不超過長度限制
    # 創建摘要工具
    summary_tool = Tool(
        name="text_summary",
        descriptinotallow="將輸入的長文本（如多條新聞）總結成指定長度的摘要，保留核心信息",
        func=mock_text_summary,
        args_schema=SummaryArgs
    )
    # 2. 創建兩個智能體
    ## 2.1 新聞調研智能體（復用前面的research_agent）
    research_agent = Agent(
        name="NewsResearchAgent",
        role="新聞調研智能體：負責根據用戶需求，調用新聞搜索工具收集最新相關新聞，整理成包含“標題、摘要、來源”的列表，不需要額外分析",
        tools=[news_search_tool]
    )
    ## 2.2 摘要智能體（新創建，依賴調研智能體的輸出）
    summary_agent = Agent(
        name="TextSummaryAgent",
        role="文本摘要智能體：負責將輸入的新聞列表總結成“中等長度”的摘要，突出核心進展和關鍵信息，不需要保留原始列表格式",
        tools=[summary_tool]
    )
    # 3. 創建協作組：摘要智能體依賴調研智能體
    crew = Crew(
        agents=[research_agent, summary_agent],
        dependencies={"TextSummaryAgent": ["NewsResearchAgent"]},  # 摘要智能體依賴調研智能體
        max_retries=1
    )
    # 4. 運行協作組：初始需求是“收集AI多智能體的最新新聞并總結”
    crew_result = crew.run(
        initial_input="收集2025年1月1日至2025年1月10日關于‘AI多智能體’的3條最新新聞，并將這些新聞總結成中等長度的摘要",
        skip_failed_agents=False
    )
    # 5. 輸出最終結果
    print("\n=== 協作組最終結果 ===")
    if crew_result["status"] == "success":
        print(crew_result["final_result"])
    else:
        print(f"協作組運行失敗：{crew_result['result']}")

9. 完整實戰：多智能體系統運行效果

我們以 “收集 AI 多智能體新聞并總結” 為例，看看整個系統的運行流程和輸出結果：

9.1 運行流程（按協作組順序）

第一步：運行 NewsResearchAgent（調研智能體）

輸入：用戶需求 “收集 2025 年 1 月 1 日至 2025 年 1 月 10 日關于‘AI 多智能體’的 3 條最新新聞”；
思考過程：需要調用??news_search??? 工具，參數設為??keyword="AI多智能體"???、??time_range="2025-01-01至2025-01-10"???、??max_results=3??；
工具輸出：3 條模擬新聞（包含標題、摘要、來源）；
智能體輸出：整理成列表格式的新聞匯總。

第二步：運行 TextSummaryAgent（摘要智能體）

輸入：原始需求 + 調研智能體的新聞列表；
思考過程：需要調用??text_summary??? 工具，參數設為??text=調研智能體輸出???、??summary_length="中等"??；
工具輸出：中等長度的新聞摘要；
智能體輸出：結構化的摘要文本。

9.2 最終輸出示例

=== 協作組最終結果 ===
【NewsResearchAgent的輸出】：
1. 標題：AI多智能體最新進展：多智能體協作效率提升30%，摘要：近日研究顯示，多智能體系統在復雜任務處理中，效率比單智能體高30%，已應用于金融分析場景，來源：科技日報，發布時間：2025-01-10
2. 標題：AI多智能體最新進展：多智能體協作效率提升30%，摘要：近日研究顯示，多智能體系統在復雜任務處理中，效率比單智能體高30%，已應用于金融分析場景，來源：科技日報，發布時間：2025-01-10
3. 標題：AI多智能體最新進展：多智能體協作效率提升30%，摘要：近日研究顯示，多智能體系統在復雜任務處理中，效率比單智能體高30%，已應用于金融分析場景，來源：科技日報，發布時間：2025-01-10


【TextSummaryAgent的輸出】：
【中等摘要】：基于2025年1月1日至1月10日的3條AI多智能體相關新聞，核心信息如下：
1. 多智能體系統效率顯著提升：研究表明其在復雜任務處理中效率比單智能體高30%；
2. 應用場景落地：該技術已在金融分析領域實際應用；
3. 信息來源：所有新聞均來自科技日報，發布時間集中在1月10日，信息時效性較強。
整體來看，AI多智能體技術正從理論走向實踐，效率優勢成為核心競爭力，金融領域或成首批規模化應用場景。

10. 擴展與優化建議

目前實現的多智能體系統是基礎版本，實際項目中可從以下方向擴展：

10.1 支持并行運行

當前 Crew 類是 “順序執行”，可優化為 “支持并行”—— 比如多個無依賴的智能體（如 “查天氣” 和 “找酒店”）可同時運行，提升效率。可基于 Python 的 ??concurrent.futures?? 實現并行邏輯。

10.2 增加動態任務分配

讓協調器（Crew）具備 “根據前一個智能體的輸出，動態選擇下一個智能體” 的能力 —— 比如調研智能體發現 “新聞涉及金融場景”，就自動調用 “金融分析智能體”，而不是固定順序。

10.3 強化日志與監控

增加更詳細的日志（如工具調用耗時、智能體思考耗時），甚至對接監控工具（如 Prometheus），方便追蹤系統性能瓶頸。

10.4 支持本地模型

將 OpenAI 替換為 Ollama + Llama3 或 Qwen（阿里通義千問開源版），實現 “完全本地部署”，適合對數據隱私敏感的場景（修改 Agent 類的 ??generate_thought?? 方法即可）。

10.5 增加安全防護

工具調用權限控制：給不同智能體分配不同權限（如 “調研智能體” 只能讀工具，不能調用 “發郵件” 等寫工具）；
輸入輸出過濾：防止智能體接收惡意輸入（如攻擊指令）或輸出違規內容（如敏感信息）。

11. 總結

本文從零實現了一個多智能體系統，核心是三個模塊：

Tool 類：封裝外部功能，讓智能體能安全、規范地調用；
Agent 類：實現 ReAct 循環，讓智能體 “會思考、能行動”，且聚焦自身角色；
Crew 類：通過拓撲排序管理依賴，讓多智能體 “有序協作”。

這個系統的優勢在于模塊化強（修改一個智能體不影響整體）、可解釋性高（每個步驟都有日志）、易擴展（新增工具或智能體只需繼承基礎類）。

無論是處理 “復雜任務拆解”“多場景協作”，還是 “提升 AI 系統可靠性”，多智能體模式都比單智能體更有優勢。后續可基于這個基礎框架，根據實際需求（如金融分析、醫療診斷、教育輔導）擴展專屬的智能體和工具，實現更復雜的 AI 應用。

好了，這就是我今天想分享的內容。

本文轉載自??玄姐聊AGI?? 作者：玄姐

?著作權歸作者所有，如需轉載，請注明出處，否則將追究法律責任

標簽

多智能體

案例架構設計

Agent

已于2025-10-10 06:42:19修改

贊

回復

舉報

回復

51CTO

51CTO博客

51CTO學堂

萬字長文剖析企業級多智能體案例架構設計與代碼級落地實踐 原創

1. 前言

2. 為什么需要多智能體？

2.1 開發與調試更簡單

2.2 降低復雜性，減少 “幻覺”

2.3 問題解決更高效

2.4 推理過程更透明

2.5 問題定位更快速

2.6 符合人類團隊協作邏輯

3. 多智能體的內部原理

4. 系統整體架構

4.1 智能體（Agent）

4.2 工具（Tool）

4.3 協作組（Crew）

4.4 信息流轉方式

5. 從零實現多智能體系統

5.1 工具類（Tool）

5.2 智能體類（Agent）

5.3 協作組類（Crew）

6. 詳細實現：工具類（Tool）

6.1 設計核心：讓智能體 “清晰認知工具”

6.2 代碼實現：基礎 Tool 類

7. 詳細實現：智能體類（Agent）

7.1 設計核心：讓智能體 “有角色、會思考、能行動”

7.2 代碼實現：基礎 Agent 類

8. 詳細實現：協作組類（Crew）

8.1 設計核心：讓多智能體 “有序協作”

8.2 代碼實現：基礎 Crew 類

9. 完整實戰：多智能體系統運行效果

9.1 運行流程（按協作組順序）

9.2 最終輸出示例

10. 擴展與優化建議

10.1 支持并行運行

10.2 增加動態任務分配

10.3 強化日志與監控

10.4 支持本地模型

10.5 增加安全防護

11. 總結

目錄

萬字長文剖析企業級多智能體案例架構設計與代碼級落地實踐原創