openmanus

function call，OpenAI没有开源数据构造格式，这种能力也是通过数据训练出来的。

config.toml (参考阿里云百炼，其他的API调用基本都是类似的思路，在公司记得加http_client=httpx.AsyncClient(verify=False))

# Global LLM configuration
[llm]
model = "qwen-long"        # The LLM model to use
base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1"  # API endpoint URL
api_key = "sk-xxxxxx"                    # Your API key
max_tokens = 8192                           # Maximum number of tokens in the response
temperature = 0.0                           # Controls randomness

两种模式，分别是稳定版本和开发版本，对应main.py和run_flow.py

main.py调用链路

main.py
    -->manus.run()
        -->while
            -->step()
                -->think()  manus.think()-->ToolCallAgent.think()
                -->act()	ToolCallAgent.act()
            -->self.is_stuck()
        -->return results

manus.think()
    -->self.get_browser_state() 初次是只有一个ToolResult.error
        -->super.think()  ToolCallAgent.think()
            -->self.llm.ask_tool
                -->self.client.chat.completions.create(**params).choices[0].message
            -->self.memory.add_message(assistant_msg)
        -->ToolCallAgent.act()
            for loop
                -->self.execute_tool(command)
                    -->json.loads
                    -->self.available_tools.execute
                    -->PythonExecute.execute  开启子进程+timeout
                        -->PythonExecute._run_code
                    -->self._handle_special_tool  判断是不是特殊工具，只有terminate是
                -->self.memory.add_message(tool_msg)

run_flow.py调用链路

PlanningFlow(agents=agents) 构造函数完成赋值，注意data字典在不断扩充，最终调用baseModel的构造完成赋值
PlanningFlow.execute(prompt)
    -->self._create_initial_plan(input_text)
        -->self.llm.ask_tool
        -->self.planning_tool.execute(**args)
        -->self.plans[plan_id]进行一系列修改
    -->while
        -->self._get_current_step_info()
            -->return i, step_info   这里返回每次第一个未完成的index和step_info，step_info就是包含steps[i]的字典
        -->self.get_executor
        -->self._execute_step
    -->return results

self._execute_step
    -->self._get_plan_text()
    -->generate step_prompt
    -->executor.run(step_prompt)  其实就是agent.run,对应前面写的manus.run,这里会进行一系列的step，直到finish
    -->self._mark_step_completed()

值得注意的点

每次调用self.llm.ask_tool的时候，有一个参数是self.messages，这玩意其实就是self.memory.messages，那么一切就解释的通了，self.memory维护所有的历史对话记录
第一次think里面拿到的response.content为空，因为第一次只选工具
文本tokens计算直接调用tiktoken计算，图像tokens计算采用固定的公式（未开源）
self._get_current_step_info()拿到的step_info会根据[]去提取type字段，然后决定用哪一个agent，否则用默认的
项目中的字典查看太麻烦，其实是python的字典，用black工具格式化看
think的每一轮，都会更新memory，在开头加user的message，在末尾加assistant的message
在执行async with self.state_context(AgentState.RUNNING)的时候，内部self.step()可能会将self.state设置为AgentState.FINISHED，但是循环仍然继续，猜测就是这个异步上下文在起作用，保证状态不发生变化！比如在执行run_flow.py的时候，会发现明明finished了，还是会继续执行？因为在main.py中只有一个agent.run调用，但是在run_flow.py中，每一步中，都会调用agent.run，在run里面上下文管理器重新设置state为AgentState.RUNNING，如果中间agent改为AgentState.FINISHED了，就直接退出循环。但是不管内部做了什么修改，在离开上下文管理器的时候，在finally部分会重新恢复state为AgentState.IDLE。谈一谈Python的上下文管理器 | 思诚之道
前面提到finally恢复state为AgentState.IDLE，但是flow的结束判断也是AgentState.FINISHED啊？那么flow岂不是一直退出不了？flow的while循环有两个break点，打日志发现是第一个break退出的。第二个break暂时不清楚。如果第一个break注释掉，后面会捕获错误，然后也能退出

整体逻辑：

1
2
3

prompt-->steps
    -->for step in steps
        -->think-->act-->think-------until finish

Config模块

app/config.py

这块没什么好说的，就是解析toml文件

llm.py：tokens数量计算、format_messages消息格式的转换，区分了两种type，text和image_url，ask添加retry重连机制等，ask默认是Steam流式处理，流式和非流式有不同的处理方式。ask_with_images和ask_tool都是类似的处理方式，但是tool不支持流式。

openManus虽然只有基础的功能，但是agent框架、API调用、Pytantic校验都值得学习。

就是要摸鱼

openManus解读

openmanus

main.py调用链路

run_flow.py调用链路

Config模块