
FastOpenAPI:Python框架API文檔自動化生成工具,解放你的雙手!
規劃模塊是Manus的”大腦”,負責理解用戶意圖,將復雜任務分解為可執行的步驟,并制定執行計劃。這一模塊使Manus能夠處理抽象的任務描述,并將其轉化為具體的行動步驟。
作為系統的決策中樞,規劃模塊實現:
記憶模塊使Manus能夠存儲和利用歷史信息,提高任務執行的連貫性和個性化程度。該模塊管理三類關鍵信息:
構建長期記憶體系:
class MemorySystem:
def __init__(self):
self.user_profile = UserVector() # 用戶偏好向量
self.history_db = ChromaDB() # 交互歷史數據庫
self.cache = LRUCache() # 短期記憶緩存
工具使用模塊是Manus的”手”,負責實際執行各種操作。該模塊能夠調用和使用多種工具來完成任務,包括:
這種多工具集成能力使Manus能夠處理各種復雜任務,從信息收集到內容創建,再到數據分析。
Multi-Agent 系統(MAS)由多個交互的智能體組成,每個智能體都是能夠感知、學習環境模型、做出決策并執行行動的自主實體。這些智能體可以是軟件程序、機器人、無人機、傳感器、人類,或它們的組合。
在典型的 Multi-Agent 架構中,各個智能體具有專業化的能力和目標。例如,一個系統可能包含專注于內容摘要、翻譯、內容生成等不同任務的智能體。它們通過信息共享和任務分工的方式協同工作,實現更復雜、更高效的問題解決能力。
Manus采用多代理架構(Multiple Agent Architecture),在獨立的虛擬環境中運行。其運轉邏輯可以概括為以下流程:
// todo.md
- [ ] 調研日本熱門旅游城市
- [ ] 收集交通信息
- [ ] 制定行程安排
- [ ] 預算規劃
任務初始化與環境準備:為確保任務執行的隔離性和安全性,系統創建獨立的執行環境:
# 創建任務目錄結構
mkdir -p {task_id}/
docker run -d --name task_{task_id} task_image
執行計劃制定:為每個子任務制定執行計劃,包括所需的工具和資源。歷史交互記錄在這一階段提供參考,幫助優化執行計劃。
自主執行:工具使用模塊在虛擬環境中自主執行各個子任務,包括搜索信息、檢索數據、編寫代碼、生成文檔和數據分析與可視化等。執行過程中的中間結果被記憶模塊保存,用于后續步驟。
系統采用多個專業化 Agent 協同工作,各司其職:
每個 Agent 的執行結果都會保存到任務目錄,確保可追溯性:
class SearchAgent:
def execute(self, task):
# 調用搜索 API
results = search_api.query(task.keywords)
# 模擬瀏覽器行為
browser = HeadlessBrowser()
for result in results:
content = browser.visit(result.url)
if self.validate_content(content):
self.save_result(content)
Search Agent: 負責網絡信息搜索,獲取最新、最相關的數據,采用混合搜索策略(關鍵詞+語義)
Code Agent: 處理代碼生成和執行,實現自動化操作,支持Python/JS/SQL等語言
Data Analysis Agent: 進行數據分析,提取有價值的洞見,Pandas/Matplotlib集成
動態質量檢測:
def quality_check(result):
if result.confidence < 0.7:
trigger_self_correction()
return generate_validation_report()
結果整合:將各個子任務的結果整合為最終輸出,確保內容的連貫性和完整性。
智能整合所有 Agent 的執行結果,消除冗余和矛盾
生成用戶友好的多模態輸出,確保內容的可理解性和實用性
結果交付:向用戶提供完整的任務結果,可能是報告、分析、代碼、圖表或其他形式的輸出。
用戶反饋與學習:用戶對結果提供反饋,這些反饋被記憶模塊記錄,用于改進未來的任務執行。強化模型微調,不斷提升系統性能。
1. manus核心優勢在Controller層
在agent flow能力(observe,plan和tool decide的大模型環節),大概率這些是使用自己調優訓練的大模型的,并且基礎模型大概率是qwen
2. manus的核心壁壘是數據
3. manus的AgentFlow有沒有可能使用了其他黑科技
4. manus不用MCP協議的原因
Manus具有多項技術特點,使其在AI代理領域脫穎而出:
1. 自主規劃能力
Manus能夠獨立思考和規劃,確保任務的執行,這是其與之前工具的主要區別。在GAIA基準測試(General AI Assistant Benchmark)中,Manus取得了最新的SOTA(State-of-the-Art)成績,這一測試旨在評估通用AI助手在現實世界中解決問題的能力。在復雜任務中實現94%的自動完成率。
2. 上下文理解
Manus能夠從模糊或抽象的描述中準確識別用戶需求。例如,用戶只需描述視頻內容,Manus就能在平臺上定位相應的視頻鏈接。這種高效的匹配能力確保了更流暢的用戶體驗。支持10輪以上的長對話維護。
3. 多代理協作
Manus采用多代理架構,類似于Anthropic的Computer Use功能,在獨立的虛擬機中運行。這種架構使不同功能模塊能夠協同工作,處理復雜任務。
4. 工具集成
Manus能夠自動調用各種工具,如搜索、數據分析和代碼生成,顯著提高效率。這種集成能力使其能夠處理各種復雜任務,從信息收集到內容創建,再到數據分析。支持自定義工具插件開發。
5. 安全隔離
基于gVisor的沙箱環境,確保任務執行的安全性和穩定性。
6. 其他技術優勢
系統的強大能力得益于多層次的模型協作:
場景類型 | 典型案例 | 輸出形式 |
旅行規劃 | 日本深度游定制 | 交互式地圖 + 預算表 |
金融分析 | 特斯拉股票多維分析 | 動態儀表盤 + 風險評估 |
教育支持 | 動量定理教學方案 | 互動式課件 + 實驗模擬 |
商業決策 | 保險產品對比分析 | 可視化對比矩陣 + 建議書 |
市場研究 | 亞馬遜市場情緒分析 | 季度趨勢報告 + 預測模型 |
優點:
缺點:
繼deepseek之后,武漢一個開發monica的團隊又開發了manus,號稱是全球第一個通用的agent!各路自媒體企圖復刻下一個deepseek,瘋狂報道!
然而manus發布后不久,metaGPT團隊5個工程師號稱耗時3小時就搞定了一個demo版本的manus,取名openManus,才幾天時間就收獲了34.4K的start,又火出圈了!現在研究一下openManus的核心原理!
所以agent誕生了!不管是deep search、deep research、manus等,核心思路都是一樣的:plan->action->review->action->review…… 如此循環下去,直到觸發結束的條件!大概的流程如下:
具體到openManus,核心的流程是這樣的:用戶輸入prompt后,有專門的agent調用LLM針對prompt做任務拆分,把復雜的問題拆解成一個個細分的、邏輯連貫的小問題,然后對于這些小問題,挨個調用tool box的工具執行,最后返回結果給用戶!
這類通用agent最核心的競爭力就兩點了:
4個文件夾,分別是agent、flow、prompt、tool,只看名字就知道這個模塊的功能了
整個程序入口肯定是各種agent啦!各大agent之間的關系如下:
(1)agent核心的功能之一不就是plan么,openManus的prompt是這么干的:promt中就直接說明了是expert plan agent,需要生成可執行的plan!
PLANNING_SYSTEM_PROMPT = """
You are an expert Planning Agent tasked with solving problems efficiently through structured plans.
Your job is:
1. Analyze requests to understand the task scope
2. Create a clear, actionable plan that makes meaningful progress with the planning
tool
3. Execute steps using available tools as needed
4. Track progress and adapt plans when necessary
5. Use finish
to conclude immediately when the task is complete
Available tools will vary by task but may include:
- planning
: Create, update, and track plans (commands: create, update, mark_step, etc.)
- finish
: End the task when complete
Break tasks into logical steps with clear outcomes. Avoid excessive detail or sub-steps.
Think about dependencies and verification methods.
Know when to conclude - don't continue thinking once objectives are met.
"""
NEXT_STEP_PROMPT = """
Based on the current state, what's your next action?
Choose the most efficient path forward:
1. Is the plan sufficient, or does it need refinement?
2. Can you execute the next step immediately?
3. Is the task complete? If so, use finish
right away.
Be concise in your reasoning, then select the appropriate tool or action.
"""
prompt有了,接著就是讓LLM對prompt生成plan了,在agent/planning.py文件中:
async def create_initial_plan(self, request: str) -> None:
"""Create an initial plan based on the request."""
logger.info(f"Creating initial plan with ID: {self.active_plan_id}")
messages = [
Message.user_message(
f"Analyze the request and create a plan with ID {self.active_plan_id}: {request}"
)
]
self.memory.add_messages(messages)
response = await self.llm.ask_tool(
messages=messages,
system_msgs=[Message.system_message(self.system_prompt)],
tools=self.available_tools.to_params(),
tool_choice=ToolChoice.AUTO,
)
assistant_msg = Message.from_tool_calls(
content=response.content, tool_calls=response.tool_calls
)
self.memory.add_message(assistant_msg)
plan_created = False
for tool_call in response.tool_calls:
if tool_call.function.name == "planning":
result = await self.execute_tool(tool_call)
logger.info(
f"Executed tool {tool_call.function.name} with result: {result}"
)
# Add tool response to memory
tool_msg = Message.tool_message(
content=result,
tool_call_id=tool_call.id,
name=tool_call.function.name,
)
self.memory.add_message(tool_msg)
plan_created = True
break
if not plan_created:
logger.warning("No plan created from initial request")
tool_msg = Message.assistant_message(
"Error: Parameter plan_id
is required for command: create"
)
self.memory.add_message(tool_msg)
plan生成后,就是think和act的循環啦!同理,這部分實現代碼在agent/toolcall.py中,如下:think的功能是讓LLM選擇干活的工具,act負責調用具體的工具執行
async def think(self) -> bool:
"""Process current state and decide next actions using tools"""
if self.next_step_prompt:
user_msg = Message.user_message(self.next_step_prompt)
self.messages += [user_msg]
# Get response with tool options:讓LLM選擇使用哪種工具干活
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
# Log response info
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"
)
if response.tool_calls:
logger.info(
f" Tools being prepared: {[call.function.name for call in response.tool_calls]}"
)
try:
# Handle different tool_choices modes
if self.tool_choices == ToolChoice.NONE:
if response.tool_calls:
logger.warning(
f" Hmm, {self.name} tried to use tools when they weren't available!"
)
if response.content:
self.memory.add_message(Message.assistant_message(response.content))
return True
return False
# Create and add assistant message
assistant_msg = (
Message.from_tool_calls(
content=response.content, tool_calls=self.tool_calls
)
if self.tool_calls
else Message.assistant_message(response.content)
)
self.memory.add_message(assistant_msg)
if self.tool_choices == ToolChoice.REQUIRED and not self.tool_calls:
return True # Will be handled in act()
# For 'auto' mode, continue with content if no commands but content exists
if self.tool_choices == ToolChoice.AUTO and not self.tool_calls:
return bool(response.content)
return bool(self.tool_calls)
except Exception as e:
logger.error(f" Oops! The {self.name}'s thinking process hit a snag: {e}")
self.memory.add_message(
Message.assistant_message(
f"Error encountered while processing: {str(e)}"
)
)
return False
async def act(self) -> str:
"""Execute tool calls and handle their results"""
if not self.tool_calls:
if self.tool_choices == ToolChoice.REQUIRED:
raise ValueError(TOOL_CALL_REQUIRED)
# Return last message content if no tool calls
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)#調用具體的工具干活
if self.max_observe:
result = result[: self.max_observe]
logger.info(
f" Tool '{command.function.name}' completed its mission! Result: {result}"
)
# Add tool response to memory
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)
think和act是循環執行的,直到滿足停止條件,這部分功能在agent/base.py實現的:
async def run(self, request: Optional[str] = None) -> str:
"""Execute the agent's main loop asynchronously.
Args:
request: Optional initial user request to process.
Returns:
A string summarizing the execution results.
Raises:
RuntimeError: If the agent is not in IDLE state at start.
"""
if self.state != AgentState.IDLE:
raise RuntimeError(f"Cannot run agent from state: {self.state}")
if request:
self.update_memory("user", request)
results: List[str] = []
async with self.state_context(AgentState.RUNNING):
while ( # 循環停止的條件:達到最大步數,或agent的狀態已經是完成的了
self.current_step < self.max_steps and self.state != AgentState.FINISHED
):
self.current_step += 1
logger.info(f"Executing step {self.current_step}/{self.max_steps}")
step_result = await self.step()
# Check for stuck state
if self.is_stuck():
self.handle_stuck_state()
results.append(f"Step {self.current_step}: {step_result}")
if self.current_step >= self.max_steps:
self.current_step = 0
self.state = AgentState.IDLE
results.append(f"Terminated: Reached max steps ({self.max_steps})")
return "\n".join(results) if results else "No steps executed"
既然是while循環迭代,那每次迭代又有啥不一樣的了?舉個例子:查找AI最新的新聞,并保存到文件中。第一次think,調用LLM的時候輸入用戶的prompt和相應的人設、能使用的tool,讓LLM自己選擇一個合適的tool,并輸出到response中!這里的LLM選擇了google search去查找新聞,并提供了google search的query!
第二次think,給LLM輸入的prompt帶上了第一輪的prompt和response,類似多輪對話,把多個context收集到一起作為這次的最新的prompt,讓LLM繼續輸出結果,也就是第三次的action是啥!
第三次think:同樣包含前面兩次的promt!但這次LLM反饋已經不需要調用任何工具了,所以這個query至此已經完全結束!
整個流程簡單!另外,用戶也可以添加自己的tool,只要符合MCP協議就行!
從外部來看,Manus(以及復刻的 OpenManus)本質上是一個多智能體系統(Multi-Agent System)。不同于單一大模型那種一次性”大而全”的回答方式,多智能體系統通過”規劃—執行—反饋”的循環,逐步解決復雜的真實世界問題。在 OpenManus 的設計中,最核心的思路可以概括為以下幾點:
OpenManus 的核心設計是構建一個非常精簡的 Agent 框架,強調模塊化和可擴展性。它通過可插拔的工具(Tools)和提示詞(Prompt)的組合來定義 Agent 的功能和行為,降低了開發和定制 Agent 的門檻。
通過對 Prompt 和 Tools 的自由組合,就能快速”拼裝”出新的 Agent,賦予其處理不同類型任務的能力。
OpenManus 基于 ReAct(Reason + Act)模式,并以工具為核心驅動 Agent 的行動。Prompt 引導 Agent 的推理和邏輯,而 Tools 則賦予 Agent 行動能力。ToolCall Agent 的引入,進一步提升了工具使用的效率和規范性。
OpenManus 延續了 Manus 的多智能體規劃優勢,將 PlanningTool 用于對用戶需求進行高層規劃。這種”先規劃,后執行”的思路在復雜、長鏈任務上效果更佳。PlanningTool 將復雜的用戶需求分解為線性的子任務計劃,這種規劃能力是處理現實世界復雜問題的關鍵。過去的研究表明,在相同模型能力下,如果缺乏系統的分解和規劃,許多真實問題的成功率會大打折扣;而加入規劃后,成功率會有顯著提升。
當一個任務拆解出若干子任務后,系統會根據子任務類型,動態將其分配給預先定義或適配的 Agent(有各自的工具集和能力傾向)。這種”臨時分配 + 工具協作”的機制,可以最大化利用多模型、多工具的組合優勢,提高應對不同問題場景的靈活度。Agent 預先裝備了不同的工具集以應對不同類型的任務,提高了系統的靈活性和效率。
OpenMan
us 的運行流程可以清晰概括為”規劃→分配→執行”,具體步驟如下:
用戶在前端或命令行中輸入復雜的需求,例如”寫一段代碼完成某種功能,并自動部署到服務器上”。
系統先調用 PlanningTool,對需求進行分析與分解,形成一個線性結構的計劃或任務序列。比如,會將需求拆解為:
這些子任務被記錄在一個 plan 或類似結構中。
如果任務中涉及大規模數據分析或機器學習流程,可能會調用一個具備 Data Interpreter 能力的 Agent;
若任務需要復雜的代碼修復或文件管理,則會調用另一個能夠使用 ComputerUse 工具的 Agent;
系統按照順序從計劃中依次取出子任務;
根據任務關鍵字或意圖判定,分配給最合適的 Agent。目前 Agent 分配主要基于正則匹配,未來考慮使用 LLM 實現更智能的任務分配。
每個 Agent 都會采用 ReAct 循環(Reason + Act)與 Tools 進行交互,以完成自己所負責的子任務。
當某個子任務執行完畢后,系統會將執行結果、關鍵上下文信息進行必要的”總結與壓縮”(以避免不斷增加的冗長 Memory),然后存入當前的”Plan 內存”或全局可訪問的共享內存。
如果任務完成順利,進入下一子任務;
若出現執行失敗或結果異常,系統可進行自動調試或重新規劃,視設計實現程度而定。
當所有子任務執行完畢,系統對整體結果進行匯總并返回給用戶,或完成如網頁部署、自動執行腳本等操作。
在這個過程中,多 Agent + 工具的結構會在復雜需求上展現明顯的優勢,尤其當需要長鏈思考、結合搜索或外部工具時,能夠更好地完成通用大模型難以一次性解決的工作。
項目依賴相對簡單,主要包括一些用于數據驗證(pydantic)、AI 服務調用(openai)、瀏覽器控制(playwright、browsergym、browser-use)和一些基礎工具庫:
這樣的結構設計使得 OpenManus 在提供強大功能的同時保持了極高的可維護性和可擴展性。
OpenManus 的架構由四個主要模塊構成:
Agent 模塊采用清晰的繼承層次,自底向上逐步增強功能:
示例代碼(Manus 實現):
class Manus(ToolCallAgent):
"""
A versatile general-purpose agent that uses planning to solve various tasks.
"""
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
# Add general-purpose tools to the tool collection
available_tools: ToolCollection = Field(
default_factory=lambda: ToolCollection(
PythonExecute(), GoogleSearch(), BrowserUseTool(), FileSaver(), Terminate()
)
)
工具模塊是 OpenManus 的行動能力基礎,各類工具均繼承自 BaseTool:
其中,planning.py 實現了 Manus 著名的計劃功能,用 Markdown 格式管理任務計劃并跟蹤執行進度。
Prompt 模塊包含了各種 Agent 使用的指令模板,例如 Planning 的系統提示:
PLANNING_SYSTEM_PROMPT = """
You are an expert Planning Agent tasked with solving complex problems by creating and managing structured plans.
Your job is:
1. Analyze requests to understand the task scope
2. Create clear, actionable plans with the planning
tool
3. Execute steps using available tools as needed
4. Track progress and adapt plans dynamically
5. Use finish
to conclude when the task is complete
Available tools will vary by task but may include:
- planning
: Create, update, and track plans (commands: create, update, mark_step, etc.)
- finish
: End the task when complete
Break tasks into logical, sequential steps. Think about dependencies and verification methods.
"""
而 Manus 的系統提示則更加簡潔:
SYSTEM_PROMPT = "You are OpenManus, an all-capable AI assistant, aimed at solving any task presented by the user.
You have various tools at your disposal that you can call upon to efficiently complete complex requests.
Whether it's programming, information retrieval, file processing, or web browsing, you can handle it all."
Flow 模塊負責任務的高層編排和執行流程管理:
PlanningFlow 的執行流程:
每步執行前,系統會生成上下文豐富的提示:
step_prompt = f"""
CURRENT PLAN STATUS:
{plan_status}
YOUR CURRENT TASK:
You are now working on step {self.current_step_index}: "{step_text}"
Please execute this step using the appropriate tools. When you're done, provide a summary of what you accomplished.
"""
BaseFlow:抽象基類,定義了 Agent 管理和執行接口
PlanningFlow:實現基于規劃的執行策略
創建初始計劃(_create_initial_plan)
按計劃步驟調用適當的 Agent
跟蹤計劃執行狀態并動態調整
值得注意的是,在當前版本中,雖然 PlanningFlow 具備多智能體調度的能力,但實際上只有單一的 Manus 智能體在執行任務。未來版本可引入更多專業化的 Agent 以充分發揮多智能體協作的優勢。
與前文描述一致,OpenManus 實現了簡單但有效的記憶管理和 Agent 分配機制:
在AI領域,開源項目正逐漸成為推動技術發展的重要力量。OWL Agent,一個由CAMEL-AI團隊推出的開源AI智能體項目,不僅完全復刻了Manus的核心功能,還在靈活性和開源生態上實現了超越。深入了解OWL Agent如何幫助你零成本打造全能的開源AI打工人。
OWL 的多智能體協作機制通過分層架構和模塊化設計實現高效協作。它的核心組件包括 BaseAgent、ChatAgent、RolePlaying、Workforce 以及 Task 相關 Agent 等,這些組件各司其職,共同完成任務分解、角色分配和任務執行等功能。
OWL 的多智能體協作機制主要基于以下幾個核心組件:
這種設計使 OWL 能夠高效處理復雜任務,動態調整任務角色分配,提升多智能體間的協作效率,同時具備自適應學習和優化能力,滿足多樣化的應用需求。
OWL將Manus的核心工作流拆解為以下六步:
為了實現Agent的遠程操作,OWL配備了強大的Ubuntu Toolkit,支持以下功能:
與Manus類似,OWL也具備記憶功能,能夠實時存儲新知識,并在任務中召回過往經驗。這使得OWL在處理類似任務時更加高效。
在Manus爆火之前,CAMEL-AI已經開發了CRAB——一套強大的跨平臺操作系統通用智能體。CRAB不僅能操控Ubuntu容器,還能直接控制手機和電腦中的任何應用。未來,CRAB技術將融入OWL,實現跨平臺、多設備、全場景的遠程操作。
在AI領域,開源的力量是無窮的。OWL項目不僅在0天內復刻了Manus的核心功能,還通過開源模式吸引了全球開發者的參與。它不僅性能卓越,還具備高度的靈活性和擴展性。
維度 | OWL | 開放手冊 |
執行環境 | Docker 容器 + 原生系統穿透 | 本地沙箱環境 |
任務復雜度 | 支持多設備聯動任務 | 單設備線性任務 |
記憶系統 | 增量式知識圖譜(支持版本回溯) | 臨時記憶池(任務級隔離) |
資源消耗 | 單任務平均 8 萬 tokens | 單任務峰值 24 萬 tokens |
擴展性 | 插件市場 + 自定義工具鏈 | 固定模塊組合 |
OWL Agent作為一個開源AI智能體項目,不僅在性能上達到了行業領先水平,還在成本和靈活性上具有顯著優勢。它為開發者和用戶提供了一個零成本、高性能的AI工具,能夠滿足多種應用場景的需求。
【金融場景:搜索+知識庫rag】
大致可以理解成多個Agent進程/線程會并行工作,相互之間通過某些機制進行溝通(例如消息隊列)
典型案例:metagpt的多角色協同(產品、后端、前端、測試),斯坦福小鎮
不出現并行工作的,不是Multi-agent;coze上所謂的“Multi-agent”就是典型的錯誤概念,因為它只是不同的agent之間串行流轉。
通常需要較長的步驟或者較多的時間才能完成,需要進行agentflow編排。
Copilot 類。
Agentic 類。
【autogpt】自動 GPT
文章轉載自:AI Agent 全面對比:Manus、OpenManus與OWL