Python 中的微智慧體:一個由 MCP 驅動的約 70 行程式碼的智慧體
受 JS 中的微智慧體 的啟發,我們將其理念移植到了 Python 🐍,並擴充套件了
huggingface_hub
客戶端 SDK,使其可以作為 MCP 客戶端,從 MCP 伺服器拉取工具,並在推理過程中將它們傳遞給大語言模型。
MCP (模型上下文協議) 是一個開放協議,它標準化了大型語言模型(LLM)與外部工具和 API 的互動方式。從本質上講,它消除了為每個工具編寫自定義整合的需要,使向您的 LLM 中插入新功能變得更加簡單。
在這篇博文中,我們將向您展示如何開始使用連線到 MCP 伺服器的微型 Python 智慧體,以解鎖強大的工具能力。您將看到啟動自己的智慧體並開始構建是多麼容易!
劇透:智慧體本質上是一個直接構建在 MCP 客戶端之上的 `while` 迴圈!
如何執行演示
本節將引導您瞭解如何使用現有的微智慧體。我們將介紹設定和執行智慧體的命令。
首先,您需要安裝最新版本的 huggingface_hub
,並帶有 mcp
額外依賴,以獲取所有必要的元件。
pip install "huggingface_hub[mcp]>=0.32.0"
現在,讓我們使用命令列介面(CLI)來執行一個智慧體!
最酷的部分是,您可以直接從 Hugging Face Hub 的 tiny-agents 資料集載入智慧體,或者指定一個指向您自己本地智慧體配置的路徑!
> tiny-agents run --help
Usage: tiny-agents run [OPTIONS] [PATH] COMMAND [ARGS]...
Run the Agent in the CLI
╭─ Arguments ───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│ path [PATH] Path to a local folder containing an agent.json file or a built-in agent stored in the 'tiny-agents/tiny-agents' Hugging Face dataset │
│ (https://huggingface.co/datasets/tiny-agents/tiny-agents) │
╰───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ Options ─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│ --help Show this message and exit. │
╰───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
如果您沒有提供特定智慧體配置的路徑,我們的微智慧體將預設連線到以下兩個 MCP 伺服器:
- “規範的”檔案系統伺服器,用於訪問您的桌面檔案,
- 以及 Playwright MCP 伺服器,它知道如何為您使用一個沙盒化的 Chromium 瀏覽器。
以下示例展示了一個網頁瀏覽智慧體,它被配置為透過 Nebius 推理提供商使用 Qwen/Qwen2.5-72B-Instruct 模型,並配備了一個 Playwright MCP 伺服器,使其能夠使用網頁瀏覽器!該智慧體配置透過指定其在 Hugging Face 資料集 tiny-agents/tiny-agents
中的路徑進行載入。
當您執行智慧體時,您會看到它載入,並列出從其連線的 MCP 伺服器發現的工具。然後,它就準備好接收您的提示了!
此演示中使用的提示
在 Brave 搜尋上進行關於 HF 推理提供商的網頁搜尋,開啟第一個結果,然後給我 Hugging Face 支援的推理提供商列表
您還可以使用 Gradio Spaces 作為 MCP 伺服器!以下示例透過 Nebius 推理提供商使用 Qwen/Qwen2.5-72B-Instruct 模型,並連線到一個 FLUX.1 [schnell]
影像生成 HF Space 作為 MCP 伺服器。該智慧體從其在 Hugging Face Hub 上的 tiny-agents/tiny-agents 資料集中的配置載入。
此演示中使用的提示
生成一張 1024x1024 的圖片,內容是一個微型宇航員在月球表面從蛋裡孵化出來。
現在您已經瞭解瞭如何執行現有的微智慧體,接下來的部分將深入探討它們的工作原理以及如何構建您自己的智慧體。
智慧體配置
每個智慧體的行為(其預設模型、推理提供商、要連線的 MCP 伺服器以及其初始系統提示)都由一個 agent.json
檔案定義。您還可以在同一目錄中提供一個自定義的 PROMPT.md
檔案,以獲得更詳細的系統提示。下面是一個示例:
agent.json
檔案中的 model
和 provider
欄位指定了智慧體使用的 LLM 和推理提供商。servers
陣列定義了智慧體將連線的 MCP 伺服器。在此示例中,配置了一個 "stdio" MCP 伺服器。這種型別的伺服器作為本地程序執行。智慧體使用指定的 command
和 args
啟動它,然後透過標準輸入/輸出(stdin/stdout)與其通訊,以發現和執行可用的工具。
{
"model": "Qwen/Qwen2.5-72B-Instruct",
"provider": "nebius",
"servers": [
{
"type": "stdio",
"command": "npx",
"args": ["@playwright/mcp@latest"]
}
]
}
PROMPT.md
You are an agent - please keep going until the user’s query is completely resolved [...]
您可以在此處找到有關 Hugging Face 推理提供商的更多詳細資訊。
大語言模型可以使用工具
現代大語言模型(LLM)專為函式呼叫(或工具使用)而構建,這使得使用者可以輕鬆構建針對特定用例和現實世界任務的應用程式。
函式由其模式(schema)定義,該模式告知 LLM 函式的功能以及它期望的輸入引數。LLM 決定何時使用工具,然後由智慧體協調執行該工具並將結果反饋給 LLM。
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current temperature for a given location.",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City and country e.g. Paris, France"
}
},
"required": ["location"],
},
}
}
]
InferenceClient
實現了與 OpenAI 聊天補全 API 相同的工具呼叫介面,這是推理提供商和社群公認的標準。
構建我們的 Python MCP 客戶端
MCPClient
是我們工具使用功能的核心。它現在是 huggingface_hub
的一部分,並使用 AsyncInferenceClient
與 LLM 進行通訊。
完整的
MCPClient
程式碼在這裡,如果您想跟著實際程式碼一起學習 🤓
MCPClient
的主要職責
- 管理與一個或多個 MCP 伺服器的非同步連線。
- 從這些伺服器發現工具。
- 為 LLM 格式化這些工具。
- 透過正確的 MCP 伺服器執行工具呼叫。
這裡簡要展示了它如何連線到 MCP 伺服器(add_mcp_server
方法)
# Lines 111-219 of `MCPClient.add_mcp_server`
# https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/inference/_mcp/mcp_client.py#L111:L219
class MCPClient:
...
async def add_mcp_server(self, type: ServerType, **params: Any):
# 'type' can be "stdio", "sse", or "http"
# 'params' are specific to the server type, e.g.:
# for "stdio": {"command": "my_tool_server_cmd", "args": ["--port", "1234"]}
# for "http": {"url": "http://my.tool.server/mcp"}
# 1. Establish connection based on type (stdio, sse, http)
# (Uses mcp.client.stdio_client, sse_client, or streamablehttp_client)
read, write = await self.exit_stack.enter_async_context(...)
# 2. Create an MCP ClientSession
session = await self.exit_stack.enter_async_context(
ClientSession(read_stream=read, write_stream=write, ...)
)
await session.initialize()
# 3. List tools from the server
response = await session.list_tools()
for tool in response.tools:
# Store session for this tool
self.sessions[tool.name] = session
# Add tool to the list of available tools and Format for LLM
self.available_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.input_schema,
},
})
它支援用於本地工具的 stdio
伺服器(例如訪問您的檔案系統),以及用於遠端工具的 http
伺服器!它還與 sse
相容,後者是遠端工具的舊標準。
使用工具:流式傳輸和處理
MCPClient
的 process_single_turn_with_tools
方法是 LLM 交互發生的地方。它透過 AsyncInferenceClient.chat.completions.create(..., stream=True)
將對話歷史和可用工具傳送給 LLM。
1. 準備工具並呼叫大語言模型
首先,該方法確定 LLM 在當前回合中應該知曉的所有工具——這包括來自 MCP 伺服器的工具和任何用於智慧體控制的特殊“退出迴圈”工具;然後,它對 LLM 進行流式呼叫。
# Lines 241-251 of `MCPClient.process_single_turn_with_tools`
# https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/inference/_mcp/mcp_client.py#L241:L251
# Prepare tools list based on options
tools = self.available_tools
if exit_loop_tools is not None:
tools = [*exit_loop_tools, *self.available_tools]
# Create the streaming request to the LLM
response = await self.client.chat.completions.create(
messages=messages,
tools=tools,
tool_choice="auto", # LLM decides if it needs a tool
stream=True,
)
隨著資料塊從 LLM 陸續到達,該方法會遍歷它們。每個資料塊都會立即被 yield,然後我們重構完整的文字響應和任何工具呼叫。
# Lines 258-290 of `MCPClient.process_single_turn_with_tools`
# https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/inference/_mcp/mcp_client.py#L258:L290
# Read from stream
async for chunk in response:
# Yield each chunk to caller
yield chunk
# Aggregate LLM's text response and parts of tool calls
…
2. 執行工具
一旦流結束,如果 LLM 請求了任何工具呼叫(現在已在 final_tool_calls
中完全重構),該方法會處理每一個呼叫。
# Lines 293-313 of `MCPClient.process_single_turn_with_tools`
# https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/inference/_mcp/mcp_client.py#L293:L313
for tool_call in final_tool_calls.values():
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments or "{}")
# Prepare a message to store the tool's result
tool_message = {"role": "tool", "tool_call_id": tool_call.id, "content": "", "name": function_name}
# a. Is this a special "exit loop" tool?
if exit_loop_tools and function_name in [t.function.name for t in exit_loop_tools]:
# If so, yield a message and terminate this turn's processing
messages.append(ChatCompletionInputMessage.parse_obj_as_instance(tool_message))
yield ChatCompletionInputMessage.parse_obj_as_instance(tool_message)
return # The Agent's main loop will handle this signal
# b. It's a regular tool: find the MCP session and execute it
session = self.sessions.get(function_name) # self.sessions maps tool names to MCP connections
if session is not None:
result = await session.call_tool(function_name, function_args)
tool_message["content"] = format_result(result) # format_result processes tool output
else:
tool_message["content"] = f"Error: No session found for tool: {function_name}"
tool_message["content"] = error_msg
# Add tool result to history and yield it
...
它首先檢查被呼叫的工具是否會退出迴圈(exit_loop_tool
)。如果不是,它會找到負責該工具的正確 MCP 會話並呼叫 session.call_tool()
。然後,結果(或錯誤響應)被格式化,新增到對話歷史中,並被 yield,以便智慧體能夠知曉該工具的輸出。
我們的微型 Python 智慧體:它(幾乎)只是一個迴圈!
由於 MCPClient
為工具互動做了所有工作,我們的 Agent
類變得非常簡單。它繼承自 MCPClient
並添加了對話管理邏輯。
Agent 類非常小巧,專注於對話迴圈,程式碼可以在這裡找到。
1. 初始化智慧體
建立 Agent 時,它會接收一個智慧體配置(模型、提供商、要使用的 MCP 伺服器、系統提示),並用系統提示初始化對話歷史。然後,load_tools()
方法會遍歷伺服器配置(在 agent.json 中定義),併為每個配置呼叫 add_mcp_server
(來自父類 MCPClient
),從而填充智慧體的工具箱。
# Lines 12-54 of `Agent`
# https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/inference/_mcp/agent.py#L12:L54
class Agent(MCPClient):
def __init__(
self,
*,
model: str,
servers: Iterable[Dict], # Configuration for MCP servers
provider: Optional[PROVIDER_OR_POLICY_T] = None,
api_key: Optional[str] = None,
prompt: Optional[str] = None, # The system prompt
):
# Initialize the underlying MCPClient with model, provider, etc.
super().__init__(model=model, provider=provider, api_key=api_key)
# Store server configurations to be loaded
self._servers_cfg = list(servers)
# Start the conversation with a system message
self.messages: List[Union[Dict, ChatCompletionInputMessage]] = [
{"role": "system", "content": prompt or DEFAULT_SYSTEM_PROMPT}
]
async def load_tools(self) -> None:
# Connect to all configured MCP servers and register their tools
for cfg in self._servers_cfg:
await self.add_mcp_server(**cfg)
2. 智慧體的核心:迴圈
Agent.run()
方法是一個非同步生成器,用於處理單個使用者輸入。它管理對話輪次,決定智慧體當前任務何時完成。
# Lines 56-99 of `Agent.run()`
# https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/inference/_mcp/agent.py#L56:L99
async def run(self, user_input: str, *, abort_event: Optional[asyncio.Event] = None, ...) -> AsyncGenerator[...]:
...
while True: # Main loop for processing the user_input
...
# Delegate to MCPClient to interact with LLM and tools for one step.
# This streams back LLM text, tool call info, and tool results.
async for item in self.process_single_turn_with_tools(
self.messages,
...
):
yield item
...
# Exit Conditions
# 1. Was an "exit" tool called?
if last.get("role") == "tool" and last.get("name") in {t.function.name for t in EXIT_LOOP_TOOLS}:
return
# 2. Max turns reached or LLM gave a final text answer?
if last.get("role") != "tool" and num_turns > MAX_NUM_TURNS:
return
if last.get("role") != "tool" and next_turn_should_call_tools:
return
next_turn_should_call_tools = (last_message.get("role") != "tool")
在 run()
迴圈內部
- 它首先將使用者提示新增到對話中。
- 然後它呼叫
MCPClient.process_single_turn_with_tools(...)
來獲取 LLM 的響應,並處理任何工具執行,完成一個推理步驟。 - 每個專案都會立即被 yield,從而實現對呼叫者的即時流式傳輸。
- 在每一步之後,它會檢查退出條件:是否使用了特殊的“退出迴圈”工具,是否達到了最大輪次限制,或者 LLM 提供的文字響應對於當前請求似乎是最終的。
下一步
有很多很酷的方式來探索和擴充套件 MCP 客戶端和微智慧體 🔥 這裡有一些想法可以幫助您開始:
- 基準測試不同的 LLM 模型和推理提供商如何影響智慧體效能:工具呼叫效能可能會有所不同,因為每個提供商可能會以不同的方式對其進行最佳化。您可以在此處找到支援的提供商列表。
- 使用本地 LLM 推理伺服器執行微智慧體,例如 llama.cpp 或 LM Studio。
- ……當然還有貢獻!在 Hugging Face Hub 的 tiny-agents/tiny-agents 資料集中分享您獨特的微智慧體並提交 PR。
歡迎提交拉取請求和貢獻!再次強調,這裡的一切都是開源的!💎❤️