跟着😺NanoBot学AI智能体设计和开发4：Python SDK！

导语：Nanobot 不仅仅是一个命令行聊天工具，它底层是一个具备“全网搜索、网页抓取、文件操作、Shell 执行”的强大智能体框架。通过 Python SDK，可以将这套强大的 Agent 能力嵌入到任何 Python 业务中。

本文将从最基础的单次对话开始，逐步解锁多轮记忆、工作区接管、内置工具链驱动，最终构建一个高并发的 Web API。

---

一、 初级篇：环境搭建与日志清理

在最基础的场景中，我们只需要让 Nanobot 跑起来并回答问题。为了避免框架内部冗长的 INFO/DEBUG 日志污染你的终端，我们可以通过 loguru 屏蔽底层日志。

实战代码 1：干净的单次问答

import asyncio
from loguru import logger
from nanobot import Nanobot

# 关闭 nanobot 框架内部的日志输出，让终端保持清爽
logger.disable("nanobot")

async def main():
    bot = Nanobot.from_config()
    
    print("正在向 Nanobot 提问...")
    result = await bot.run("请用一句话介绍 Python。")
    
    print(f"🤖 回答: {result.content}")

if __name__ == "__main__":
    asyncio.run(main())

---

二、 中级篇 1：构建 CLI 交互式多轮对话

单个问题往往不够，我们需要一个可以持续对话的控制台程序。要让 Nanobot 记住上下文（即实现“多轮对话”），核心在于复用同一个 session_key。

实战代码 2：打造你自己的 ChatGPT 终端

import asyncio
from loguru import logger
from nanobot import Nanobot

logger.disable("nanobot")

async def main():
    bot = Nanobot.from_config()
    
    print("========================================")
    print("🤖 Nanobot CLI 模式已启动！(输入 'exit' 退出)")
    print("========================================")
    
    # 设定一个固定的 session_key，这是保持多轮对话记忆的关键
    SESSION_ID = "my-cli-session"
    
    while True:
        try:
            user_input = input("\n🧑 你: ").strip()
        except (KeyboardInterrupt, EOFError):
            break
            
        if user_input.lower() in ['exit', 'quit']:
            break
        if not user_input:
            continue
            
        print("🤖 思考中...", end="\r")
        result = await bot.run(user_input, session_key=SESSION_ID)
        
        print(" " * 20, end="\r") 
        print(f"🤖 Nanobot: {result.content}")
        
        if result.tools_used:
            print(f"   [🔧 辅助操作: 使用了 {', '.join(result.tools_used)} 工具]")

if __name__ == "__main__":
    asyncio.run(main())

---

三、 中级篇 2：工作区接管与多用户隔离

在企业级开发中（例如构建一个面向多用户的微信机器人），所有用户不能共享同一个对话历史，同时 Agent 可能需要针对特定目录下的代码文件进行操作。

实战代码 3：多用户聊天模拟器

import asyncio
from loguru import logger
from nanobot import Nanobot

logger.disable("nanobot")

async def main():
    # 强制指定 Agent 工作的根目录，文件读写等工具将被限制在此目录下
    bot = Nanobot.from_config(workspace="/var/www/my_project")
    
    print("--- 👱‍♀️ Alice 的会话 ---")
    await bot.run("你好，我是前端开发 Alice", session_key="user-alice")
    res_alice = await bot.run("你还记得我是做什么的吗？", session_key="user-alice")
    print(f"Alice 收到回答: {res_alice.content}") 
    
    print("\n--- 👨‍🦱 Bob 的会话 ---")
    # Bob 属于另一个 session_key，无法访问 Alice 的记忆
    res_bob = await bot.run("你还记得我是做什么的吗？", session_key="user-bob")
    print(f"Bob 收到回答: {res_bob.content}") 

if __name__ == "__main__":
    asyncio.run(main())

---

四、 高级篇 1：释放 Nanobot 的内置工具链

Nanobot 底层不仅是一个聊天模型，它还内置了强大的工具链：

1. 网络工具：web_search（全网搜索）、web_fetch（网页提取为 Markdown）。
2. 系统工具：exec（受限 Shell 执行）、read_file / write_file / edit_file / grep / glob 等。

你可以直接用自然语言向它下达复合指令，它会自动规划并调用这些工具。

实战代码 4：自动网页抓取与文件分析助手

import asyncio
from loguru import logger
from nanobot import Nanobot

logger.disable("nanobot")

async def main():
    # 设定一个测试工作区
    bot = Nanobot.from_config(workspace="./research_workspace")
    
    task_prompt = """
    1. 请帮我搜索一下 '2024年诺贝尔物理学奖得主是谁'。
    2. 找到确切结果后，将他们的名字和主要贡献总结成一段话。
    3. 将这段话保存到当前工作区的 'nobel_2024.md' 文件中。
    """
    
    print("🤖 正在执行复杂的网络检索与文件写入任务，请稍候...")
    result = await bot.run(task_prompt, session_key="research-task")
    
    print("\n✅ 任务完成！")
    print(f"最终汇报:\n{result.content}")
    print(f"\n🛠️ 期间 Agent 自主调用的工具列表: {result.tools_used}")

if __name__ == "__main__":
    asyncio.run(main())

在这个例子中，Agent 会自动调用 web_search 查找新闻，然后调用 write_file 写入文件，你无需编写任何爬虫或文件 IO 代码！

---

五、 高级篇 2：利用 Hooks 实现流式输出与安全审计

SDK 最强大的特性是 Hooks（钩子）。它允许你在不修改 Nanobot 核心代码的情况下，监控、拦截 Agent 的行为。我们可以利用 on_stream 钩子，给前面的 CLI 对话加上流式打字机效果。

实战代码 5：支持流式输出与审计的高级终端

import sys
import asyncio
from loguru import logger
from nanobot import Nanobot
from nanobot.agent import AgentHook, AgentHookContext

logger.disable("nanobot")

# 1. 流式输出 Hook：捕获每个 token 并实时打印
class StreamingHook(AgentHook):
    def on_stream(self, ctx: AgentHookContext, delta: str) -> None:
        sys.stdout.write(delta)
        sys.stdout.flush()

    def on_stream_end(self, ctx: AgentHookContext) -> None:
        sys.stdout.write("\n")

# 2. 审计 Hook：记录所有将要执行的工具操作
class AuditHook(AgentHook):
    async def before_execute_tools(self, ctx: AgentHookContext) -> None:
        for tc in ctx.tool_calls:
            print(f"\n⚠️ [安全审计] 拦截到 Agent 尝试调用工具: {tc.name} ({tc.arguments})")

async def main():
    bot = Nanobot.from_config()
    print("🤖 高级流式终端已启动！请提问：")
    
    # 组合传入多个 Hook
    hooks = [StreamingHook(), AuditHook()]
    
    while True:
        user_input = input("\n🧑 你: ").strip()
        if not user_input: continue
            
        print("🤖 Nanobot: ", end="")
        
        # 运行并传入 hooks，Bot 的回答将通过 StreamingHook 实时打印
        await bot.run(user_input, session_key="stream-session", hooks=hooks)

if __name__ == "__main__":
    asyncio.run(main())

---

六、 终极实战：构建一个基于 FastAPI 的智能 Agent API

将上述能力综合起来，我们可以将其封装成一个标准的 Web 后端 API。供前端、小程序或客户端调用。

实战代码 6：结合 FastAPI 构建高并发后端

# 依赖安装: pip install fastapi uvicorn pydantic
import time
from fastapi import FastAPI, Header
from pydantic import BaseModel
from loguru import logger
from nanobot import Nanobot
from nanobot.agent import AgentHook, AgentHookContext

logger.disable("nanobot")

app = FastAPI()

# 全局单例加载 Bot
bot = Nanobot.from_config(workspace="./api_workspace")

# 自定义性能监控 Hook
class TimingHook(AgentHook):
    async def before_iteration(self, ctx: AgentHookContext) -> None:
        ctx.metadata["_t0"] = time.time()

    async def after_iteration(self, ctx: AgentHookContext, response) -> None:
        elapsed = time.time() - ctx.metadata.get("_t0", 0)
        print(f"[监控] LLM 思考耗时: {elapsed:.2f}s")

class ChatRequest(BaseModel):
    message: str

@app.post("/api/v1/agent/chat")
async def chat_endpoint(request: ChatRequest, user_id: str = Header(..., alias="X-User-Id")):
    """
    处理对话请求，使用 X-User-Id 请求头进行会话隔离
    """
    session_key = f"web_session_{user_id}"
    
    result = await bot.run(
        request.message,
        session_key=session_key,
        hooks=[TimingHook()]
    )
    
    return {
        "status": "success",
        "user_id": user_id,
        "reply": result.content,
        "tools_used": result.tools_used
    }

# 启动命令: uvicorn main:app --host 0.0.0.0 --port 8000

七、 终极硬核篇：通过底层消息系统发送图片附件 (多模态)

在 Nanobot 的设计中，高层的 bot.run(message) 是为了纯文本对话设计的便捷接口。如果你在做深度的 SDK 集成（例如将 Nanobot 接入微信、企业内部 IM），当用户直接发来一张图片时，你需要构造底层的 InboundMessage 并通过底层的消息处理方法交由 Agent 分析。

Agent 在处理带有 media（附件）的消息时，会自动读取该路径的图片，将其转换为大模型支持的视觉输入块（Base64），从而让 Agent 具备“看图说话”的能力。

实战代码 7：模拟 IM 机器人接收并分析图片

import asyncio
from loguru import logger
from nanobot import Nanobot
from nanobot.bus.events import InboundMessage

logger.disable("nanobot")

async def main():
    bot = Nanobot.from_config()
    
    # 1. 构造一条底层的 InboundMessage
    # 假设这是从微信或 Slack 接收到的消息和下载到本地的图片
    msg = InboundMessage(
        channel="wechat",          # 来源渠道
        sender_id="user_10086",    # 用户 ID
        chat_id="chat_001",        # 会话 ID
        content="请帮我分析一下这张架构图里有几个核心模块？",
        media=["/absolute/path/to/architecture_diagram.png"]  # 图片的绝对路径
    )
    
    print("📸 正在将图片和问题发送给 Nanobot...")
    
    # 2. 绕过 bot.run()，直接调用底层的 _process_message 处理多模态消息
    # 这样可以同步等待结果返回，而不必去处理复杂的异步总线队列 (bus)
    session_key = f"im_session_{msg.sender_id}"
    result = await bot._loop._process_message(msg, session_key=session_key)
    
    print("\n✅ 分析完成！")
    print(f"🤖 回答:\n{result.content}")

if __name__ == "__main__":
    asyncio.run(main())

注意：这里的 media 列表可以包含多个图片的绝对路径，Agent 会将它们一起发送给大模型进行多模态推理。