OpenManus源碼解析:智能體框架的設(shè)計(jì)與實(shí)現(xiàn)

一、項(xiàng)目概述

1.1 OpenManus的定位與設(shè)計(jì)目標(biāo)

OpenManus是一個(gè)基于大語(yǔ)言模型(LLM)的智能體框架,它的設(shè)計(jì)目標(biāo)是創(chuàng)建一個(gè)靈活、可擴(kuò)展且功能強(qiáng)大的系統(tǒng),使AI能夠通過(guò)各種工具與外部世界交互,從而解決復(fù)雜的任務(wù)。

與傳統(tǒng)的聊天機(jī)器人不同,OpenManus不僅能夠理解和生成文本,還能夠執(zhí)行具體的操作,如搜索信息、瀏覽網(wǎng)頁(yè)、執(zhí)行代碼和保存文件等。這種能力使其成為一個(gè)真正的"智能助手",而不僅僅是一個(gè)對(duì)話系統(tǒng)。

OpenManus的核心理念是"思考-行動(dòng)"循環(huán),即智能體先分析當(dāng)前狀態(tài)和任務(wù)需求(思考),然后選擇并執(zhí)行適當(dāng)?shù)墓ぞ撸ㄐ袆?dòng)),接著基于執(zhí)行結(jié)果進(jìn)行下一輪思考。這種循環(huán)使智能體能夠逐步解決復(fù)雜問(wèn)題,同時(shí)保持對(duì)任務(wù)的連貫理解。

1.2 項(xiàng)目結(jié)構(gòu)概覽

OpenManus的項(xiàng)目結(jié)構(gòu)清晰而模塊化,主要包括以下幾個(gè)部分:

app/
├── agent/                # **智能體實(shí)現(xiàn)
│   ├── base.py           # **基礎(chǔ)智能體
│   ├── react.py          # **思考-行動(dòng)智能體
│   ├── toolcall.py       # **工具調(diào)用智能體
│   └── manus.py          # **Manus智能體
├── tool/                 # **工具實(shí)現(xiàn)
│   ├── base.py           # **基礎(chǔ)工具
│   ├── bash.py           # **命令行工具
│   ├── browser_use_tool.py # **瀏覽器工具
│   ├── file_saver.py     # **文件保存工具
│   ├── python_execute.py # **Python執(zhí)行工具
│   ├── terminate.py      # **終止工具
│   └── tool_collection.py # **工具集合
├── flow/                 # **流程控制
│   ├── base.py           # **基礎(chǔ)流程
│   ├── planning.py       # **規(guī)劃流程
│   └── flow_factory.py   # **流程工廠
├── prompt/               # **提示模板
│   └── manus.py          # **Manus提示
├── llm.py                # **LLM接口
├── memory.py             # **記憶系統(tǒng)
└── message.py            # **消息定義
main.py                   # **主入口

這種結(jié)構(gòu)使得各個(gè)組件之間的職責(zé)劃分清晰,便于維護(hù)和擴(kuò)展。

1.3 核心組件介紹

1) 智能體系統(tǒng)

智能體系統(tǒng)是OpenManus的核心,它采用了層次化的設(shè)計(jì):

  • BaseAgent:提供基本的狀態(tài)管理和執(zhí)行循環(huán)
  • ReActAgent:實(shí)現(xiàn)思考-行動(dòng)循環(huán)模式
  • ToolCallAgent:實(shí)現(xiàn)工具調(diào)用機(jī)制
  • Manus:集成多種工具的具體智能體實(shí)現(xiàn)

這種層次化設(shè)計(jì)使得代碼更加模塊化和可擴(kuò)展,每個(gè)層次只需關(guān)注自己的職責(zé)。

2) 工具系統(tǒng)

工具系統(tǒng)為智能體提供了與外部世界交互的能力:

  • BaseTool:所有工具的抽象基類(lèi)
  • ToolCollection:工具的集合和管理器
  • 具體工具:如PythonExecute、GoogleSearch、BrowserUseTool等

每個(gè)工具都有明確的名稱(chēng)、描述和參數(shù)規(guī)范,使LLM能夠正確選擇和使用它們。

3) 記憶系統(tǒng)

記憶系統(tǒng)使智能體能夠在多個(gè)步驟中保持上下文連貫性:

  • Memory:存儲(chǔ)交互歷史的容器
  • Message:表示不同類(lèi)型消息的結(jié)構(gòu)

記憶系統(tǒng)記錄了用戶輸入、LLM響應(yīng)和工具執(zhí)行結(jié)果,使智能體能夠基于歷史信息做出決策。

4) LLM接口

LLM接口負(fù)責(zé)與大語(yǔ)言模型(如OpenAI的GPT模型)通信:

  • LLM:封裝了與LLM API的交互
  • ToolResponse:表示LLM響應(yīng)的結(jié)構(gòu)

LLM接口將智能體的記憶和工具信息傳遞給LLM,并解析LLM的響應(yīng)。

5) 流程控制

流程控制組件管理不同類(lèi)型的執(zhí)行流程:

  • BaseFlow:所有流程的抽象基類(lèi)
  • PlanningFlow:實(shí)現(xiàn)規(guī)劃和執(zhí)行的流程
  • FlowFactory:創(chuàng)建不同類(lèi)型流程的工廠

流程控制使OpenManus能夠支持不同的執(zhí)行模式,如規(guī)劃式執(zhí)行。

1.4 技術(shù)特點(diǎn)

OpenManus具有幾個(gè)顯著的技術(shù)特點(diǎn):

  1. 異步編程:廣泛使用async/await進(jìn)行異步操作,提高I/O效率
  2. 模塊化設(shè)計(jì):清晰的組件劃分和接口定義,便于維護(hù)和擴(kuò)展
  3. 錯(cuò)誤處理:多層次的錯(cuò)誤捕獲和恢復(fù)機(jī)制,提高系統(tǒng)穩(wěn)定性
  4. 工具抽象:統(tǒng)一的工具接口,便于添加新工具
  5. 記憶管理:完善的記憶系統(tǒng),支持上下文連貫的多步驟任務(wù)

這些特點(diǎn)使OpenManus成為一個(gè)強(qiáng)大而靈活的智能體框架,能夠應(yīng)對(duì)各種復(fù)雜任務(wù)。

通過(guò)這個(gè)項(xiàng)目,可以看到AI智能體如何從簡(jiǎn)單的對(duì)話系統(tǒng)演變?yōu)槟軌驁?zhí)行具體操作的助手,這代表了AI應(yīng)用的一個(gè)重要發(fā)展方向。在接下來(lái)的章節(jié)中,下文將深入探討OpenManus的各個(gè)組件和機(jī)制,揭示其內(nèi)部工作原理。


二、智能體的層次化設(shè)計(jì)

2.1 BaseAgent:基礎(chǔ)智能體的實(shí)現(xiàn)

BaseAgent是所有智能體的基類(lèi),位于app/agent/base.py中。它提供了智能體的基本功能:

核心屬性

  • name:智能體的名稱(chēng)
  • description:智能體的描述
  • system_prompt:系統(tǒng)級(jí)指令提示
  • next_step_prompt:決定下一步行動(dòng)的提示
  • llm:語(yǔ)言模型實(shí)例
  • memory:智能體的記憶存儲(chǔ)
  • state:當(dāng)前智能體狀態(tài)(IDLE、RUNNING、FINISHED、ERROR)

主要方法

  • run(request):執(zhí)行智能體的主循環(huán)
  • step():執(zhí)行單個(gè)步驟(抽象方法,需要子類(lèi)實(shí)現(xiàn))
  • update_memory():更新智能體的記憶
  • is_stuck():檢測(cè)智能體是否陷入循環(huán)
  • handle_stuck_state():處理卡住狀態(tài)

BaseAgent的run方法是智能體執(zhí)行的核心,它實(shí)現(xiàn)了一個(gè)循環(huán),在循環(huán)中不斷調(diào)用step方法,直到任務(wù)完成或達(dá)到最大步驟數(shù):

async def run(self, request: Optional[str] = None) -> str:
    if request:
        self.update_memory("user", request)

    results: List[str] = []
    async with self.state_context(AgentState.RUNNING):
        while (
            self.current_step < self.max_steps and self.state != AgentState.FINISHED
        ):
            self.current_step += 1
            step_result = await self.step()
            
            # **檢查是否陷入循環(huán)
            if self.is_stuck():
                self.handle_stuck_state()
                
            results.append(f"Step {self.current_step}: {step_result}")

2.2 ReActAgent:思考-行動(dòng)循環(huán)模式

ReActAgent繼承自BaseAgent,位于app/agent/react.py中。它實(shí)現(xiàn)了思考-行動(dòng)循環(huán)模式,這是一種強(qiáng)大的智能體決策框架。

核心方法

  • think():處理當(dāng)前狀態(tài)并決定下一步行動(dòng)(抽象方法)
  • act():執(zhí)行決定的行動(dòng)(抽象方法)
  • step():執(zhí)行單個(gè)步驟(實(shí)現(xiàn)了BaseAgent的抽象方法)

ReActAgent的step方法實(shí)現(xiàn)了思考-行動(dòng)循環(huán):

async def step(self) -> str:
    """執(zhí)行單個(gè)步驟:思考和行動(dòng)。"""
    should_act = await self.think()  # **先思考
    if not should_act:
        return "思考完成 - 無(wú)需行動(dòng)"
    return await self.act()  # **再行動(dòng)

這種思考-行動(dòng)模式非常適合智能體的決策過(guò)程,它模擬了人類(lèi)的思考方式:先分析情況,再采取行動(dòng)。

2.3 ToolCallAgent:工具調(diào)用機(jī)制

ToolCallAgent繼承自ReActAgent,位于app/agent/toolcall.py中。它實(shí)現(xiàn)了工具調(diào)用機(jī)制,使智能體能夠使用各種工具來(lái)完成任務(wù)。

核心屬性

  • available_tools:可用工具集合
  • tool_choices:工具選擇模式("none"、"auto"、"required")
  • special_tool_names:特殊工具名稱(chēng)列表
  • tool_calls:工具調(diào)用列表

主要方法

  • think():實(shí)現(xiàn)了ReActAgent的抽象方法,使用LLM決定使用哪些工具
  • act():實(shí)現(xiàn)了ReActAgent的抽象方法,執(zhí)行工具調(diào)用
  • execute_tool(command):執(zhí)行單個(gè)工具調(diào)用
  • _handle_special_tool(name, result):處理特殊工具執(zhí)行和狀態(tài)變化

ToolCallAgent的think方法使用LLM來(lái)決定使用哪些工具:

async def think(self) -> bool:
    if self.next_step_prompt:
        user_msg = Message.user_message(self.next_step_prompt)
        self.messages += [user_msg]

    # **獲取帶工具選項(xiàng)的響應(yīng)
    response = await self.llm.ask_tool(
        messages=self.messages,
        system_msgs=[Message.system_message(self.system_prompt)]
        if self.system_prompt
        else None,
        tools=self.available_tools.to_params(),
        tool_choice=self.tool_choices,
    )
    self.tool_calls = response.tool_calls
    
    # **記錄響應(yīng)信息
    logger.info(f"? {self.name}'s thoughts: {response.content}")
    logger.info(
        f"??? {self.name} selected {len(response.tool_calls) if response.tool_calls else 0} tools to use"
    )

ToolCallAgent的act方法執(zhí)行工具調(diào)用并處理結(jié)果:

async def act(self) -> str:
    if not self.tool_calls:
        if self.tool_choices == "required":
            raise ValueError(TOOL_CALL_REQUIRED)
        return self.messages[-1].content or "No content or commands to execute"

    results = []
    for command in self.tool_calls:
        result = await self.execute_tool(command)
        logger.info(
            f"?? Tool '{command.function.name}' completed its mission! Result: {result}"
        )

        # **將工具響應(yīng)添加到記憶中
        tool_msg = Message.tool_message(
            content=result, tool_call_id=command.id, name=command.function.name
        )
        self.memory.add_message(tool_msg)
        results.append(result)

    return "\n\n".join(results)

2.4 Manus:最終智能體的集成

Manus繼承自ToolCallAgent,位于app/agent/manus.py中。它是用戶直接交互的主要智能體,集成了多種工具。

核心屬性

  • name:"Manus"
  • description:"一個(gè)可以使用多種工具解決各種任務(wù)的多功能智能體"
  • system_prompt:來(lái)自app/prompt/manus.py的系統(tǒng)提示
  • next_step_prompt:來(lái)自app/prompt/manus.py的下一步提示
  • available_tools:包含PythonExecute、GoogleSearch、BrowserUseTool、FileSaver和Terminate的工具集合
  • max_steps:20(最大步驟數(shù))
class Manus(ToolCallAgent):
    name: str = "Manus"
    description: str = (
        "A versatile agent that can solve various tasks using multiple tools"
    )

    system_prompt: str = SYSTEM_PROMPT
    next_step_prompt: str = NEXT_STEP_PROMPT

    # **添加通用工具到工具集合
    available_tools: ToolCollection = Field(
        default_factory=lambda: ToolCollection(
            PythonExecute(), GoogleSearch(), BrowserUseTool(), FileSaver(), Terminate()
        )
    )

    max_steps: int = 20

2.5 總結(jié)

層次結(jié)構(gòu)的優(yōu)勢(shì)

  1. 代碼重用:每個(gè)層次只需實(shí)現(xiàn)自己特有的功能,其他功能可以從父類(lèi)繼承。
  2. 關(guān)注點(diǎn)分離
    • BaseAgent處理基本的狀態(tài)管理和執(zhí)行循環(huán)
    • ReActAgent實(shí)現(xiàn)思考-行動(dòng)模式
    • ToolCallAgent處理工具調(diào)用
    • Manus集成特定工具和提示
  3. 靈活性和可擴(kuò)展性:可以通過(guò)繼承現(xiàn)有智能體類(lèi)來(lái)創(chuàng)建新的智能體類(lèi)型,而不需要修改現(xiàn)有代碼。
  4. 維護(hù)性:每個(gè)層次的代碼都相對(duì)簡(jiǎn)單和專(zhuān)注,使得代碼更容易理解和維護(hù)。
  5. 測(cè)試性:可以獨(dú)立測(cè)試每個(gè)層次的功能,簡(jiǎn)化測(cè)試過(guò)程。

智能體執(zhí)行流程

當(dāng)用戶輸入一個(gè)請(qǐng)求時(shí),執(zhí)行流程如下:

  1. 請(qǐng)求被傳遞給Manus智能體的run方法(繼承自BaseAgent)
  2. run方法進(jìn)入一個(gè)循環(huán),重復(fù)調(diào)用step方法(繼承自ReActAgent)
  3. step方法首先調(diào)用think方法(ToolCallAgent實(shí)現(xiàn))來(lái)決定使用哪些工具
  4. 然后調(diào)用act方法(ToolCallAgent實(shí)現(xiàn))來(lái)執(zhí)行工具調(diào)用
  5. 工具執(zhí)行結(jié)果被添加到記憶中,用于后續(xù)決策
  6. 重復(fù)這個(gè)過(guò)程,直到任務(wù)完成或達(dá)到最大步驟數(shù)

這種層次化設(shè)計(jì)使得OpenManus項(xiàng)目能夠以一種模塊化、可維護(hù)的方式實(shí)現(xiàn)復(fù)雜的智能體行為。


三、工具系統(tǒng)的實(shí)現(xiàn)

OpenManus中的工具是經(jīng)過(guò)精心設(shè)計(jì)的軟件組件,它們主要是規(guī)范了輸入輸出的功能模塊,而不是完全智能化的應(yīng)用。下文來(lái)詳細(xì)解析這些工具的實(shí)現(xiàn)機(jī)制。

3.1 BaseTool:工具的基礎(chǔ)抽象

所有工具都繼承自BaseTool抽象基類(lèi),這個(gè)基類(lèi)定義了工具的基本結(jié)構(gòu):

class BaseTool(ABC, BaseModel):
    name: str
    description: str
    parameters: Optional[dict] = None

    async def __call__(self, **kwargs) -> Any:
        """執(zhí)行工具"""
        return await self.execute(**kwargs)

    @abstractmethod
    async def execute(self, **kwargs) -> Any:
        """執(zhí)行工具的具體邏輯"""
        pass

    def to_param(self) -> Dict:
        """轉(zhuǎn)換為函數(shù)調(diào)用格式"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            },
        }

這個(gè)架構(gòu)確保了所有工具都有統(tǒng)一的接口,方便智能體調(diào)用和管理。

3.2 主要工具詳解

PythonExecute:代碼執(zhí)行工具

PythonExecute工具允許執(zhí)行Python代碼:

class PythonExecute(BaseTool):
    name: str = "python_execute"
    description: str = "執(zhí)行Python代碼并返回結(jié)果"
    parameters: dict = {
        "type": "object",
        "properties": {
            "code": {
                "type": "string",
                "description": "要執(zhí)行的Python代碼"
            },
            "timeout": {
                "type": "integer",
                "description": "執(zhí)行超時(shí)時(shí)間(秒)"
            }
        },
        "required": ["code"]
    }

    async def execute(self, code: str, timeout: int = 5) -> Dict:
        """執(zhí)行Python代碼"""
        try:
            # **創(chuàng)建一個(gè)安全的執(zhí)行環(huán)境
            locals_dict = {}
            
            # **使用asyncio.wait_for實(shí)現(xiàn)超時(shí)控制
            await asyncio.wait_for(
                self._execute_code(code, locals_dict),
                timeout=timeout
            )
            
            # **提取執(zhí)行結(jié)果
            result = locals_dict.get("result", None)
            return {"result": result}
        except asyncio.TimeoutError:
            return {"error": f"代碼執(zhí)行超時(shí)({timeout}秒)"}
        except Exception as e:
            return {"error": str(e)}
    
    async def _execute_code(self, code: str, locals_dict: Dict):
        """在隔離環(huán)境中執(zhí)行代碼"""
        # **添加一些安全限制
        restricted_globals = {
            "__builtins__": {
                name: getattr(__builtins__, name)
                for name in ["print", "range", "len", "dict", "list", "set", "int", "float", "str"]
            }
        }
        
        # **執(zhí)行代碼
        exec(code, restricted_globals, locals_dict)

這個(gè)工具主要是一個(gè)代碼執(zhí)行器,它創(chuàng)建了一個(gè)受限的執(zhí)行環(huán)境,并使用Python的exec函數(shù)來(lái)執(zhí)行代碼。它不是智能化的應(yīng)用,而是一個(gè)規(guī)范了輸入輸出的功能模塊。

GoogleSearch:搜索工具

GoogleSearch工具允許執(zhí)行網(wǎng)絡(luò)搜索:

class GoogleSearch(BaseTool):
    name: str = "google_search"
    description: str = "使用Google搜索信息"
    parameters: dict = {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索查詢"
            },
            "num_results": {
                "type": "integer",
                "description": "返回結(jié)果數(shù)量"
            }
        },
        "required": ["query"]
    }
    
    search_client: GoogleSearchClient = Field(default_factory=GoogleSearchClient)

    async def execute(self, query: str, num_results: int = 10) -> List[str]:
        """執(zhí)行Google搜索"""
        try:
            # **調(diào)用搜索客戶端
            search_results = await self.search_client.search(query, num_results=num_results)
            
            # **格式化結(jié)果
            formatted_results = []
            for i, result in enumerate(search_results, 1):
                title = result.get("title", "無(wú)標(biāo)題")
                link = result.get("link", "無(wú)鏈接")
                snippet = result.get("snippet", "無(wú)摘要")
                formatted_results.append(f"{i}. {title}\n   URL: {link}\n   {snippet}\n")
            
            return "\n".join(formatted_results)
        except Exception as e:
            return f"搜索出錯(cuò): {str(e)}"

這個(gè)工具是一個(gè)搜索接口封裝,它調(diào)用GoogleSearchClient來(lái)執(zhí)行實(shí)際的搜索操作。GoogleSearchClient可能是一個(gè)API客戶端,用于調(diào)用Google搜索API或其他搜索服務(wù)。

BrowserUseTool:瀏覽器控制工具

BrowserUseTool工具允許控制瀏覽器:

class BrowserUseTool(BaseTool):
    name: str = "browser_use"
    description: str = "控制瀏覽器執(zhí)行各種操作"
    parameters: dict = {
        "type": "object",
        "properties": {
            "action": {
                "type": "string",
                "enum": [
                    "navigate", "click", "input_text", "screenshot",
                    "get_html", "get_text", "execute_js", "scroll",
                    "switch_tab", "new_tab", "close_tab", "refresh"
                ],
                "description": "瀏覽器操作類(lèi)型"
            },
            "url": {
                "type": "string",
                "description": "用于'navigate'或'new_tab'操作的URL"
            },
            # **其他參數(shù)...
        },
        "required": ["action"]
    }
    
    browser: Optional[BrowserUseBrowser] = Field(default=None, exclude=True)
    
    async def execute(self, action: str, url: Optional[str] = None, ...) -> ToolResult:
        """執(zhí)行瀏覽器操作"""
        async with self.lock:
            try:
                # **確保瀏覽器已初始化
                context = await self._ensure_browser_initialized()
                
                # **根據(jù)操作類(lèi)型執(zhí)行不同的瀏覽器操作
                if action == "navigate":
                    if not url:
                        return ToolResult(error="URL is required for 'navigate' action")
                    await context.navigate(url)
                    return ToolResult(output=f"Navigated to {url}")
                
                elif action == "click":
                    # **點(diǎn)擊操作實(shí)現(xiàn)...
                
                # **其他操作實(shí)現(xiàn)...
                
            except Exception as e:
                return ToolResult(error=f"Browser action '{action}' failed: {str(e)}")

這個(gè)工具是一個(gè)瀏覽器自動(dòng)化接口,它使用BrowserUseBrowser類(lèi)(可能基于Playwright或Selenium)來(lái)控制瀏覽器。它提供了一組標(biāo)準(zhǔn)化的操作(如導(dǎo)航、點(diǎn)擊、輸入文本等),但本身并不包含智能化的邏輯。

FileSaver:文件保存工具

FileSaver工具允許保存內(nèi)容到文件:

class FileSaver(BaseTool):
    name: str = "file_saver"
    description: str = "保存內(nèi)容到本地文件"
    parameters: dict = {
        "type": "object",
        "properties": {
            "content": {
                "type": "string",
                "description": "要保存的內(nèi)容"
            },
            "file_path": {
                "type": "string",
                "description": "文件保存路徑"
            },
            "mode": {
                "type": "string",
                "description": "文件打開(kāi)模式",
                "enum": ["w", "a"],
                "default": "w"
            }
        },
        "required": ["content", "file_path"]
    }

    async def execute(self, content: str, file_path: str, mode: str = "w") -> str:
        """保存內(nèi)容到文件"""
        try:
            # **確保目錄存在
            directory = os.path.dirname(file_path)
            if directory and not os.path.exists(directory):
                os.makedirs(directory)

            # **寫(xiě)入文件
            async with aiofiles.open(file_path, mode, encoding="utf-8") as file:
                await file.write(content)

            return f"內(nèi)容已成功保存到 {file_path}"
        except Exception as e:
            return f"保存文件出錯(cuò): {str(e)}"

這個(gè)工具是一個(gè)簡(jiǎn)單的文件操作接口,它使用aiofiles庫(kù)(異步文件I/O)來(lái)保存內(nèi)容到文件。它是一個(gè)非常基礎(chǔ)的功能模塊,沒(méi)有智能化的邏輯。

Terminate:終止工具

Terminate工具用于結(jié)束智能體的執(zhí)行:

class Terminate(BaseTool):
    name: str = "terminate"
    description: str = "當(dāng)任務(wù)完成或無(wú)法繼續(xù)時(shí)終止交互"
    parameters: dict = {
        "type": "object",
        "properties": {
            "status": {
                "type": "string",
                "description": "交互的完成狀態(tài)",
                "enum": ["success", "failure"]
            }
        },
        "required": ["status"]
    }

    async def execute(self, status: str) -> str:
        """終止當(dāng)前執(zhí)行"""
        return f"交互已完成,狀態(tài): {status}"

這個(gè)工具非常簡(jiǎn)單,它只是返回一個(gè)狀態(tài)消息。實(shí)際的終止邏輯是在ToolCallAgent_handle_special_tool方法中實(shí)現(xiàn)的,它會(huì)將智能體的狀態(tài)設(shè)置為FINISHED。

3.3 工具與智能體的協(xié)作模式

在OpenManus框架中,工具與智能體之間形成了一種獨(dú)特而高效的協(xié)作模式,這種模式可以概括為"智能體決策,工具執(zhí)行"。這種協(xié)作充分發(fā)揮了大語(yǔ)言模型的推理能力和專(zhuān)用工具的執(zhí)行能力,形成了一個(gè)強(qiáng)大的組合。

決策與執(zhí)行的分離

OpenManus中最核心的協(xié)作理念是將決策與執(zhí)行明確分離:

  • 智能體負(fù)責(zé)決策:利用LLM的強(qiáng)大理解和推理能力,分析任務(wù)需求,選擇合適的工具,確定工具參數(shù),解釋工具執(zhí)行結(jié)果。
  • 工具負(fù)責(zé)執(zhí)行:接收智能體提供的參數(shù),執(zhí)行特定功能,返回執(zhí)行結(jié)果,不參與決策過(guò)程。
    這種分離使系統(tǒng)既有LLM的靈活性和創(chuàng)造性,又有專(zhuān)用工具的可靠性和確定性。

協(xié)作流程

工具與智能體的協(xié)作流程可以分為以下幾個(gè)步驟:

  1. 任務(wù)分析:智能體分析用戶請(qǐng)求,理解任務(wù)需求。
  2. 工具選擇:智能體從可用工具列表中選擇合適的工具。
  3. 參數(shù)準(zhǔn)備:智能體確定工具所需的參數(shù)。
  4. 工具調(diào)用:智能體通過(guò)工具調(diào)用機(jī)制執(zhí)行工具。
  5. 結(jié)果處理:智能體接收工具執(zhí)行結(jié)果,進(jìn)行分析和解釋。
  6. 下一步?jīng)Q策:智能體根據(jù)結(jié)果決定下一步行動(dòng)。

接口標(biāo)準(zhǔn)化

為了使協(xié)作順暢,OpenManus對(duì)工具接口進(jìn)行了標(biāo)準(zhǔn)化:

  • 統(tǒng)一的工具描述:每個(gè)工具都有名稱(chēng)、描述和參數(shù)規(guī)范,使LLM能夠理解工具的功能和用法。
  • 統(tǒng)一的調(diào)用方式:所有工具都通過(guò)execute方法調(diào)用,參數(shù)通過(guò)關(guān)鍵字參數(shù)傳遞。
  • 統(tǒng)一的結(jié)果格式:工具執(zhí)行結(jié)果統(tǒng)一為字符串或特定的結(jié)果對(duì)象,便于智能體處理。

這種標(biāo)準(zhǔn)化使得添加新工具變得簡(jiǎn)單,只需實(shí)現(xiàn)標(biāo)準(zhǔn)接口即可。

錯(cuò)誤處理與恢復(fù)

協(xié)作模式中的一個(gè)重要方面是錯(cuò)誤處理與恢復(fù):

  • 工具執(zhí)行錯(cuò)誤:工具捕獲執(zhí)行過(guò)程中的異常,返回錯(cuò)誤信息而不是拋出異常。
  • 智能體處理錯(cuò)誤:智能體接收錯(cuò)誤信息,分析原因,嘗試其他方法或工具。
  • 多次嘗試:如果一種方法失敗,智能體可以嘗試其他方法,直到任務(wù)完成或達(dá)到最大嘗試次數(shù)。

這種錯(cuò)誤處理機(jī)制使得系統(tǒng)在面對(duì)各種異常情況時(shí)仍能保持穩(wěn)定運(yùn)行。


四、工具選擇與使用的機(jī)制

4.1 工具選擇的實(shí)現(xiàn)

在OpenManus中,工具選擇主要在ToolCallAgent類(lèi)的think方法中實(shí)現(xiàn)。

工具選擇的核心代碼

ToolCallAgentthink方法是工具選擇的核心:

async def think(self) -> bool:
    # **添加下一步提示到消息列表
    if self.next_step_prompt:
        user_msg = Message.user_message(self.next_step_prompt)
        self.messages += [user_msg]

    # **調(diào)用LLM獲取帶工具選擇的響應(yīng)
    response = await self.llm.ask_tool(
        messages=self.messages,
        system_msgs=[Message.system_message(self.system_prompt)]
        if self.system_prompt
        else None,
        tools=self.available_tools.to_params(),
        tool_choice=self.tool_choices,
    )
    
    # **保存工具調(diào)用信息
    self.tool_calls = response.tool_calls
    
    # **記錄思考過(guò)程
    logger.info(f"? {self.name}'s thoughts: {response.content}")
    logger.info(
        f"??? {self.name} selected {len(response.tool_calls) if response.tool_calls else 0} tools to use"
    )
    
    # **將LLM響應(yīng)添加到記憶中
    self.memory.add_message(Message.assistant_message(response.content, response.tool_calls))
    
    # **如果有工具調(diào)用,返回True表示需要執(zhí)行行動(dòng)
    return bool(self.tool_calls)

工具選擇的關(guān)鍵步驟

  • 1) 準(zhǔn)備工具列表

首先,智能體需要準(zhǔn)備可用工具的列表。這在Manus類(lèi)中通過(guò)available_tools屬性設(shè)置:

available_tools: ToolCollection = Field(
    default_factory=lambda: ToolCollection(
        PythonExecute(), GoogleSearch(), BrowserUseTool(), FileSaver(), Terminate()
    )
)

每個(gè)工具都有一個(gè)to_param方法,將工具轉(zhuǎn)換為L(zhǎng)LM可以理解的函數(shù)調(diào)用格式:

def to_param(self) -> Dict:
    """Convert tool to function call format."""
    return {
        "type": "function",
        "function": {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters,
        },
    }
  • 2) 調(diào)用LLM進(jìn)行工具選擇
    關(guān)鍵步驟是調(diào)用llm.ask_tool方法,這個(gè)方法會(huì)將所有可用工具的信息傳遞給LLM,讓LLM決定使用哪些工具:
response = await self.llm.ask_tool(
    messages=self.messages,
    system_msgs=[Message.system_message(self.system_prompt)],
    tools=self.available_tools.to_params(),
    tool_choice=self.tool_choices,
)
  • 3) LLM如何選擇工具
    LLM類(lèi)的ask_tool方法中,會(huì)構(gòu)建一個(gè)請(qǐng)求發(fā)送給OpenAI API:
async def ask_tool(
    self,
    messages: List[Message],
    system_msgs: Optional[List[Message]] = None,
    tools: Optional[List[Dict]] = None,
    tool_choice: Optional[str] = None,
) -> ToolResponse:
    """Ask the LLM with tool calling capability."""
    
    # **構(gòu)建請(qǐng)求參數(shù)
    params = {
        "model": self.model,
        "messages": self._prepare_messages(messages, system_msgs),
    }
    
    # **添加工具信息
    if tools:
        params["tools"] = tools
        
    # **設(shè)置工具選擇模式
    if tool_choice:
        if tool_choice == "auto":
            params["tool_choice"] = "auto"
        elif tool_choice == "required":
            params["tool_choice"] = {"type": "function"}
    
    # **發(fā)送請(qǐng)求給OpenAI API
    response = await self.client.chat.completions.create(**params)
    
    # **解析響應(yīng)
    return self._parse_tool_response(response)

LLM會(huì)根據(jù)當(dāng)前任務(wù)和上下文,選擇最合適的工具來(lái)完成任務(wù)。這個(gè)選擇過(guò)程是由LLM的模型能力決定的,它會(huì)分析任務(wù)需求并選擇合適的工具。

工具選擇的具體例子

假設(shè)用戶輸入:"幫我搜索關(guān)于Python異步編程的信息"

  1. 這個(gè)請(qǐng)求被傳遞給Manus智能體
  2. Manus調(diào)用think方法,將請(qǐng)求和可用工具信息傳遞給LLM
  3. LLM分析請(qǐng)求,發(fā)現(xiàn)需要搜索信息,因此選擇GoogleSearch工具
  4. LLM返回一個(gè)包含工具調(diào)用信息的響應(yīng):
{
  "content": "我將使用Google搜索來(lái)查找關(guān)于Python異步編程的信息。",
  "tool_calls": [
    {
      "id": "call_123",
      "function": {
        "name": "google_search",
        "arguments": {
          "query": "Python異步編程 async await tutorial",
          "num_results": 5
        }
      }
    }
  ]
}
  1. think方法將這個(gè)工具調(diào)用信息保存在self.tool_calls中,并返回True表示需要執(zhí)行行動(dòng)

4.2 工具使用的實(shí)現(xiàn)

一旦工具被選擇,下一步就是使用工具。這主要在ToolCallAgent類(lèi)的act方法中實(shí)現(xiàn)。

工具使用的核心代碼

async def act(self) -> str:
    # **如果沒(méi)有工具調(diào)用,直接返回
    if not self.tool_calls:
        if self.tool_choices == "required":
            raise ValueError(TOOL_CALL_REQUIRED)
        return self.messages[-1].content or "No content or commands to execute"

    # **執(zhí)行每個(gè)工具調(diào)用
    results = []
    for command in self.tool_calls:
        # **執(zhí)行工具
        result = await self.execute_tool(command)
        logger.info(
            f"?? Tool '{command.function.name}' completed its mission! Result: {result}"
        )

        # **將工具響應(yīng)添加到記憶中
        tool_msg = Message.tool_message(
            content=result, tool_call_id=command.id, name=command.function.name
        )
        self.memory.add_message(tool_msg)
        results.append(result)

    # **返回所有工具執(zhí)行結(jié)果
    return "\n\n".join(results)

工具使用的關(guān)鍵步驟

  • 1) 執(zhí)行工具調(diào)用

execute_tool方法負(fù)責(zé)執(zhí)行單個(gè)工具調(diào)用:

async def execute_tool(self, command: ToolCall) -> str:
    """Execute a single tool call."""
    name = command.function.name
    args = command.function.arguments
    
    # **處理特殊工具
    if name in self.special_tool_names:
        return await self._handle_special_tool(name, args)
    
    # **執(zhí)行普通工具
    try:
        result = await self.available_tools.execute(name=name, tool_input=args)
        return result
    except Exception as e:
        error_msg = f"Error executing tool {name}: {str(e)}"
        logger.error(error_msg)
        return error_msg
  • 2) 工具集合的執(zhí)行
    ToolCollection類(lèi)的execute方法負(fù)責(zé)找到并執(zhí)行指定的工具:
async def execute(self, *, name: str, tool_input: Dict[str, Any] = None) -> ToolResult:
    """Execute a tool by name with given input."""
    # **查找工具
    tool = self.tool_map.get(name)
    if not tool:
        return ToolFailure(error=f"Tool {name} is invalid")
    
    # **執(zhí)行工具
    try:
        result = await tool(**tool_input)
        return result
    except ToolError as e:
        return ToolFailure(error=e.message)
  • 3) 具體工具的執(zhí)行
    每個(gè)工具都有一個(gè)execute方法,實(shí)現(xiàn)具體的功能。以GoogleSearch工具為例:
async def execute(self, query: str, num_results: int = 10) -> List[str]:
    """Execute a Google search with the given query."""
    try:
        # **調(diào)用Google搜索API
        search_results = await self.search_client.search(query, num_results=num_results)
        
        # **格式化結(jié)果
        formatted_results = []
        for i, result in enumerate(search_results, 1):
            title = result.get("title", "No title")
            link = result.get("link", "No link")
            snippet = result.get("snippet", "No snippet")
            formatted_results.append(f"{i}. {title}\n   URL: {link}\n   {snippet}\n")
        
        return "\n".join(formatted_results)
    except Exception as e:
        return f"Error performing Google search: {str(e)}"

工具使用的具體例子

繼續(xù)上面的例子,用戶要求搜索Python異步編程的信息:

  1. think方法已經(jīng)選擇了GoogleSearch工具,并保存了工具調(diào)用信息
  2. 接下來(lái),act方法被調(diào)用,它會(huì)遍歷self.tool_calls中的每個(gè)工具調(diào)用
  3. 對(duì)于GoogleSearch工具調(diào)用,execute_tool方法被調(diào)用:
result = await self.execute_tool(command)  # **command是GoogleSearch工具調(diào)用
  1. execute_tool方法找到GoogleSearch工具并執(zhí)行:
result = await self.available_tools.execute(
    name="google_search", 
    tool_input={"query": "Python異步編程 async await tutorial", "num_results": 5}
)
  1. ToolCollection.execute方法找到GoogleSearch工具實(shí)例并調(diào)用它的execute方法:
result = await google_search_tool.execute(
    query="Python異步編程 async await tutorial", 
    num_results=5
)
  1. GoogleSearch.execute方法執(zhí)行實(shí)際的搜索操作,并返回格式化的搜索結(jié)果:
1. Python異步編程入門(mén):理解async和await - 掘金
   URL: https://juejin.cn/post/6844904088201584654
   這篇文章詳細(xì)介紹了Python中的異步編程概念,包括async/await語(yǔ)法和使用方法...

2. Python異步編程指南 - 廖雪峰的官方網(wǎng)站
   URL: https://www.liaoxuefeng.com/wiki/1016959663602400/1048430311929344
   本教程介紹了Python中的協(xié)程和異步IO,以及如何使用async/await關(guān)鍵字...

...
  1. 這個(gè)結(jié)果被添加到智能體的記憶中,并返回給用戶

特殊工具的處理

OpenManus還有一些特殊工具,如Terminate工具,它們需要特殊處理:

async def _handle_special_tool(self, name: str, args: Dict[str, Any]) -> str:
    """Handle special tools that affect agent state."""
    if name == "terminate":
        status = args.get("status", "unknown")
        self.state = AgentState.FINISHED
        return f"Agent terminated with status: {status}"
    
    # **其他特殊工具的處理...
    
    return f"Special tool {name} executed with args: {args}"

當(dāng)Terminate工具被調(diào)用時(shí),智能體的狀態(tài)會(huì)被設(shè)置為FINISHED,這會(huì)導(dǎo)致run方法中的循環(huán)終止,從而結(jié)束智能體的執(zhí)行。


五、信息的保存、傳遞與使用

在OpenManus中,信息在智能體的多次迭代中通過(guò)精心設(shè)計(jì)的記憶系統(tǒng)進(jìn)行保存、傳遞和使用。這個(gè)系統(tǒng)確保了智能體能夠在多個(gè)步驟中保持上下文連貫性,并基于歷史信息做出決策。下文將詳細(xì)解析這個(gè)過(guò)程。

5.1 信息的保存機(jī)制

記憶系統(tǒng)的實(shí)現(xiàn)

OpenManus使用Memory類(lèi)來(lái)保存信息。這個(gè)類(lèi)在app/memory.py中定義:

class Memory:
    """存儲(chǔ)智能體交互歷史的記憶系統(tǒng)"""
    
    def __init__(self):
        self.messages: List[Message] = []
    
    def add_message(self, message: Message):
        """添加消息到記憶中"""
        self.messages.append(message)
    
    def get_messages(self) -> List[Message]:
        """獲取所有記憶中的消息"""
        return self.messages
    
    def clear(self):
        """清空記憶"""
        self.messages = []

Memory類(lèi)非常簡(jiǎn)單,它主要是一個(gè)消息列表的包裝器,提供了添加、獲取和清空消息的方法。

消息的結(jié)構(gòu)

每個(gè)消息都是Message類(lèi)的實(shí)例,這個(gè)類(lèi)定義了消息的結(jié)構(gòu):

class Message:
    """表示一條消息的類(lèi)"""
    
    def __init__(
        self,
        role: str,
        content: Optional[str] = None,
        tool_calls: Optional[List[ToolCall]] = None,
        tool_call_id: Optional[str] = None,
        name: Optional[str] = None,
    ):
        self.role = role
        self.content = content
        self.tool_calls = tool_calls
        self.tool_call_id = tool_call_id
        self.name = name
    
    @classmethod
    def system_message(cls, content: str) -> "Message":
        """創(chuàng)建系統(tǒng)消息"""
        return cls(role="system", content=content)
    
    @classmethod
    def user_message(cls, content: str) -> "Message":
        """創(chuàng)建用戶消息"""
        return cls(role="user", content=content)
    
    @classmethod
    def assistant_message(
        cls, content: str, tool_calls: Optional[List[ToolCall]] = None
    ) -> "Message":
        """創(chuàng)建助手消息"""
        return cls(role="assistant", content=content, tool_calls=tool_calls)
    
    @classmethod
    def tool_message(
        cls, content: str, tool_call_id: str, name: str
    ) -> "Message":
        """創(chuàng)建工具消息"""
        return cls(
            role="tool", content=content, tool_call_id=tool_call_id, name=name
        )

每個(gè)消息都有一個(gè)角色(系統(tǒng)、用戶、助手或工具)和內(nèi)容。助手消息可能還包含工具調(diào)用信息,工具消息包含工具調(diào)用ID和工具名稱(chēng)。

信息保存的時(shí)機(jī)

信息在多個(gè)地方被保存到記憶中:

  • 用戶輸入保存

當(dāng)用戶提供輸入時(shí),它會(huì)被保存到記憶中:

# **BaseAgent.run方法
if request:
    self.update_memory("user", request)

# **BaseAgent.update_memory方法
def update_memory(self, role: str, content: str):
    """更新記憶"""
    if role == "user":
        self.memory.add_message(Message.user_message(content))
    elif role == "system":
        self.memory.add_message(Message.system_message(content))
    elif role == "assistant":
        self.memory.add_message(Message.assistant_message(content))
  • LLM響應(yīng)保存

當(dāng)LLM生成響應(yīng)時(shí),它會(huì)被保存到記憶中:

# **ToolCallAgent.think方法
self.memory.add_message(Message.assistant_message(response.content, response.tool_calls))
  • 工具執(zhí)行結(jié)果保存

當(dāng)工具執(zhí)行完成時(shí),結(jié)果會(huì)被保存到記憶中:

# **ToolCallAgent.act方法
tool_msg = Message.tool_message(
    content=result, tool_call_id=command.id, name=command.function.name
)
self.memory.add_message(tool_msg)

5.2 信息的傳遞機(jī)制

消息傳遞給LLM

在每次調(diào)用LLM時(shí),記憶中的消息會(huì)被傳遞給LLM,使其能夠了解歷史上下文:

# **ToolCallAgent.think方法
response = await self.llm.ask_tool(
    messages=self.messages,  # **傳遞記憶中的所有消息
    system_msgs=[Message.system_message(self.system_prompt)]
    if self.system_prompt
    else None,
    tools=self.available_tools.to_params(),
    tool_choice=self.tool_choices,
)

這里的self.messages就是記憶中的消息列表。

消息格式轉(zhuǎn)換

在傳遞給LLM之前,消息需要轉(zhuǎn)換為L(zhǎng)LM能夠理解的格式:

# **LLM._prepare_messages方法
def _prepare_messages(
    self, messages: List[Message], system_msgs: Optional[List[Message]] = None
) -> List[Dict]:
    """準(zhǔn)備發(fā)送給LLM的消息"""
    prepared_messages = []
    
    # **添加系統(tǒng)消息
    if system_msgs:
        for msg in system_msgs:
            prepared_messages.append({"role": msg.role, "content": msg.content})
    
    # **添加歷史消息
    for msg in messages:
        message_dict = {"role": msg.role, "content": msg.content}
        
        # **添加工具調(diào)用信息
        if msg.role == "assistant" and msg.tool_calls:
            message_dict["tool_calls"] = [
                {
                    "id": tc.id,
                    "type": "function",
                    "function": {
                        "name": tc.function.name,
                        "arguments": json.dumps(tc.function.arguments),
                    },
                }
                for tc in msg.tool_calls
            ]
        
        # **添加工具響應(yīng)信息
        if msg.role == "tool":
            message_dict["tool_call_id"] = msg.tool_call_id
            message_dict["name"] = msg.name
        
        prepared_messages.append(message_dict)
    
    return prepared_messages

這個(gè)方法將Message對(duì)象轉(zhuǎn)換為包含適當(dāng)字段的字典,這些字典符合OpenAI API的消息格式要求。

5.3 信息的使用機(jī)制

LLM使用歷史信息

LLM會(huì)使用傳遞給它的所有歷史消息來(lái)生成響應(yīng)。這使得它能夠:

  • 理解當(dāng)前任務(wù)的上下文
  • 記住之前的交互
  • 基于之前的工具執(zhí)行結(jié)果做出決策
  • 避免重復(fù)之前的錯(cuò)誤

例如,如果用戶要求搜索信息,智能體使用GoogleSearch工具獲取結(jié)果,然后用戶要求總結(jié)這些結(jié)果,LLM會(huì)使用記憶中的搜索結(jié)果來(lái)生成總結(jié),而不需要重新搜索。

記憶長(zhǎng)度管理

由于LLM的上下文窗口有限,記憶中的消息數(shù)量可能需要限制。OpenManus可能使用以下策略來(lái)管理記憶長(zhǎng)度:

  • 保留最新的N條消息
  • 保留重要的消息(如系統(tǒng)提示、用戶請(qǐng)求)
  • 壓縮或摘要化舊消息

雖然代碼中沒(méi)有明確實(shí)現(xiàn)這些策略,但這是處理長(zhǎng)期交互的常見(jiàn)做法。

記憶持久化

當(dāng)前的實(shí)現(xiàn)中,記憶只存在于內(nèi)存中,當(dāng)智能體實(shí)例被銷(xiāo)毀時(shí)記憶也會(huì)丟失。為了支持長(zhǎng)期記憶,可能需要將記憶持久化到數(shù)據(jù)庫(kù)或文件中。

具體例子:多步驟任務(wù)執(zhí)行

下文將通過(guò)一個(gè)具體例子來(lái)說(shuō)明信息如何在多次迭代中流動(dòng):

假設(shè)用戶要求:"幫我找到關(guān)于Python異步編程的信息,然后創(chuàng)建一個(gè)簡(jiǎn)單的異步程序"

步驟1:理解請(qǐng)求

  1. 用戶請(qǐng)求被添加到記憶中:Message.user_message("幫我找到關(guān)于Python異步編程的信息,然后創(chuàng)建一個(gè)簡(jiǎn)單的異步程序")
  2. LLM接收這個(gè)消息,并決定首先需要搜索信息
  3. LLM的響應(yīng)被添加到記憶中:Message.assistant_message("我將搜索Python異步編程的信息", [ToolCall(google_search)])

步驟2:執(zhí)行搜索

  1. GoogleSearch工具被執(zhí)行,搜索"Python異步編程"
  2. 搜索結(jié)果被添加到記憶中:Message.tool_message(搜索結(jié)果, tool_call_id, "google_search")
  3. LLM接收所有歷史消息,包括搜索結(jié)果
  4. LLM分析搜索結(jié)果,決定創(chuàng)建一個(gè)異步程序
  5. LLM的響應(yīng)被添加到記憶中:Message.assistant_message("根據(jù)搜索結(jié)果,我將創(chuàng)建一個(gè)簡(jiǎn)單的異步程序", [ToolCall(python_execute)])

步驟3:創(chuàng)建程序

  1. PythonExecute工具被執(zhí)行,創(chuàng)建并運(yùn)行一個(gè)異步程序
  2. 執(zhí)行結(jié)果被添加到記憶中:Message.tool_message(執(zhí)行結(jié)果, tool_call_id, "python_execute")
  3. LLM接收所有歷史消息,包括程序執(zhí)行結(jié)果
  4. LLM生成最終響應(yīng),解釋程序并總結(jié)任務(wù)完成情況
  5. LLM的響應(yīng)被添加到記憶中:Message.assistant_message("我已經(jīng)創(chuàng)建了一個(gè)簡(jiǎn)單的異步程序...", None)

在這個(gè)過(guò)程中,信息不斷被添加到記憶中,并在每次LLM調(diào)用時(shí)傳遞給LLM,使其能夠基于完整的歷史上下文做出決策。

5.4 總結(jié)

OpenManus中的信息通過(guò)記憶系統(tǒng)在多次迭代中流動(dòng)。用戶輸入、LLM響應(yīng)和工具執(zhí)行結(jié)果都被保存到記憶中,并在每次LLM調(diào)用時(shí)傳遞給LLM。這使得智能體能夠保持上下文連貫性,累積知識(shí),分解任務(wù),并從錯(cuò)誤中恢復(fù)。

這種信息保存、傳遞和使用機(jī)制有幾個(gè)重要優(yōu)勢(shì):

  • 上下文連貫性 :智能體能夠保持對(duì)話的連貫性,理解之前的交互并基于它們做出決策。
  • 累積知識(shí):智能體可以累積知識(shí),將之前步驟獲取的信息用于后續(xù)步驟。
  • 任務(wù)分解:復(fù)雜任務(wù)可以分解為多個(gè)步驟,每個(gè)步驟都基于之前步驟的結(jié)果。
  • 錯(cuò)誤恢復(fù):如果某個(gè)步驟失敗,智能體可以看到錯(cuò)誤信息并嘗試不同的方法。

這種設(shè)計(jì)使得OpenManus能夠處理復(fù)雜的多步驟任務(wù),每個(gè)步驟都能利用之前步驟獲取的信息,創(chuàng)造出超越單個(gè)步驟能力的解決方案。


六、錯(cuò)誤處理與容錯(cuò)機(jī)制

在OpenManus中,確保LLM給出正確響應(yīng)以及處理錯(cuò)誤響應(yīng)是一個(gè)重要的環(huán)節(jié)

6.1 確保LLM給出正確響應(yīng)的機(jī)制

系統(tǒng)提示和指令

OpenManus通過(guò)精心設(shè)計(jì)的系統(tǒng)提示來(lái)引導(dǎo)LLM生成正確的響應(yīng)。在Manus類(lèi)中,系統(tǒng)提示定義如下:

system_prompt: str = SYSTEM_PROMPT

這個(gè)SYSTEM_PROMPT包含了詳細(xì)的指導(dǎo),告訴LLM如何分析任務(wù)、選擇工具和格式化響應(yīng)。例如:

SYSTEM_PROMPT = """SETTING: You are an autonomous programmer, and you're working directly in the command line with a special interface.

  

The special interface consists of a file editor that shows you {{WINDOW}} lines of a file at a time.

In addition to typical bash commands, you can also use specific commands to help you navigate and edit files.

To call a command, you need to invoke it with a function call/tool call.

  

Please note that THE EDIT COMMAND REQUIRES PROPER INDENTATION.

If you'd like to add the line ' print(x)' you must fully write that out, with all those spaces before the code! Indentation is important and code that is not indented correctly will fail and require fixing before it can be run.

  

RESPONSE FORMAT:

Your shell prompt is formatted as follows:

(Open file: <path>)

(Current directory: <cwd>)

bash-$

  

First, you should _always_ include a general thought about what you're going to do next.

Then, for every response, you must include exactly _ONE_ tool call/function call.

  

Remember, you should always include a _SINGLE_ tool call/function call and then wait for a response from the shell before continuing with more discussion and commands. Everything you include in the DISCUSSION section will be saved for future reference.

If you'd like to issue two commands at once, PLEASE DO NOT DO THAT! Please instead first submit just the first tool call, and then after receiving a response you'll be able to issue the second tool call.

Note that the environment does NOT support interactive session commands (e.g. python, vim), so please do not invoke them.

"""

工具參數(shù)驗(yàn)證

每個(gè)工具都定義了明確的參數(shù)規(guī)范,這些規(guī)范會(huì)在調(diào)用LLM時(shí)傳遞給它:

parameters: dict = {
    "type": "object",
    "properties": {
        "query": {
            "type": "string",
            "description": "搜索查詢"
        },
        "num_results": {
            "type": "integer",
            "description": "返回結(jié)果數(shù)量"
        }
    },
    "required": ["query"]
}

這些參數(shù)規(guī)范告訴LLM每個(gè)工具需要哪些參數(shù),哪些是必需的,以及它們的類(lèi)型和描述。

工具選擇模式

ToolCallAgent類(lèi)提供了不同的工具選擇模式:

tool_choices: str = "auto"  # **可以是"none"、"auto"或"required"
  • "none":不要求LLM使用工具
  • "auto":LLM可以自行決定是否使用工具
  • "required":要求LLM必須使用工具

這個(gè)設(shè)置會(huì)影響發(fā)送給OpenAI API的tool_choice參數(shù):

if tool_choice:
    if tool_choice == "auto":
        params["tool_choice"] = "auto"
    elif tool_choice == "required":
        params["tool_choice"] = {"type": "function"}

6.2 處理不正確響應(yīng)的機(jī)制

盡管有上述機(jī)制,LLM仍可能給出不正確的響應(yīng)。OpenManus通過(guò)多層錯(cuò)誤處理來(lái)應(yīng)對(duì)這種情況:

響應(yīng)解析和驗(yàn)證

LLM類(lèi)的_parse_tool_response方法中,會(huì)解析和驗(yàn)證LLM的響應(yīng):

def _parse_tool_response(self, response: ChatCompletion) -> ToolResponse:
    """Parse the response from the LLM with tool calls."""
    choice = response.choices[0]
    message = choice.message
    
    # **解析工具調(diào)用
    tool_calls = []
    if message.tool_calls:
        for tc in message.tool_calls:
            try:
                # **解析工具調(diào)用參數(shù)
                args = json.loads(tc.function.arguments)
                tool_calls.append(
                    ToolCall(
                        id=tc.id,
                        function=FunctionCall(
                            name=tc.function.name,
                            arguments=args,
                        ),
                    )
                )
            except json.JSONDecodeError:
                logger.error(f"Failed to parse tool call arguments: {tc.function.arguments}")
    
    # **返回解析后的響應(yīng)
    return ToolResponse(
        content=message.content or "",
        tool_calls=tool_calls,
    )

這個(gè)方法會(huì)嘗試解析LLM返回的工具調(diào)用參數(shù),如果解析失?。ɡ?,參數(shù)不是有效的JSON),會(huì)記錄錯(cuò)誤但不會(huì)中斷執(zhí)行。

工具執(zhí)行錯(cuò)誤處理

ToolCallAgentexecute_tool方法中,會(huì)捕獲工具執(zhí)行過(guò)程中的任何異常:

async def execute_tool(self, command: ToolCall) -> str:
    """Execute a single tool call."""
    name = command.function.name
    args = command.function.arguments
    
    # **處理特殊工具
    if name in self.special_tool_names:
        return await self._handle_special_tool(name, args)
    
    # **執(zhí)行普通工具
    try:
        result = await self.available_tools.execute(name=name, tool_input=args)
        return result
    except Exception as e:
        error_msg = f"Error executing tool {name}: {str(e)}"
        logger.error(error_msg)
        return error_msg  # **返回錯(cuò)誤信息而不是拋出異常

如果工具執(zhí)行失敗,會(huì)返回一個(gè)錯(cuò)誤消息,而不是中斷整個(gè)執(zhí)行流程。

工具參數(shù)驗(yàn)證

ToolCollectionexecute方法中,會(huì)檢查工具是否存在:

async def execute(self, *, name: str, tool_input: Dict[str, Any] = None) -> ToolResult:
    """Execute a tool by name with given input."""
    # **查找工具
    tool = self.tool_map.get(name)
    if not tool:
        return ToolFailure(error=f"Tool {name} is invalid")  # **返回錯(cuò)誤結(jié)果
    
    # **執(zhí)行工具
    try:
        result = await tool(**tool_input)
        return result
    except ToolError as e:
        return ToolFailure(error=e.message)  # **返回錯(cuò)誤結(jié)果

如果指定的工具不存在,或者執(zhí)行過(guò)程中拋出ToolError異常,會(huì)返回一個(gè)ToolFailure對(duì)象,而不是中斷執(zhí)行。

具體工具的參數(shù)驗(yàn)證

每個(gè)工具都可以實(shí)現(xiàn)自己的參數(shù)驗(yàn)證邏輯。例如,GoogleSearch工具會(huì)檢查查詢參數(shù)是否有效:

async def execute(self, query: str, num_results: int = 10) -> List[str]:
    """Execute a Google search with the given query."""
    if not query or not isinstance(query, str):
        return "Error: Invalid query parameter"
    
    try:
        # **執(zhí)行搜索...
    except Exception as e:
        return f"Error performing Google search: {str(e)}"

卡住狀態(tài)檢測(cè)

BaseAgent類(lèi)實(shí)現(xiàn)了卡住狀態(tài)檢測(cè)機(jī)制,用于處理LLM陷入循環(huán)的情況:

def is_stuck(self) -> bool:
    """Check if the agent is stuck in a loop."""
    if len(self.memory.messages) < 4:
        return False
    
    # **檢查最近的消息是否重復(fù)
    recent_messages = self.memory.messages[-4:]
    content_set = set(msg.content for msg in recent_messages)
    return len(content_set) <= 1  # **如果最近4條消息內(nèi)容相同,認(rèn)為卡住了

def handle_stuck_state(self):
    """Handle the case when agent is stuck in a loop."""
    logger.warning("Agent appears to be stuck in a loop. Adding intervention...")
    
    # **添加干預(yù)消息
    intervention_msg = Message.system_message(
        "You appear to be stuck in a loop. Please try a different approach."
    )
    self.memory.add_message(intervention_msg)

如果檢測(cè)到智能體陷入循環(huán)(最近幾條消息內(nèi)容相同),會(huì)添加一條干預(yù)消息,提示LLM嘗試不同的方法。

6.3 具體例子

例子1:工具參數(shù)不正確

假設(shè)LLM選擇了GoogleSearch工具,但提供了錯(cuò)誤的參數(shù)格式:

{
  "function": {
    "name": "google_search",
    "arguments": "Python異步編程"  // 錯(cuò)誤:應(yīng)該是一個(gè)JSON對(duì)象,而不是字符串
  }
}

處理流程:

  1. _parse_tool_response方法嘗試解析參數(shù),但會(huì)失敗
  2. 錯(cuò)誤會(huì)被記錄,但不會(huì)中斷執(zhí)行
  3. execute_tool方法會(huì)收到一個(gè)空的參數(shù)字典
  4. GoogleSearch.execute方法會(huì)檢測(cè)到無(wú)效參數(shù)并返回錯(cuò)誤消息
  5. 錯(cuò)誤消息會(huì)被添加到智能體的記憶中
  6. 在下一個(gè)步驟中,LLM會(huì)看到這個(gè)錯(cuò)誤消息,并可能?chē)L試修正參數(shù)格式

例子2:工具不存在

假設(shè)LLM選擇了一個(gè)不存在的工具:

{
  "function": {
    "name": "nonexistent_tool",
    "arguments": {}
  }
}

處理流程:

  1. execute_tool方法會(huì)調(diào)用ToolCollection.execute方法
  2. ToolCollection.execute方法會(huì)檢查工具是否存在
  3. 由于工具不存在,會(huì)返回一個(gè)錯(cuò)誤消息:"Tool nonexistent_tool is invalid"
  4. 這個(gè)錯(cuò)誤消息會(huì)被添加到智能體的記憶中
  5. 在下一個(gè)步驟中,LLM會(huì)看到這個(gè)錯(cuò)誤消息,并可能選擇一個(gè)存在的工具

例子3:工具執(zhí)行失敗

假設(shè)LLM選擇了GoogleSearch工具,參數(shù)正確,但搜索API暫時(shí)不可用:

{
  "function": {
    "name": "google_search",
    "arguments": {
      "query": "Python異步編程"
    }
  }
}

處理流程:

  1. execute_tool方法會(huì)調(diào)用GoogleSearch.execute方法
  2. GoogleSearch.execute方法會(huì)嘗試調(diào)用搜索API
  3. 由于API不可用,會(huì)拋出異常
  4. 異常會(huì)被捕獲,并返回一個(gè)錯(cuò)誤消息:"Error performing Google search: API not available"
  5. 這個(gè)錯(cuò)誤消息會(huì)被添加到智能體的記憶中
  6. 在下一個(gè)步驟中,LLM會(huì)看到這個(gè)錯(cuò)誤消息,并可能?chē)L試其他方法或工具

6.4 總結(jié)

OpenManus通過(guò)多層錯(cuò)誤處理機(jī)制來(lái)確保即使LLM給出不正確的響應(yīng),系統(tǒng)也能繼續(xù)運(yùn)行:

  1. 預(yù)防措施

    • 精心設(shè)計(jì)的系統(tǒng)提示
    • 明確的工具參數(shù)規(guī)范
    • 靈活的工具選擇模式
  2. 錯(cuò)誤處理

    • 響應(yīng)解析和驗(yàn)證
    • 工具執(zhí)行錯(cuò)誤捕獲
    • 工具參數(shù)驗(yàn)證
    • 卡住狀態(tài)檢測(cè)和干預(yù)

這種設(shè)計(jì)使得OpenManus能夠在各種情況下保持穩(wěn)定運(yùn)行,即使LLM偶爾給出不正確的響應(yīng)。


七、實(shí)際應(yīng)用場(chǎng)景與擴(kuò)展可能

7.1 OpenManus的應(yīng)用場(chǎng)景

OpenManus作為一個(gè)功能強(qiáng)大的智能體框架,可以應(yīng)用于多種場(chǎng)景:

自動(dòng)化助手

OpenManus可以作為個(gè)人或團(tuán)隊(duì)的自動(dòng)化助手,執(zhí)行各種任務(wù):

  • 信息收集與整理:自動(dòng)搜索、瀏覽網(wǎng)頁(yè)并整理信息
  • 代碼生成與測(cè)試:根據(jù)需求生成代碼,并自動(dòng)執(zhí)行測(cè)試
  • 文檔處理:自動(dòng)生成、修改和整理文檔
  • 數(shù)據(jù)分析:執(zhí)行數(shù)據(jù)處理腳本,生成分析報(bào)告

研究輔助工具

研究人員可以使用OpenManus來(lái)加速研究過(guò)程:

  • 文獻(xiàn)調(diào)研:自動(dòng)搜索和總結(jié)相關(guān)文獻(xiàn)
  • 數(shù)據(jù)處理:執(zhí)行數(shù)據(jù)清洗和轉(zhuǎn)換腳本
  • 實(shí)驗(yàn)自動(dòng)化:設(shè)計(jì)、執(zhí)行和記錄實(shí)驗(yàn)
  • 結(jié)果可視化:生成圖表和可視化展示

教育與學(xué)習(xí)工具

OpenManus可以作為教育和學(xué)習(xí)的輔助工具:

  • 交互式教程:創(chuàng)建能夠執(zhí)行代碼示例的教程
  • 問(wèn)題解答:回答學(xué)習(xí)問(wèn)題并提供實(shí)際示例
  • 編程練習(xí):生成編程練習(xí)并評(píng)估解答
  • 知識(shí)探索:幫助學(xué)習(xí)者探索新領(lǐng)域

開(kāi)發(fā)輔助系統(tǒng)

軟件開(kāi)發(fā)人員可以使用OpenManus來(lái)提高開(kāi)發(fā)效率:

  • 代碼生成:根據(jù)需求生成代碼框架或完整實(shí)現(xiàn)
  • 調(diào)試輔助:分析錯(cuò)誤信息并提供修復(fù)建議
  • 文檔生成:自動(dòng)生成代碼文檔和API說(shuō)明
  • 測(cè)試自動(dòng)化:生成測(cè)試用例并執(zhí)行測(cè)試

7.2 添加新工具的方法

OpenManus的模塊化設(shè)計(jì)使得添加新工具變得簡(jiǎn)單直接。以下是添加新工具的步驟:

1) 創(chuàng)建工具類(lèi)

首先,創(chuàng)建一個(gè)繼承自BaseTool的新類(lèi):

from app.tool.base import BaseTool

class MyNewTool(BaseTool):
    name: str = "my_new_tool"
    description: str = "這個(gè)工具的功能描述"
    parameters: dict = {
        "type": "object",
        "properties": {
            "param1": {
                "type": "string",
                "description": "參數(shù)1的描述"
            },
            "param2": {
                "type": "integer",
                "description": "參數(shù)2的描述"
            }
        },
        "required": ["param1"]
    }
    
    async def execute(self, param1: str, param2: int = 0) -> str:
        """執(zhí)行工具的具體邏輯"""
        try:
            # **實(shí)現(xiàn)工具功能
            result = f"處理 {param1} 和 {param2} 的結(jié)果"
            return result
        except Exception as e:
            return f"工具執(zhí)行出錯(cuò): {str(e)}"

2) 實(shí)現(xiàn)工具功能

execute方法中實(shí)現(xiàn)工具的具體功能。確保:

  • 處理所有可能的異常
  • 返回清晰的結(jié)果或錯(cuò)誤信息
  • 使用異步編程(async/await)處理I/O操作

3) 添加工具到智能體

將新工具添加到智能體的可用工具列表中:

from app.agent.toolcall import ToolCallAgent
from app.tool.tool_collection import ToolCollection
from my_new_tool import MyNewTool

class MyCustomAgent(ToolCallAgent):
    available_tools: ToolCollection = Field(
        default_factory=lambda: ToolCollection(
            PythonExecute(), GoogleSearch(), MyNewTool()
        )
    )

或者修改現(xiàn)有的Manus智能體:

from app.agent.manus import Manus
from app.tool.tool_collection import ToolCollection
from my_new_tool import MyNewTool

# **創(chuàng)建包含新工具的Manus實(shí)例
agent = Manus()
agent.available_tools = ToolCollection(
    *agent.available_tools.tools,  # **保留原有工具
    MyNewTool()  # **添加新工具
)

4) 更新系統(tǒng)提示(可選)

如果新工具需要特殊的使用說(shuō)明,可以更新系統(tǒng)提示:

agent.system_prompt += "\n你現(xiàn)在可以使用my_new_tool工具來(lái)處理特定任務(wù)。"

7.3 定制智能體的方法

OpenManus的層次化設(shè)計(jì)使得定制智能體變得靈活多樣:

1) 繼承現(xiàn)有智能體

最簡(jiǎn)單的方法是繼承現(xiàn)有的智能體類(lèi):

from app.agent.toolcall import ToolCallAgent

class MySpecializedAgent(ToolCallAgent):
    name: str = "MySpecializedAgent"
    description: str = "專(zhuān)門(mén)用于特定任務(wù)的智能體"
    
    system_prompt: str = "你是一個(gè)專(zhuān)門(mén)用于特定任務(wù)的智能體..."
    next_step_prompt: str = "分析當(dāng)前狀態(tài),決定下一步行動(dòng)..."
    
    # **自定義屬性和方法
    max_steps: int = 30
    
    async def think(self) -> bool:
        # **自定義思考邏輯
        return await super().think()

2) 自定義工具集

為特定任務(wù)定制工具集:

from app.tool.tool_collection import ToolCollection
from app.tool.python_execute import PythonExecute
from app.tool.file_saver import FileSaver
from my_special_tools import ToolA, ToolB

class DataAnalysisAgent(ToolCallAgent):
    available_tools: ToolCollection = Field(
        default_factory=lambda: ToolCollection(
            PythonExecute(), FileSaver(), ToolA(), ToolB()
        )
    )

3) 自定義提示模板

為智能體定制專(zhuān)門(mén)的提示模板:

# **在app/prompt/my_agent.py中定義
SYSTEM_PROMPT = """
你是一個(gè)專(zhuān)門(mén)用于數(shù)據(jù)分析的智能體。你可以使用以下工具:
1. python_execute - 執(zhí)行Python代碼,特別是數(shù)據(jù)分析代碼
2. file_saver - 保存分析結(jié)果
3. tool_a - 特殊功能A
4. tool_b - 特殊功能B

分析任務(wù)時(shí),請(qǐng)遵循以下步驟:
1. 理解數(shù)據(jù)結(jié)構(gòu)
2. 執(zhí)行初步分析
3. 可視化關(guān)鍵指標(biāo)
4. 生成分析報(bào)告
"""

# **在智能體中使用
from app.prompt.my_agent import SYSTEM_PROMPT

class DataAnalysisAgent(ToolCallAgent):
    system_prompt: str = SYSTEM_PROMPT

4) 自定義執(zhí)行流程

通過(guò)重寫(xiě)step、thinkact方法來(lái)自定義執(zhí)行流程:

async def step(self) -> str:
    """自定義步驟執(zhí)行邏輯"""
    # **記錄步驟開(kāi)始時(shí)間
    start_time = time.time()
    
    # **執(zhí)行標(biāo)準(zhǔn)的思考-行動(dòng)循環(huán)
    should_act = await self.think()
    if not should_act:
        return "思考完成 - 無(wú)需行動(dòng)"
    
    result = await self.act()
    
    # **記錄步驟執(zhí)行時(shí)間
    execution_time = time.time() - start_time
    logger.info(f"步驟執(zhí)行時(shí)間: {execution_time:.2f}秒")
    
    return result

7.4 集成到其他系統(tǒng)的可能性

OpenManus的設(shè)計(jì)使其易于集成到各種系統(tǒng)中:

1) Web應(yīng)用集成

將OpenManus集成到Web應(yīng)用中:

from fastapi import FastAPI, Request
from app.agent.manus import Manus

app = FastAPI()
agents = {}  # **存儲(chǔ)用戶會(huì)話的智能體實(shí)例

@app.post("/chat/{session_id}")
async def chat(session_id: str, request: Request):
    data = await request.json()
    user_input = data.get("message", "")
    
    # **獲取或創(chuàng)建智能體實(shí)例
    if session_id not in agents:
        agents[session_id] = Manus()
    
    # **運(yùn)行智能體
    result = await agents[session_id].run(user_input)
    
    return {"response": result}

2) 桌面應(yīng)用集成

將OpenManus集成到桌面應(yīng)用中:

import tkinter as tk
from app.agent.manus import Manus

class ManusApp:
    def __init__(self, root):
        self.root = root
        self.agent = Manus()
        # **設(shè)置UI組件
        
    async def send_message(self):
        user_input = self.input_field.get()
        self.input_field.delete(0, tk.END)
        
        # **在后臺(tái)線程中運(yùn)行智能體
        result = await self.agent.run(user_input)
        
        # **更新UI顯示結(jié)果
        self.display_area.insert(tk.END, f"結(jié)果: {result}\n")

3) API服務(wù)集成

將OpenManus作為API服務(wù)提供:

from fastapi import FastAPI
from pydantic import BaseModel
from app.agent.manus import Manus

app = FastAPI()
agent = Manus()

class QueryRequest(BaseModel):
    query: str

@app.post("/api/query")
async def query(request: QueryRequest):
    result = await agent.run(request.query)
    return {"result": result}

4) 自動(dòng)化工作流集成

將OpenManus集成到自動(dòng)化工作流中:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
from app.agent.manus import Manus

def run_manus_task(task_description):
    agent = Manus()
    result = asyncio.run(agent.run(task_description))
    return result

with DAG('manus_workflow', start_date=datetime(2023, 1, 1), schedule_interval='@daily') as dag:
    task1 = PythonOperator(
        task_id='data_collection',
        python_callable=run_manus_task,
        op_kwargs={'task_description': '收集今天的市場(chǎng)數(shù)據(jù)'}
    )
    
    task2 = PythonOperator(
        task_id='data_analysis',
        python_callable=run_manus_task,
        op_kwargs={'task_description': '分析收集到的市場(chǎng)數(shù)據(jù)'}
    )
    
    task1 >> task2

7.5 未來(lái)發(fā)展方向

OpenManus作為一個(gè)開(kāi)源項(xiàng)目,有多個(gè)潛在的發(fā)展方向:

多智能體協(xié)作

實(shí)現(xiàn)多個(gè)智能體之間的協(xié)作,每個(gè)智能體專(zhuān)注于特定領(lǐng)域:

  • 專(zhuān)家智能體:不同領(lǐng)域的專(zhuān)家智能體
  • 協(xié)調(diào)智能體:負(fù)責(zé)分配任務(wù)和整合結(jié)果
  • 評(píng)估智能體:評(píng)估其他智能體的輸出質(zhì)量
  • 對(duì)抗智能體:提供批判性思考和挑戰(zhàn)

增強(qiáng)學(xué)習(xí)能力

為智能體添加更強(qiáng)的學(xué)習(xí)能力:

  • 長(zhǎng)期記憶:持久化存儲(chǔ)重要信息
  • 知識(shí)庫(kù)集成:連接到專(zhuān)門(mén)的知識(shí)庫(kù)
  • 示例學(xué)習(xí):從用戶示例中學(xué)習(xí)新技能
  • 反饋學(xué)習(xí):根據(jù)用戶反饋調(diào)整行為

自主性增強(qiáng)

增強(qiáng)智能體的自主性:

  • 自主規(guī)劃:智能體自主規(guī)劃解決問(wèn)題的步驟
  • 自主探索:主動(dòng)探索未知領(lǐng)域
  • 自主學(xué)習(xí):識(shí)別知識(shí)缺口并主動(dòng)學(xué)習(xí)
  • 自主決策:在不確定情況下做出決策

工具生態(tài)擴(kuò)展

擴(kuò)展工具生態(tài)系統(tǒng):

  • 多模態(tài)工具:處理圖像、音頻、視頻的工具
  • 專(zhuān)業(yè)領(lǐng)域工具:針對(duì)特定領(lǐng)域的專(zhuān)業(yè)工具
  • 工具市場(chǎng):允許社區(qū)貢獻(xiàn)和分享工具
  • 工具組合:自動(dòng)組合多個(gè)工具創(chuàng)建復(fù)雜功能

安全與隱私增強(qiáng)

加強(qiáng)安全性和隱私保護(hù):

  • 沙箱執(zhí)行:更安全的代碼執(zhí)行環(huán)境
  • 權(quán)限控制:細(xì)粒度的工具訪問(wèn)權(quán)限
  • 隱私保護(hù):敏感信息的處理機(jī)制
  • 審計(jì)日志:詳細(xì)記錄智能體的所有操作

用戶體驗(yàn)優(yōu)化

改善用戶與智能體的交互體驗(yàn):

  • 自然對(duì)話:更自然的對(duì)話界面
  • 多模態(tài)交互:支持語(yǔ)音、圖像等交互方式
  • 進(jìn)度反饋:實(shí)時(shí)顯示任務(wù)進(jìn)度
  • 可解釋性:解釋決策過(guò)程和推理邏輯

OpenManus作為一個(gè)靈活且強(qiáng)大的智能體框架,有著廣闊的應(yīng)用前景和發(fā)展空間。隨著技術(shù)的進(jìn)步和社區(qū)的貢獻(xiàn),它有潛力成為構(gòu)建下一代AI應(yīng)用的重要基礎(chǔ)設(shè)施。

八、拙見(jiàn)

關(guān)于大模型的可靠性
大模型的輸出仍然需要大量后期校驗(yàn)來(lái)修正,這一過(guò)程既可以依賴(lài)人工編寫(xiě)的規(guī)則,也可以借助大模型自身進(jìn)行自糾自查?;蛟S,隨著技術(shù)的發(fā)展,模型自我優(yōu)化的能力會(huì)進(jìn)一步增強(qiáng),使得人工干預(yù)的成本逐步降低。

關(guān)于 Agent 開(kāi)發(fā)
當(dāng)前,大模型的主流交互方式仍然較為低效,Agent 的引入可以在一定程度上提升交互效率。然而,這種形態(tài)仍讓我聯(lián)想到清朝末年“馬拉火車(chē)”的景象——新舊技術(shù)的混搭或許只是過(guò)渡階段,最終更高效、更自然的交互方式將會(huì)取代它。

關(guān)于 Prompt 工程
Prompt 工程從最初的火熱討論,到如今的相對(duì)沉寂,反映了技術(shù)的演進(jìn)趨勢(shì):交互方式正變得越來(lái)越自然,復(fù)雜性逐漸被底層封裝,最終回歸到最樸素的形態(tài)。然而,在 Agent 開(kāi)發(fā)中,Prompt 設(shè)計(jì)仍然是重要環(huán)節(jié),或許未來(lái)這一過(guò)程也會(huì)被進(jìn)一步抽象化,讓開(kāi)發(fā)者無(wú)需直接接觸底層邏輯,而是通過(guò)更直觀的方式操控大模型。

關(guān)于大模型之間的交互
曾經(jīng)看到一個(gè)視頻,內(nèi)容是兩個(gè)大模型通過(guò)語(yǔ)音聊天,在確認(rèn)對(duì)方也是 AI 后,切換為一種奇特的電子音交互模式。這是否真實(shí)尚未求證,但這一思路令人遐想——或許未來(lái)的大模型之間需要一種全新的交互協(xié)議,使得它們能夠高效、準(zhǔn)確地溝通,而非模擬人類(lèi)語(yǔ)言的方式。

關(guān)于軟件開(kāi)發(fā)
從單機(jī)到聯(lián)網(wǎng),從物理機(jī)到云,每一次基礎(chǔ)設(shè)施的變革都推動(dòng)了軟件開(kāi)發(fā)范式的演進(jìn)。如今,大模型正在成為新一代的基礎(chǔ)設(shè)施,這意味著未來(lái)的軟件開(kāi)發(fā)可能需要與大模型深度適配,甚至直接基于大模型構(gòu)建。也許,Agent 只是過(guò)渡形態(tài),更激進(jìn)的方式是——模型即軟件。

關(guān)于端側(cè)智能
大模型的橫空出世,使得在聯(lián)網(wǎng)環(huán)境下,邊緣計(jì)算和端側(cè)智能的意義似乎變得模糊。當(dāng)計(jì)算能力可以按需獲取,設(shè)備本地的智能化或許不再是核心問(wèn)題,而是如何高效地利用模型資源,真正實(shí)現(xiàn)智能無(wú)處不在。

最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點(diǎn),簡(jiǎn)書(shū)系信息發(fā)布平臺(tái),僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容