NanoBot - 轻量级 AI Agent 框架与智能体构建工具

快速开始

安装方式

# 方式一：从 GitHub 直接安装
pip install git+https://github.com/HKUDS/nanobot.git

# 方式二：克隆后本地安装
git clone https://github.com/HKUDS/nanobot.git
cd nanobot
pip install -e .

# 方式三：如果已发布到 PyPI
pip install nanobot

系统要求：

• Python 3.8+
• 支持的 LLM API（OpenAI、Anthropic 等）

基本使用

1. 创建简单的 Agent

from nanobot import Agent, LLM

# 初始化 LLM
llm = LLM(provider="openai", model="gpt-4")

# 创建 Agent
agent = Agent(
    name="Assistant",
    llm=llm,
    system_prompt="You are a helpful assistant."
)

# 与 Agent 对话
response = agent.chat("Hello, how are you?")
print(response)

2. 添加工具支持

from nanobot import Agent, Tool

# 定义工具函数
def get_weather(location: str) -> str:
    """Get weather information for a location."""
    return f"Weather in {location}: Sunny, 25°C"

# 创建工具
weather_tool = Tool(
    name="get_weather",
    description="Get weather information",
    function=get_weather
)

# 创建带工具的 Agent
agent = Agent(
    name="WeatherBot",
    llm=llm,
    tools=[weather_tool]
)

# Agent 可以自动使用工具
response = agent.chat("What's the weather in Hong Kong?")
print(response)

3. 使用规划器

from nanobot import Agent, Planner

# 创建规划器
planner = Planner(llm=llm)

# 创建带规划器的 Agent
agent = Agent(
    name="PlannerBot",
    llm=llm,
    planner=planner
)

# Agent 可以规划复杂任务
response = agent.chat("Plan a trip to Japan: research flights, hotels, and attractions")
print(response)

4. 流式响应

# 启用流式响应
for chunk in agent.chat_stream("Tell me a story"):
    print(chunk, end="", flush=True)

架构设计

NanoBot 采用模块化、可扩展的架构设计，核心组件清晰分离，便于理解和定制。

核心架构

NanoBot/
├── Agent (核心)
│   ├── LLM 接口
│   ├── 工具系统
│   ├── 规划器
│   ├── 记忆管理
│   └── 执行引擎
├── Tools (工具)
│   ├── 内置工具
│   ├── 自定义工具
│   └── 工具链
├── Planner (规划器)
│   ├── 任务分解
│   ├── 步骤规划
│   └── 执行策略
├── Memory (记忆)
│   ├── 对话历史
│   ├── 长期记忆
│   └── 上下文管理
└── LLM (大模型)
    ├── 多提供商支持
    ├── 统一接口
    └── 流式响应

设计原则

1. 简洁性优先：最少的代码实现最多的功能
2. 模块化设计：每个组件都是独立的模块，可以单独使用或替换
3. 可扩展性：通过继承和组合轻松扩展功能

核心模块详解

1. Agent 核心模块

功能：

• Agent 的创建和配置
• 对话管理和响应生成
• 工具调用和规划执行
• 状态管理和上下文维护

技术实现：

class Agent:
    def __init__(
        self,
        name: str,
        llm: LLM,
        system_prompt: str = None,
        tools: List[Tool] = None,
        planner: Planner = None,
        memory: Memory = None
    ):
        self.name = name
        self.llm = llm
        self.system_prompt = system_prompt
        self.tools = tools or []
        self.planner = planner
        self.memory = memory or SimpleMemory()
        self.conversation_history = []

    def chat(self, message: str) -> str:
        # 1. 添加到对话历史
        self.conversation_history.append({"role": "user", "content": message})
        
        # 2. 如果有规划器，先规划
        if self.planner:
            plan = self.planner.plan(message, self.conversation_history)
            # 执行规划...
        
        # 3. 检查是否需要调用工具
        tool_calls = self._detect_tool_calls(message)
        if tool_calls:
            results = self._execute_tools(tool_calls)
            message = self._format_with_tool_results(message, results)
        
        # 4. 调用 LLM
        response = self.llm.chat(
            messages=self._build_messages(),
            tools=self._format_tools()
        )
        
        # 5. 保存响应
        self.conversation_history.append({"role": "assistant", "content": response})
        return response

    def _detect_tool_calls(self, message: str) -> List[dict]:
        # 使用 LLM 判断是否需要调用工具
        pass

    def _execute_tools(self, tool_calls: List[dict]) -> List[dict]:
        results = []
        for tool_call in tool_calls:
            tool = self._find_tool(tool_call["name"])
            result = tool.execute(tool_call["arguments"])
            results.append(result)
        return results

2. 工具系统模块

功能：

• 工具的定义和注册
• 工具调用的执行
• 工具链的组合
• 工具结果的格式化

技术实现：

class Tool:
    def __init__(self, name: str, description: str, function: Callable, parameters: dict = None):
        self.name = name
        self.description = description
        self.function = function
        self.parameters = parameters or {}

    def execute(self, **kwargs) -> Any:
        # 验证参数
        self._validate_parameters(kwargs)
        try:
            result = self.function(**kwargs)
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}

    def to_openai_format(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters
            }
        }

3. 规划器模块

功能：

• 任务分解和步骤规划
• 执行策略制定
• 计划优化和调整

技术实现：

class Planner:
    def __init__(self, llm: LLM):
        self.llm = llm

    def plan(self, task: str, context: List[dict] = None) -> dict:
        prompt = self._build_planning_prompt(task, context)
        response = self.llm.chat(
            messages=[{"role": "user", "content": prompt}],
            response_format="json"
        )
        plan = json.loads(response)
        return {
            "task": task,
            "steps": plan["steps"],
            "estimated_time": plan.get("estimated_time"),
            "dependencies": plan.get("dependencies", [])
        }

    def _build_planning_prompt(self, task: str, context: List[dict]) -> str:
        return f"""
        Given the following task, create a detailed plan:
        Task: {task}
        Context: {json.dumps(context, indent=2) if context else "None"}
        Please provide a JSON response with:
        - steps: List of steps to complete the task
        - estimated_time: Estimated time for each step
        - dependencies: Dependencies between steps
        """

4. 记忆管理模块

功能：

• 对话历史管理
• 上下文窗口优化
• 长期记忆存储

技术实现：

class SimpleMemory:
    def __init__(self, max_history: int = 100):
        self.max_history = max_history
        self.history = []

    def add(self, role: str, content: str):
        self.history.append({"role": role, "content": content, "timestamp": time.time()})
        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history:]

    def get_context(self, max_tokens: int = None) -> List[dict]:
        if max_tokens:
            return self._truncate_by_tokens(self.history, max_tokens)
        return self.history

    def _truncate_by_tokens(self, history: List[dict], max_tokens: int) -> List[dict]:
        truncated = []
        current_tokens = 0
        for message in reversed(history):
            message_tokens = self._count_tokens(message["content"])
            if current_tokens + message_tokens > max_tokens:
                break
            truncated.insert(0, message)
            current_tokens += message_tokens
        return truncated

5. LLM 集成模块

功能：

• 多 LLM 提供商支持
• 统一的接口抽象
• 流式响应处理

技术实现：

class LLM:
    def __init__(self, provider: str, model: str, api_key: str = None):
        self.provider = provider
        self.model = model
        self.api_key = api_key or os.getenv(f"{provider.upper()}_API_KEY")
        self.client = self._create_client()

    def _create_client(self):
        if self.provider == "openai":
            import openai
            return openai.OpenAI(api_key=self.api_key)
        elif self.provider == "anthropic":
            import anthropic
            return anthropic.Anthropic(api_key=self.api_key)

    def chat(self, messages: List[dict], tools: List[dict] = None, response_format: str = None) -> str:
        if self.provider == "openai":
            return self._openai_chat(messages, tools, response_format)
        elif self.provider == "anthropic":
            return self._anthropic_chat(messages, tools, response_format)

    def chat_stream(self, messages: List[dict], tools: List[dict] = None):
        if self.provider == "openai":
            stream = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=tools,
                stream=True
            )
            for chunk in stream:
                if chunk.choices[0].delta.content:
                    yield chunk.choices[0].delta.content

关键技术实现

1. 工具自动调用

NanoBot 通过 LLM 的函数调用能力自动检测和执行工具：

def _detect_and_execute_tools(self, message: str, context: List[dict]) -> str:
    # 1. 构建包含工具信息的消息
    messages = self._build_messages_with_tools(context)
    
    # 2. 调用 LLM，启用函数调用
    response = self.llm.chat(
        messages=messages,
        tools=[tool.to_openai_format() for tool in self.tools]
    )
    
    # 3. 检查是否有工具调用
    if hasattr(response, "tool_calls") and response.tool_calls:
        # 4. 执行工具
        tool_results = []
        for tool_call in response.tool_calls:
            tool = self._find_tool(tool_call.function.name)
            result = tool.execute(**json.loads(tool_call.function.arguments))
            tool_results.append({
                "tool_call_id": tool_call.id,
                "role": "tool",
                "name": tool_call.function.name,
                "content": json.dumps(result)
            })
        
        # 5. 将工具结果添加到上下文，再次调用 LLM
        messages.extend(tool_results)
        final_response = self.llm.chat(messages=messages)
        return final_response.content
    
    return response.content

2. 规划与执行循环

def execute_with_planning(self, task: str) -> str:
    # 1. 生成计划
    plan = self.planner.plan(task, self.conversation_history)
    
    # 2. 执行计划中的每个步骤
    results = []
    for step in plan["steps"]:
        if step.get("requires_tool"):
            tool_result = self._execute_tool_for_step(step)
            results.append(tool_result)
        else:
            response = self.llm.chat(
                messages=self._build_messages() + [{"role": "user", "content": step["description"]}]
            )
            results.append(response.content)
    
    # 3. 反思和调整（可选）
    if step.get("requires_reflection"):
        reflection = self._reflect_on_step(step, results[-1])
        if reflection["should_adjust"]:
            plan = self._adjust_plan(plan, reflection)
    
    # 4. 总结结果
    summary = self._summarize_execution(plan, results)
    return summary

3. 上下文窗口优化

def optimize_context(self, messages: List[dict], max_tokens: int) -> List[dict]:
    # 策略 1: 保留系统提示和最近的对话
    system_messages = [msg for msg in messages if msg["role"] == "system"]
    recent_messages = messages[-10:]  # 保留最近 10 条
    
    # 策略 2: 如果还是超限，使用摘要
    current_tokens = self._count_tokens(system_messages + recent_messages)
    if current_tokens > max_tokens:
        old_messages = messages[:-10]
        summary = self._summarize_messages(old_messages)
        return system_messages + [
            {"role": "assistant", "content": f"Previous conversation summary: {summary}"}
        ] + recent_messages
    
    return system_messages + recent_messages

扩展机制

自定义 Agent

class CustomAgent(Agent):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.custom_state = {}

    def custom_method(self, input_data):
        result = self.llm.chat([{"role": "user", "content": f"Process: {input_data}"}])
        self.custom_state["last_result"] = result
        return result

自定义工具

class DatabaseTool(Tool):
    def __init__(self, connection_string: str):
        super().__init__(
            name="query_database",
            description="Query a SQL database",
            function=self._query
        )
        self.db = connect(connection_string)

    def _query(self, sql: str) -> dict:
        try:
            result = self.db.execute(sql)
            return {
                "success": True,
                "data": result.fetchall(),
                "columns": result.keys()
            }
        except Exception as e:
            return {"success": False, "error": str(e)}

自定义规划器

class HierarchicalPlanner(Planner):
    def plan(self, task: str, context: List[dict] = None) -> dict:
        # 1. 高层次规划
        high_level_plan = self._high_level_planning(task)
        
        # 2. 对每个高层次目标进行详细规划
        detailed_steps = []
        for goal in high_level_plan["goals"]:
            steps = self._detailed_planning(goal)
            detailed_steps.extend(steps)
        
        return {
            "task": task,
            "high_level_goals": high_level_plan["goals"],
            "detailed_steps": detailed_steps
        }

资源链接

• GitHub: https://github.com/HKUDS/nanobot
• 文档: GitHub README 和 Wiki
• 社区: GitHub Issues 和 Discussions
• Issue Tracker: GitHub Issues