近日，國(guó)產(chǎn)編程語(yǔ)言MoonBit補(bǔ)全了關(guān)鍵語(yǔ)言特性的最后一塊拼圖：異步編程庫(kù)moonbitlang/async。

本次發(fā)布時(shí)間距離MoonBit Beta Release相距僅僅半年，足見(jiàn)MoonBit團(tuán)隊(duì)對(duì)異步編程的重視。

moonbitlang/async吸收了現(xiàn)有語(yǔ)言的經(jīng)驗(yàn)和教訓(xùn)，語(yǔ)法更加簡(jiǎn)潔，基于結(jié)構(gòu)化并發(fā)理念，幫助用戶(hù)寫(xiě)出更健壯、安全的異步程序。未來(lái)很可能「占領(lǐng)」包括云服務(wù)、AI agent 等重度依賴(lài)異步編程的領(lǐng)域。

0 1

什么是異步編程？

你開(kāi)了一家飯店，雇傭了5個(gè)店小二來(lái)招待顧客，但是這幾個(gè)店小二的干活兒的模式一模一樣：

客人來(lái)到飯店，馬上有個(gè)店小二殷勤迎上去，帶著找座位，點(diǎn)菜，給后廚下單。

由于后廚做菜需要很長(zhǎng)時(shí)間，店小二就在客人的旁邊等著。

后廚一搖鈴鐺，大喊一聲：上菜，店小二馬上端到客人面前， 然后站在一邊等著客人吃完。

客人說(shuō)：結(jié)賬，小二收錢(qián)，找錢(qián)，送客， 迎接下一位。

由于只有5個(gè)店小二，你飯店同時(shí)只能招待5個(gè)顧客。

很快，你的飯店倒閉了。

倒閉的核心原因就是店小二采用的是“同步模式”，即使有耗時(shí)的工作（廚師做菜，顧客吃飯），他也會(huì)干等著，非常浪費(fèi)。

你接受了教訓(xùn)，開(kāi)了一家新飯店，這次只雇傭了一個(gè)店小二，他的工作方式和之前大相徑庭：

客人來(lái)到飯店，唯一的店小二殷勤迎上去，帶著找座位，點(diǎn)菜，給后廚下單

由于后廚做菜需要很長(zhǎng)時(shí)間，店小二閃電般的離開(kāi)，去干別的活了，可能是迎客，點(diǎn)菜，找座等，總之是那些不用等待，迅速干完的活。

后廚大喊一聲：上菜，這個(gè)小二馬上端到客人面前，然后離開(kāi)，干其他活。

客人說(shuō)：結(jié)賬，小二收錢(qián)，找錢(qián)，然后還是迅速閃人，干其他活。

這一次，店小二采用的是“異步模式”，即對(duì)于耗時(shí)的操作，店小二會(huì)暫時(shí)離開(kāi)，做其他事兒，等到操作完成以后再回來(lái)接著干。

對(duì)應(yīng)到計(jì)算機(jī)世界，耗時(shí)的操作就是訪問(wèn)文件/數(shù)據(jù)庫(kù)/網(wǎng)絡(luò)，網(wǎng)絡(luò)服務(wù)器的線程遇到了這些I/O操作，堅(jiān)決不能等待，因?yàn)榉?wù)器收到的請(qǐng)求可不是幾十個(gè)幾百個(gè)，而是成千上萬(wàn)個(gè)，所以一定要采用異步編程。

但是對(duì)程序員來(lái)說(shuō)，異步編程很麻煩，為了支持任務(wù)的中斷和切換，程序的邏輯會(huì)被分散到程序的不同部分，使得開(kāi)發(fā)效率和程序的可維護(hù)性極大下降。

所以各種編程語(yǔ)言Go/Rust/Python/JS都在語(yǔ)言層面直接支持異步編程，降低程序員的負(fù)擔(dān)，MoonBit也不例外。

0 2

MoonBit 異步性能優(yōu)勢(shì)

MoonBit 的異步運(yùn)行時(shí)在底層基于線程池并結(jié)合epoll/kqueue實(shí)現(xiàn)，支持 Linux 與 macOS 的 native 后端。其設(shè)計(jì)思路與 Node.js 類(lèi)似：采用單線程、多任務(wù)模型。

在這一模式下，異步程序中的同步部分始終在同一線程上執(zhí)行。對(duì)開(kāi)發(fā)者而言，這帶來(lái)顯著的簡(jiǎn)化效果：程序的行為與單線程應(yīng)用一致，無(wú)需額外加鎖，也不必?fù)?dān)心競(jìng)爭(zhēng)條件等并發(fā)錯(cuò)誤。

雖然仍處于早期階段，這一運(yùn)行時(shí)已經(jīng)展現(xiàn)出出色的性能表現(xiàn)。

為了檢驗(yàn) MoonBit 異步運(yùn)行時(shí)的性能，我們搭建了一個(gè)簡(jiǎn)單的 TCP 服務(wù)器：它會(huì)把收到的數(shù)據(jù)原樣返回給客戶(hù)端。這個(gè)測(cè)試幾乎沒(méi)有計(jì)算成分，因此能夠直接反映運(yùn)行時(shí)在高并發(fā)場(chǎng)景下的處理能力。

在測(cè)試中，我們同時(shí)維持多個(gè)連接，不斷收發(fā)數(shù)據(jù)，并記錄吞吐量和響應(yīng)延遲。結(jié)果顯示，MoonBit 在并發(fā)連接數(shù)不斷增加的情況下，依然保持了優(yōu)異的吞吐表現(xiàn)和極低的響應(yīng)延遲，充分體現(xiàn)了其運(yùn)行時(shí)系統(tǒng)的高效性和穩(wěn)定性。

對(duì)比的對(duì)象是 Node.js 和 Go 語(yǔ)言。

性能測(cè)試的結(jié)果如下：

測(cè)試結(jié)果顯示，MoonBit 在 200 到 1000 個(gè)并發(fā)連接下始終保持最高吞吐量，在高并發(fā)場(chǎng)景中明顯優(yōu)于 Node.js 和 Go。這表明其異步運(yùn)行時(shí)具備出色的擴(kuò)展性。

在高并發(fā)場(chǎng)景下，MoonBit 的平均延遲始終保持在個(gè)位數(shù)毫秒，即便在 1000 個(gè)連接時(shí)也只有 4.43ms；相比之下，Node.js 延遲超過(guò) 116ms。這意味著 MoonBit 的異步運(yùn)行時(shí)能夠在大規(guī)模連接下依然保持快速響應(yīng)。

下面是一個(gè) HTTP 服務(wù)器的例子，相比 TCP 服務(wù)器，HTTP 例子需要進(jìn)行 HTTP 協(xié)議的解析，有更多的計(jì)算成分，不是單純的 I/O。

這個(gè)測(cè)試會(huì)使用 (github.com/wg/wrk) 工具，通過(guò)多個(gè)連接不斷向 HTTP 服務(wù)器發(fā)送 GET / HTTP/1.1 的請(qǐng)求，服務(wù)器應(yīng)當(dāng)返回一個(gè)空的回復(fù)。測(cè)試會(huì)記錄服務(wù)器每秒處理的請(qǐng)求數(shù)以及每個(gè)請(qǐng)求的平均延遲。測(cè)試的結(jié)果如下：

可以看到，得益于 MoonBit 語(yǔ)言本身的優(yōu)秀性能，在這個(gè)測(cè)試中 MoonBit 依然表現(xiàn)良好。

MoonBit 在所有并發(fā)連接數(shù)下的請(qǐng)求處理效率和延遲都穩(wěn)定高于 Node.js 和單線程的 Go。

0 3

構(gòu)建簡(jiǎn)單代碼智能體的示例

MoonBit 不只是寫(xiě)服務(wù)器更方便,它甚至能直接驅(qū)動(dòng) AI 智能體。下面，我們就用它構(gòu)建一個(gè)最小可運(yùn)行的代碼智能體（Code Agent）。

這個(gè)代碼智能體除了可以調(diào)用大模型，還支持工具調(diào)用（如讀取本地文件，執(zhí)行l(wèi)s命令等）。

例如，用戶(hù)的請(qǐng)求是：請(qǐng)讀取本地文件 /home/user/data.txt 并告訴我里面的內(nèi)容。

代碼智能體會(huì)把這個(gè)消息發(fā)給大模型，并且告訴大模型，我這里有兩個(gè)工具可以調(diào)用，工具的名稱(chēng)，參數(shù)也給你發(fā)過(guò)去了。

大模型看到看到“請(qǐng)讀取本地文件......”，它當(dāng)然不會(huì)直接讀取文件，而是看看根據(jù)智能體都發(fā)來(lái)了什么樣的工具，然后發(fā)揮自己的強(qiáng)項(xiàng)，選擇對(duì)應(yīng)的工具，生成調(diào)用請(qǐng)求：

{
  "tool_calls": [
    {
      "function": {
        "name": "read_file"
      },
      "arguments": {
        "path": "/home/user/data.txt"
      }
    }
  ]
}

智能體收到大模型發(fā)回的工具調(diào)用請(qǐng)求，執(zhí)行真正的工具調(diào)用，讀取 /home/user/data.txt，假設(shè)結(jié)果是：MoonBit is the future programming language!

智能體會(huì)將結(jié)果包裝成消息，發(fā)送給大模型模型，大模型收到工具返回的內(nèi)容后，會(huì)判斷：“我已經(jīng)得到了文件內(nèi)容，不需要再調(diào)用工具了，我可以生成最終回答”

最終響應(yīng)可能是這樣的：

{
  "role": "assistant",
  "content": "我已經(jīng)讀取了文件 /home/user/data.txt，里面的內(nèi)容是：\nMoonBit is the future programming language!"
}

在這個(gè)代碼智能體中，需要處理網(wǎng)絡(luò)調(diào)用，文件讀取，命令執(zhí)行，會(huì)使用MoonBit的這些異步操作：

1. @http.post 發(fā)送消息到 LLM 接口。

2. @fs.read_file 從文件讀取內(nèi)容。

3. @process.collect_output_merged 來(lái)執(zhí)行外部程序并收集其輸出。

值得注意的是，在MoonBit中所有異步函數(shù)調(diào)用默認(rèn)會(huì)被隱式 await，并且異步調(diào)用實(shí)現(xiàn)了結(jié)構(gòu)化并發(fā)（Structured Concurrency） ，這意味著MoonBit 的異步程序幾乎不可能產(chǎn)生僵尸后臺(tái)任務(wù)，并且程序員能夠更加容易地理解并分析異步代碼的行為。

1、向 LLM 接口發(fā)起請(qǐng)求

MoonBit 異步網(wǎng)絡(luò)庫(kù)提供了 @http.post 用于發(fā)送 HTTP POST 請(qǐng)求。我們可以簡(jiǎn)單地將其包裝一下，用來(lái)更方便地發(fā)送消息到 LLM：

///|
async fn generate(request : Request) -> Response {
  let (response, body) = @http.post(
    "\{base_url}/chat/completions",
    request.to_json(),
    headers={
      "Authorization": "Bearer \{api_key}",
      "Content-Type": "application/json",
      "Connection": "close",
    },
  )
  guard response.code is (200..=299) else {
    fail("HTTP request failed: \{response.code} \{response.reason}")
  }
  body.json() |> @json.from_json()
}

接下來(lái)，我們將展示如何讓 LLM 使用工具。

2、定義工具

為了讓代碼智能體更有用，我們需要通過(guò)工具擴(kuò)展它與外部世界交互的能力。

請(qǐng)求體中的 "tools" 字段描述了我們向 LLM 提供的工具。一個(gè)典型的工具描述包含以下字段：

• name：工具名稱(chēng)，將在工具調(diào)用中使用。

• description：對(duì)工具的簡(jiǎn)短描述。

• parameters：描述工具參數(shù)的 JSON Schema。本示例中為簡(jiǎn)化處理，我們只使用 type、properties 和 required 字段。

例如，下面的 JSON 描述了一個(gè)名為 read_file 的工具：

{
  "name": "read_file",
  "description": "Read a file from local disk",
  "parameters": {
    "type": "object",
    "properties": {
      "path": {
        "type": "string",
        "description": "The path of the file to read"
      }
    },
    "required": ["path"]
  }
}

我們?cè)?MoonBit 中將該工具描述建模為如下結(jié)構(gòu)：

///|
struct Tool {
  name : String
  description : String
  parameters : Json
  /// 執(zhí)行工具的函數(shù)
  execute : async (String) -> String
}

在本演示中，我們將定義兩個(gè)簡(jiǎn)單工具：

read_file：從本地磁盤(pán)讀取文件。

execute_command：執(zhí)行一個(gè)外部程序。

3、read_file 工具

使用 moonbitlang/async 與文件系統(tǒng)交互非常簡(jiǎn)單?？梢灾苯邮褂?@fs.read_file/@fs.write_file 來(lái)進(jìn)行對(duì)文件的讀取/寫(xiě)入。對(duì)于更加靈活的需求，moonbitlang/async 也提供了 @fs.open ，用戶(hù)可以傳入自定義選項(xiàng)，并在后續(xù)調(diào)用 read / write 方法進(jìn)行 I/O 操作。

我們可以將 read_file 工具實(shí)現(xiàn)為：

///|
let read_file_tool : Tool = {
  name: "read_file",
  description: "Read a file from local disk",
  parameters: {
    "type": "object",
    "properties": {
      "path": {
        "type": "string",
        "description": "The path of the file to read",
      },
    },
    "required": ["path"],
  },
  execute: args => {
    guard @json.parse(args) is { "path": String(path), .. } else {
      fail("Invalid arguments for read_file, expected {\"path\": String}")
    }
    @moonbitlang/async/fs.read_file(path).text()
  },
}

4、execute_command 工具

在 moonbitlang/async 中實(shí)現(xiàn) execute_command 工具也非常簡(jiǎn)單。我們可以使用 @process.collect_output_merged 來(lái)執(zhí)行一個(gè)外部程序，并收集其 stdout 和 stderr 輸出。

對(duì)于更高級(jí)的需求，我們可以使用 @process.run 來(lái)啟動(dòng)一個(gè)進(jìn)程，并通過(guò)管道（pipe）與其交互。

execute_command 工具實(shí)現(xiàn)如下：

///|
let execute_command_tool : Tool = {
  name: "execute_command",
  description: "Execute an external program",
  parameters: {
    "type": "object",
    "properties": {
      "command": { "type": "string", "description": "The command to execute" },
      "arguments": {
        "type": "array",
        "items": { "type": "string" },
        "description": "The arguments to pass to the command",
      },
    },
    "required": ["command", "arguments"],
  },
  execute: arguments => {
    guard @json.parse(arguments)
      is { "command": String(command), "arguments": arguments, .. } else {
      fail(
        "Invalid arguments for execute_command, expected {\"command\": String, \"args\": Array[String]}",
      )
    }
    let arguments : Array[String] = @json.from_json(arguments)
    let (status, output) = @process.collect_output_merged(
      command,
      arguments.map(argument => argument),
    )
    let output = output.text()
    (
      $|Exit status: \{status}
      $|Output:
      $|\{output}
    )
  },
}

5、處理工具調(diào)用與智能體主循環(huán)

得到大模型發(fā)回的工具調(diào)用請(qǐng)求以后，代碼智能體需要進(jìn)行處理，使用的是這個(gè)異步函數(shù)：

///|
async fn handle_tool_call(
  tools : Map[String, Tool],
  tool_call : ToolCall,
) -> Json {
  guard tools.get(tool_call.function.name) is Some(tool) else {
    return {
      "role": "tool",
      "content": "Tool not found: \{tool_call.function.name}",
      "tool_call_id": tool_call.id,
    }
  }
  return {
    "role": "tool",
    "content": (tool.execute)(tool_call.function.arguments),
    "tool_call_id": tool_call.id,
  } catch {
    error =>
      {
        "role": "user",
        "content": "Error executing tool \{tool_call.function.name}: \{error}",
      }
  }
}

有了處理工具調(diào)用的能力后，我們就可以實(shí)現(xiàn)智能體的主循環(huán)了。我們定義了一個(gè) Agent 結(jié)構(gòu)來(lái)保存智能體狀態(tài)，包括工具集合、對(duì)話歷史和消息隊(duì)列：

///|
struct Agent {
  tools : Map[String, Tool]
  conversation : Array[Json]
  mut message_queue : Array[Json]
}

然后我們?yōu)?Agent 實(shí)現(xiàn) run 方法，持續(xù)處理消息隊(duì)列中的消息，直到隊(duì)列為空：

///|
async fn Agent::run(self : Agent) -> Unit {
  while !self.message_queue.is_empty() {
    // Take all messages from the message queue
    let messages = self.message_queue
    self.message_queue = []
    // Send the messages to LLM endpoint
    let response = generate({
      model,
      messages: [..self.conversation, ..messages],
      tools: self.tools.values().collect(),
    })
    let response = response.choices[0].message
    // Save the response to the conversation history
    self.conversation.push(response)
    if response is { "content": String(content), .. } {
      // Print the assistant's response
      println("Assistant: \{content}")
    }
    let tool_calls : Array[ToolCall] = if response
      is { "tool_calls": tool_calls, .. } {
      @json.from_json(tool_calls)
    } else {
      []
    }
    // Handle tool calls
    for tool_call in tool_calls {
      let message = handle_tool_call(self.tools, tool_call)
      self.message_queue.push(message)
      println("Tool: \{tool_call.function.name}")
      println("Response: \{message.stringify(indent=2)}")
    }
  }
}

大功告成，接下來(lái)測(cè)試一下。

讓這個(gè)智能體獲取當(dāng)前時(shí)間，并把結(jié)果告訴我們：

///|
async test "agent/current-time" {
  let agent = Agent::{
    tools: {
      "read_file": read_file_tool,
      "execute_command": execute_command_tool,
    },
    conversation: [],
    message_queue: [],
  }
  agent.message_queue.push({
    "role": "user",
    "content": "Can you please tell me what time is it now?",
  })
  agent.run()
}

0 4

結(jié)論

在這篇文章中，我們展示了如何使用 moonbitlang/async 構(gòu)建一個(gè)簡(jiǎn)單的代碼智能體。該智能體可以通過(guò)調(diào)用工具從本地磁盤(pán)讀取文件并執(zhí)行外部程序。當(dāng)然，這只是一個(gè)基礎(chǔ)示例，市面上的智能體通常會(huì)更加復(fù)雜，例如會(huì)添加更多工具、更優(yōu)雅地處理錯(cuò)誤、實(shí)現(xiàn)更復(fù)雜的對(duì)話流程等。

如果你想了解 moonbitlang/async 的更多信息，請(qǐng)參閱其文檔。你也可以查看 maria 項(xiàng)目源碼，了解我們是如何基于 moonbitlang/async 構(gòu)建代碼智能體的。

(1) MoonBit 再添異步能力，實(shí)現(xiàn) AI Agent 高效與穩(wěn)定開(kāi)發(fā)：

https://mp.weixin.qq.com/s/t5k9bUmuE-rs3qaGB0yLVw

(2) AI Agent 案例完整代碼：

https://gist.github.com/tonyfettes/2953d5bef1610fce12cca05ea20655e2