2.4.2 Agent Builder - 无代码智能体构建器

难度级别: 高级 前置知识: Agent 基础概念、LangGraph 状态管理、工具调用 预计阅读时间: 45 分钟

概述

什么是 Agent Builder?

Agent Builder 是 LangSmith 提供的一个 无代码智能体构建平台（目前处于 Beta 阶段）。它允许用户通过自然语言描述来创建生产级的 AI 智能体，无需编写代码。

核心理念: 将自然语言概念转换为可运行的生产级智能体

与传统的可视化工作流构建器不同，Agent Builder 采用 对话式构建方式：

用户描述任务 → 系统提问澄清 → 自动生成提示词 → 连接工具 → 设置触发器

架构概览

Agent Builder 基于开源项目 deep-agents 构建，整合了 LangChain 和 LangGraph 的最佳实践。

┌─────────────────────────────────────────────────────────────────┐
│                     Agent Builder 架构                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐      │
│  │   Prompt     │    │    Tools     │    │   Triggers   │      │
│  │  (智能体大脑) │    │ (工具连接)   │    │ (自动触发)   │      │
│  └──────┬───────┘    └──────┬───────┘    └──────┬───────┘      │
│         │                   │                   │               │
│         └───────────────────┼───────────────────┘               │
│                             │                                   │
│                    ┌────────▼────────┐                          │
│                    │   Sub-agents    │                          │
│                    │  (子智能体)      │                          │
│                    └────────┬────────┘                          │
│                             │                                   │
│                    ┌────────▼────────┐                          │
│                    │     Memory      │                          │
│                    │  (记忆系统)      │                          │
│                    └─────────────────┘                          │
│                                                                 │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                    Deep Agents 引擎                      │   │
│  │  (Planning + Computer Access + Sub-agent Delegation)     │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

与传统方式的对比

特性	传统代码开发	可视化工作流	Agent Builder
技术门槛	高	中	低
灵活性	高	中	中高
构建方式	编写代码	拖拽节点	自然语言对话
适合人群	开发者	技术产品经理	所有人
维护成本	高	中	低

核心概念

Agent Builder 的每个智能体由 四个核心组件 构成：

1. Prompt（提示词）

提示词是智能体的"大脑"，包含智能体应该做什么的逻辑描述。

特点：

所有复杂逻辑都推入提示词，而非复杂的可视化工作流
通过对话式引导自动生成
支持后续修改和优化

python

# 示例：智能体提示词结构
system_prompt = """
你是一个智能会议助手。

职责：
1. 检查用户的日历
2. 提取会议参与者信息
3. 查找相关上下文
4. 在 Slack 中发送会议摘要

注意事项：
- 确保信息准确
- 保持摘要简洁
- 标注重要议题
"""

2. Tools（工具）

工具使用 MCP（Model Context Protocol） 协议连接外部服务。

两种工具来源：

内置工具：Gmail、Slack、Google Calendar、Linear 等
自定义 MCP 服务器：用户自己部署的工具服务

3. Triggers（触发器）

触发器使智能体能够自动响应后台事件：

触发器类型	说明	使用场景
Slack	特定频道的消息触发	客服机器人、团队助手
Gmail	邮件到达触发	邮件分类、自动回复
Cron	定时触发	日报生成、定期检查

4. Sub-agents（子智能体）

子智能体是处理特定任务的专门化小型智能体：

┌─────────────────────────────────────────┐
│            主智能体 (Orchestrator)       │
│                                         │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐ │
│  │ 数据检索 │  │ 内容生成 │  │ 通知发送 │ │
│  │ 子智能体 │  │ 子智能体 │  │ 子智能体 │ │
│  └────┬────┘  └────┬────┘  └────┬────┘ │
│       │            │            │       │
│       └────────────┼────────────┘       │
│                    ▼                    │
│              结果汇总和协调              │
└─────────────────────────────────────────┘

子智能体的优势：

专注于特定任务，提高效率
独立的上下文窗口，避免混乱
可并行执行，缩短总体时间
便于维护和调试

5. Memory（记忆系统）

Agent Builder 的智能体具有内置记忆功能：

记忆特性：

跨会话保持：相关信息在会话之间持续存储
自我更新：智能体可以修改自己的工具和指令
纠错学习：当用户纠正智能体时，会自动记住以避免重复错误

不可修改项：

智能体名称
智能体描述
已附加的触发器

6. Human-in-the-Loop（人机协作）

智能体在执行关键操作前会暂停，等待人工审核：

智能体提议操作 → 人工审核 → [批准/修改/拒绝] → 继续执行

三种响应选项：

批准 (Approve)：同意提议的操作
修改 (Edit)：调整参数后继续
反馈 (Feedback)：提供改进建议

环境配置

工作区密钥设置

Agent Builder 需要在 LangSmith 工作区中配置密钥才能正常工作。

配置步骤

登录 LangSmith，点击 Settings（设置） 图标
选择 Secrets（密钥） 标签页
点击 Add secret（添加密钥）
输入密钥名称和对应的值
点击 Save secret（保存密钥）

必需的密钥

密钥名称	说明	必需性
`ANTHROPIC_API_KEY`	Anthropic API 密钥，用于模型推理	必需

重要: Agent Builder 默认使用 Anthropic 的 Claude 模型，因此 ANTHROPIC_API_KEY 是必需的。

可选的工具密钥

密钥名称	功能
`EXA_API_KEY`	启用 Exa 搜索（网络搜索、LinkedIn 档案搜索）
`TAVILY_API_KEY`	启用 Tavily 网络搜索
`TWITTER_API_KEY`	Twitter/X 只读操作
`TWITTER_API_KEY_SECRET`	Twitter/X API 密钥对

注意: 添加工作区密钥时，确保密钥名称与模型提供商期望的环境变量名称完全匹配。

内置工具

Agent Builder 提供 8 类内置工具：

通讯与消息

Gmail

功能	说明
读取邮件	支持搜索过滤、可选包含正文
发送邮件	撰写并发送新邮件
创建草稿	保存邮件草稿
管理标签	添加/删除邮件标签

Slack

功能	说明
发送私信	向用户发送直接消息
频道发布	在频道中发布消息
线程回复	回复特定消息线程
历史访问	读取频道和线程历史

日历与项目管理

Google Calendar

列出日历事件
获取事件详情
创建新的日历条目

Linear

管理问题（创建、过滤、删除）
团队管理功能

搜索与发现

工具	功能
Exa	网络搜索、LinkedIn 档案搜索
Tavily	高质量网络搜索

社交媒体

发布帖子
支持附加图片或链接

Twitter/X

读取单条推文
获取列表的最新帖子
注意：仅支持只读操作，不支持发布

实用工具

功能	说明
提取网页文本	从 URL 获取页面内容
获取图片 URL	提取页面中的图片链接
提取元数据	获取页面元信息
用户通知	发送通知给用户

认证方式

方式	适用工具
OAuth	Google、Slack、Linear、LinkedIn
工作区密钥	Exa、Tavily、Twitter/X

MCP 框架

什么是 MCP?

MCP（Model Context Protocol） 是一个标准化协议，用于创建、部署和管理智能体工具。LangSmith Tool Server 是基于 MCP 的工具服务器框架。

安装

bash

# 安装工具服务器
pip install langsmith-tool-server

# 安装 CLI 工具
pip install langchain-cli-v2

创建工具包

bash

# 创建新的工具包项目
langchain tools new my-toolkit
cd my-toolkit

这会创建以下目录结构：

my-toolkit/
├── pyproject.toml      # 项目配置
├── toolkit.toml        # 工具包配置
└── my_toolkit/
    ├── __init__.py     # 工具定义
    └── auth.py         # 认证处理（可选）

定义工具

使用 @tool 装饰器定义工具：

python

# my_toolkit/__init__.py
from langsmith_tool_server import tool

@tool
def hello(name: str) -> str:
    """
    向某人问好。

    Args:
        name: 要问候的人的名字
    """
    return f"Hello, {name}!"

@tool
def add(x: int, y: int) -> int:
    """
    计算两个数的和。

    Args:
        x: 第一个数
        y: 第二个数
    """
    return x + y

@tool
def search_database(query: str, limit: int = 10) -> str:
    """
    搜索数据库。

    Args:
        query: 搜索查询
        limit: 返回结果数量限制
    """
    # 实际实现中连接数据库
    return f"找到关于 '{query}' 的 {limit} 条结果"

# 导出工具列表
TOOLS = [hello, add, search_database]

运行服务器

bash

# 启动 MCP 服务器（默认端口 8000）
langchain tools serve

服务器运行后，可通过 http://localhost:8000/mcp 访问 JSON-RPC 端点。

客户端调用

python

import asyncio
import aiohttp

async def mcp_request(url: str, method: str, params: dict = None):
    """发送 MCP 请求"""
    async with aiohttp.ClientSession() as session:
        payload = {
            "jsonrpc": "2.0",
            "method": method,
            "params": params or {},
            "id": 1
        }
        async with session.post(f"{url}/mcp", json=payload) as response:
            return await response.json()

async def main():
    url = "http://localhost:8000"

    # 列出所有可用工具
    tools = await mcp_request(url, "tools/list")
    print(f"可用工具: {tools}")

    # 调用 add 工具
    result = await mcp_request(
        url,
        "tools/call",
        {"name": "add", "arguments": {"x": 5, "y": 3}}
    )
    print(f"计算结果: {result}")

    # 调用 hello 工具
    greeting = await mcp_request(
        url,
        "tools/call",
        {"name": "hello", "arguments": {"name": "LangGraph"}}
    )
    print(f"问候结果: {greeting}")

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

OAuth 认证工具

需要 OAuth 认证的工具示例：

python

from langsmith_tool_server import tool, Context
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build

@tool(
    auth_provider="google",
    scopes=["https://www.googleapis.com/auth/gmail.readonly"],
    integration="gmail"
)
async def read_emails(context: Context, max_results: int = 10) -> str:
    """
    读取 Gmail 中的最近邮件。

    Args:
        context: 认证上下文（自动注入）
        max_results: 返回的最大邮件数量
    """
    # 使用 OAuth token 创建凭证
    credentials = Credentials(token=context.token)
    service = build('gmail', 'v1', credentials=credentials)

    # 获取邮件列表
    results = service.users().messages().list(
        userId='me',
        maxResults=max_results
    ).execute()

    messages = results.get('messages', [])
    return f"获取了 {len(messages)} 封邮件"

@tool(
    auth_provider="slack",
    integration="slack"
)
async def send_slack_message(
    context: Context,
    channel: str,
    message: str
) -> str:
    """
    发送 Slack 消息。

    Args:
        context: 认证上下文
        channel: 频道 ID 或名称
        message: 消息内容
    """
    # 使用 context.token 调用 Slack API
    # ...
    return f"消息已发送到 {channel}"

自定义认证

python

# my_toolkit/auth.py
from langsmith_tool_server import Auth

auth = Auth()

@auth.authenticate
async def authenticate(authorization: str = None) -> dict:
    """
    验证请求并返回用户身份。

    Args:
        authorization: Authorization 请求头
    """
    if not authorization or not authorization.startswith("Bearer "):
        raise auth.exceptions.HTTPException(
            status_code=401,
            detail="未授权：需要 Bearer token"
        )

    token = authorization.replace("Bearer ", "")

    # 使用你的身份提供商验证 token
    user = await verify_token_with_idp(token)

    return {
        "identity": user.id,
        "permissions": user.permissions
    }

async def verify_token_with_idp(token: str):
    """验证 token（示例实现）"""
    # 实际实现中调用 IdP API
    class User:
        id = "user_123"
        permissions = ["read", "write"]
    return User()

MCP 网关配置

聚合多个 MCP 服务器的工具：

toml

# toolkit.toml
[toolkit]
name = "my-toolkit"
tools = "./my_toolkit/__init__.py:TOOLS"

# HTTP 传输的远程服务器
[[mcp_servers]]
name = "weather"
transport = "streamable_http"
url = "http://localhost:8001/mcp/"

# stdio 传输的本地服务器
[[mcp_servers]]
name = "math"
transport = "stdio"
command = "python"
args = ["-m", "mcp_server_math"]

# 另一个远程服务器
[[mcp_servers]]
name = "database"
transport = "streamable_http"
url = "https://db-tools.example.com/mcp/"

配置后，工具会以服务器名称为前缀（如 weather_get_forecast、math_calculate）以避免命名冲突。

Slack 应用集成

安装步骤

在 LangSmith 中访问 Agent Builder
创建新智能体或编辑现有智能体
添加 Slack 作为触发器或启用 Slack 工具
按提示完成授权
完成 OAuth 流程，授予 Slack 工作区权限

应用会在授权完成后自动部署。

核心功能

集成后，智能体获得以下能力：

功能	说明
发送私信	向 Slack 用户发送直接消息
频道发布	在频道中发布消息
线程阅读	读取线程消息历史
线程回复	回复特定消息线程
历史访问	访问对话历史记录

权限要求

Slack 集成需要以下权限：

权限类型	说明
消息	发送私信和频道消息
阅读	读取频道历史和线程消息
频道访问	获取基本频道信息
用户查找	查找用户信息以发送消息

数据与隐私

关键保障：

Slack 数据不会用于训练 LLM
工作区数据保持隔离，仅用于操作目的
数据保留遵循 LangSmith 组织策略
数据存储位置与 LangSmith 配置一致

费用说明

Slack 应用本身免费
费用来自智能体执行和跟踪，通过 LangSmith 订阅计费

Deep Agents 底层框架

Agent Builder 基于开源的 Deep Agents 框架构建。

核心设计原则

Deep Agents 受 Claude Code 等领先 AI 智能体启发，实现三个基础原则：

规划 (Planning): 执行前的结构化任务分解
计算机访问 (Computer Access): Shell 和文件系统集成
子智能体委托 (Sub-agent Delegation): 隔离的执行上下文

内置工具

Deep Agents 为所有智能体提供 9 个标准工具：

工具	功能	来源中间件
`write_todos`	写入待办事项	TodoListMiddleware
`read_todos`	读取待办事项	TodoListMiddleware
`ls`	列出目录	FilesystemMiddleware
`read_file`	读取文件	FilesystemMiddleware
`write_file`	写入文件	FilesystemMiddleware
`edit_file`	编辑文件	FilesystemMiddleware
`glob`	文件匹配	FilesystemMiddleware
`grep`	内容搜索	FilesystemMiddleware
`task`	创建子智能体	SubAgentMiddleware

中间件系统

Deep Agents 使用可组合的中间件扩展功能：

中间件	功能
TodoListMiddleware	任务规划和进度跟踪
FilesystemMiddleware	文件操作和上下文管理
SubAgentMiddleware	子智能体委托
SummarizationMiddleware	170k token 时自动上下文压缩
AnthropicPromptCachingMiddleware	Anthropic 模型成本优化
HumanInTheLoopMiddleware	敏感操作的审批工作流

后端存储选项

后端	说明	适用场景
StateBackend	临时、会话范围存储（默认）	开发、测试
FilesystemBackend	直接磁盘访问	需要持久化的本地应用
StoreBackend	通过 LangGraph Store 持久存储	生产环境、跨会话
CompositeBackend	路由不同路径到不同后端	复杂场景

完整案例代码

以下是一个完整的示例，展示如何使用 Deep Agents 框架构建智能体，以及如何创建和调用 MCP 工具服务器。

python

"""
Agent Builder 完整示例
演示：Deep Agents 创建、MCP 工具服务器、子智能体委托

场景：智能研究助手
1. 接收用户研究请求
2. 使用搜索工具获取信息
3. 委托给专门的子智能体处理
4. 支持任务管理和文件操作
"""

import os
import asyncio
from typing import List, Optional
from typing_extensions import TypedDict
from pydantic import BaseModel, Field

# ========== 1. 数据模型定义 ==========

class SearchResult(BaseModel):
    """搜索结果模型"""
    title: str = Field(description="结果标题")
    snippet: str = Field(description="内容摘要")
    url: str = Field(description="来源链接")

class ResearchTask(BaseModel):
    """研究任务模型"""
    topic: str = Field(description="研究主题")
    questions: List[str] = Field(description="需要回答的问题")
    deadline: Optional[str] = Field(default=None, description="截止日期")

class TodoItem(BaseModel):
    """待办事项模型"""
    task: str = Field(description="任务描述")
    status: str = Field(default="pending", description="状态: pending/in_progress/completed")
    priority: str = Field(default="medium", description="优先级: low/medium/high")

# ========== 2. 模拟 MCP 工具定义 ==========

class MCPTool:
    """MCP 工具基类"""
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description

    def invoke(self, **kwargs):
        raise NotImplementedError

class SearchWebTool(MCPTool):
    """网络搜索工具"""
    def __init__(self):
        super().__init__(
            name="search_web",
            description="搜索网络信息"
        )

    def invoke(self, query: str, max_results: int = 5) -> List[dict]:
        """执行搜索"""
        # 模拟搜索结果
        results = [
            {
                "title": f"关于 {query} 的权威指南",
                "snippet": f"这是关于 {query} 的详细介绍，涵盖核心概念和最佳实践...",
                "url": f"https://example.com/guide/{query.replace(' ', '-')}"
            },
            {
                "title": f"{query} 入门教程",
                "snippet": f"从零开始学习 {query}，包含实战案例和代码示例...",
                "url": f"https://tutorial.com/{query.replace(' ', '-')}"
            },
            {
                "title": f"{query} 最新进展",
                "snippet": f"2024-2025 年 {query} 领域的最新发展和趋势分析...",
                "url": f"https://news.com/{query.replace(' ', '-')}"
            }
        ]
        print(f"🔍 搜索完成: {query} (找到 {len(results)} 条结果)")
        return results[:max_results]

class ReadFileTool(MCPTool):
    """文件读取工具"""
    def __init__(self):
        super().__init__(
            name="read_file",
            description="读取文件内容"
        )

    def invoke(self, path: str) -> str:
        """读取文件"""
        print(f"📄 读取文件: {path}")
        # 模拟文件内容
        return f"文件 {path} 的内容：\n这是一份示例文档..."

class WriteFileTool(MCPTool):
    """文件写入工具"""
    def __init__(self):
        super().__init__(
            name="write_file",
            description="写入文件内容"
        )

    def invoke(self, path: str, content: str) -> str:
        """写入文件"""
        print(f"📝 写入文件: {path}")
        return f"文件已保存到 {path}"

class SendNotificationTool(MCPTool):
    """通知发送工具"""
    def __init__(self):
        super().__init__(
            name="send_notification",
            description="发送用户通知"
        )

    def invoke(self, message: str, channel: str = "default") -> str:
        """发送通知"""
        print(f"🔔 通知 [{channel}]: {message}")
        return f"通知已发送到 {channel}"

# ========== 3. 任务管理系统 ==========

class TodoListManager:
    """任务列表管理器 (模拟 TodoListMiddleware)"""

    def __init__(self):
        self.todos: List[TodoItem] = []

    def add_todo(self, task: str, priority: str = "medium") -> TodoItem:
        """添加任务"""
        item = TodoItem(task=task, priority=priority)
        self.todos.append(item)
        print(f"📋 添加任务: {task} (优先级: {priority})")
        return item

    def update_status(self, task: str, status: str):
        """更新任务状态"""
        for todo in self.todos:
            if todo.task == task:
                todo.status = status
                print(f"✅ 任务状态更新: {task} -> {status}")
                return
        print(f"⚠️ 任务未找到: {task}")

    def get_todos(self, status: str = None) -> List[TodoItem]:
        """获取任务列表"""
        if status:
            return [t for t in self.todos if t.status == status]
        return self.todos

    def display(self):
        """显示任务列表"""
        print("\n📋 任务列表:")
        for i, todo in enumerate(self.todos, 1):
            status_icon = {"pending": "⏳", "in_progress": "🔄", "completed": "✅"}
            print(f"  {status_icon.get(todo.status, '❓')} {i}. {todo.task}")

# ========== 4. 子智能体定义 ==========

class SubAgent:
    """子智能体基类"""

    def __init__(self, name: str, description: str, tools: List[MCPTool]):
        self.name = name
        self.description = description
        self.tools = {tool.name: tool for tool in tools}
        print(f"🤖 子智能体 [{name}] 已创建")
        print(f"   工具: {list(self.tools.keys())}")

    def invoke(self, task: str) -> str:
        raise NotImplementedError

class ResearchSubAgent(SubAgent):
    """研究专用子智能体"""

    def __init__(self):
        super().__init__(
            name="research-agent",
            description="专门负责搜索和整理研究资料",
            tools=[SearchWebTool(), ReadFileTool()]
        )

    def invoke(self, task: str) -> str:
        """执行研究任务"""
        print(f"\n📚 研究子智能体开始处理: {task[:50]}...")

        # 使用搜索工具
        search_tool = self.tools["search_web"]
        results = search_tool.invoke(query=task)

        # 整理结果
        summary = f"研究主题: {task}\n\n"
        summary += "发现的资源:\n"
        for i, result in enumerate(results, 1):
            summary += f"\n{i}. {result['title']}\n"
            summary += f"   摘要: {result['snippet'][:100]}...\n"
            summary += f"   链接: {result['url']}\n"

        return summary

class WriterSubAgent(SubAgent):
    """写作专用子智能体"""

    def __init__(self):
        super().__init__(
            name="writer-agent",
            description="专门负责撰写和整理文档",
            tools=[WriteFileTool(), SendNotificationTool()]
        )

    def invoke(self, task: str) -> str:
        """执行写作任务"""
        print(f"\n✍️ 写作子智能体开始处理: {task[:50]}...")

        # 生成内容
        content = f"""
# {task}

## 概述

这是关于 {task} 的研究报告。

## 主要发现

1. 发现一：核心概念解析
2. 发现二：最佳实践总结
3. 发现三：未来趋势展望

## 结论

综合以上分析，{task} 是一个重要且值得深入研究的领域。

---
*由 AI 研究助手自动生成*
"""

        # 保存文件
        write_tool = self.tools["write_file"]
        filename = f"/reports/{task.replace(' ', '_')}.md"
        write_tool.invoke(path=filename, content=content)

        # 发送通知
        notify_tool = self.tools["send_notification"]
        notify_tool.invoke(
            message=f"报告已生成: {filename}",
            channel="slack"
        )

        return content

# ========== 5. 主智能体（协调器） ==========

class OrchestratorAgent:
    """
    协调器智能体
    模拟 Agent Builder 的主智能体功能
    """

    def __init__(self):
        self.name = "orchestrator"
        self.todo_manager = TodoListManager()
        self.memory = {}  # 简单的记忆存储

        # 注册子智能体
        self.subagents = {
            "research": ResearchSubAgent(),
            "writer": WriterSubAgent()
        }

        # 注册工具
        self.tools = {
            "search_web": SearchWebTool(),
            "read_file": ReadFileTool(),
            "write_file": WriteFileTool(),
            "send_notification": SendNotificationTool()
        }

        print(f"\n🎯 协调器智能体已初始化")
        print(f"   子智能体: {list(self.subagents.keys())}")
        print(f"   工具: {list(self.tools.keys())}")

    def delegate_to_subagent(self, agent_name: str, task: str) -> str:
        """委托任务给子智能体"""
        if agent_name not in self.subagents:
            return f"❌ 未找到子智能体: {agent_name}"

        print(f"\n📤 委托任务给 [{agent_name}]")
        subagent = self.subagents[agent_name]
        result = subagent.invoke(task)
        return result

    def save_to_memory(self, key: str, value: str):
        """保存到记忆"""
        self.memory[key] = value
        print(f"💾 保存到记忆: {key}")

    def get_from_memory(self, key: str) -> Optional[str]:
        """从记忆中获取"""
        return self.memory.get(key)

    def process_request(self, request: str) -> str:
        """处理用户请求"""
        print(f"\n{'='*60}")
        print(f"👤 用户请求: {request}")
        print('='*60)

        # 添加主任务
        self.todo_manager.add_todo(f"处理请求: {request[:30]}...", "high")
        self.todo_manager.update_status(f"处理请求: {request[:30]}...", "in_progress")

        # 分析请求并分配任务
        if "研究" in request or "搜索" in request or "查找" in request:
            # 委托给研究子智能体
            self.todo_manager.add_todo("执行研究任务", "medium")
            research_result = self.delegate_to_subagent("research", request)
            self.todo_manager.update_status("执行研究任务", "completed")

            # 保存研究结果
            self.save_to_memory(f"research_{request[:20]}", research_result)

            # 委托给写作子智能体生成报告
            self.todo_manager.add_todo("生成研究报告", "medium")
            report = self.delegate_to_subagent("writer", request)
            self.todo_manager.update_status("生成研究报告", "completed")

            result = f"研究完成！\n\n{report}"

        elif "写" in request or "生成" in request or "创建" in request:
            # 委托给写作子智能体
            self.todo_manager.add_todo("执行写作任务", "medium")
            result = self.delegate_to_subagent("writer", request)
            self.todo_manager.update_status("执行写作任务", "completed")

        else:
            # 使用工具直接处理
            self.todo_manager.add_todo("直接处理请求", "medium")
            search_tool = self.tools["search_web"]
            search_results = search_tool.invoke(query=request)

            result = f"找到以下信息:\n\n"
            for i, r in enumerate(search_results, 1):
                result += f"{i}. {r['title']}\n   {r['snippet']}\n\n"

            self.todo_manager.update_status("直接处理请求", "completed")

        # 更新主任务状态
        self.todo_manager.update_status(f"处理请求: {request[:30]}...", "completed")

        return result

# ========== 6. MCP 服务器模拟 ==========

class MCPServer:
    """
    模拟 MCP 服务器
    展示 langsmith-tool-server 的核心功能
    """

    def __init__(self, name: str, tools: List[MCPTool]):
        self.name = name
        self.tools = {tool.name: tool for tool in tools}
        print(f"\n🖥️ MCP 服务器 [{name}] 已启动")
        print(f"   端点: http://localhost:8000/mcp")
        print(f"   工具: {list(self.tools.keys())}")

    def handle_request(self, method: str, params: dict) -> dict:
        """处理 JSON-RPC 请求"""
        if method == "tools/list":
            return {
                "jsonrpc": "2.0",
                "result": [
                    {"name": name, "description": tool.description}
                    for name, tool in self.tools.items()
                ],
                "id": 1
            }

        elif method == "tools/call":
            tool_name = params.get("name")
            arguments = params.get("arguments", {})

            if tool_name not in self.tools:
                return {
                    "jsonrpc": "2.0",
                    "error": {"code": -32601, "message": f"工具不存在: {tool_name}"},
                    "id": 1
                }

            tool = self.tools[tool_name]
            result = tool.invoke(**arguments)

            return {
                "jsonrpc": "2.0",
                "result": result,
                "id": 1
            }

        return {
            "jsonrpc": "2.0",
            "error": {"code": -32601, "message": "方法不存在"},
            "id": 1
        }

# ========== 7. Human-in-the-Loop 模拟 ==========

class HumanInTheLoop:
    """人机协作控制器"""

    def __init__(self, auto_approve: bool = True):
        self.auto_approve = auto_approve
        self.pending_approvals = []

    def request_approval(self, action: str, details: dict) -> str:
        """请求人工审批"""
        print(f"\n⏸️ 等待人工审批:")
        print(f"   操作: {action}")
        print(f"   详情: {details}")

        if self.auto_approve:
            print(f"   ✅ 自动批准（演示模式）")
            return "approved"

        # 在实际应用中，这里会等待用户输入
        self.pending_approvals.append({"action": action, "details": details})
        return "pending"

    def get_pending(self) -> List[dict]:
        """获取待审批项"""
        return self.pending_approvals

# ========== 8. 完整系统集成 ==========

class AgentBuilderSystem:
    """
    完整的 Agent Builder 系统
    整合：协调器、MCP 服务器、人机协作
    """

    def __init__(self):
        # 初始化 MCP 服务器
        self.mcp_server = MCPServer(
            name="research-toolkit",
            tools=[
                SearchWebTool(),
                ReadFileTool(),
                WriteFileTool(),
                SendNotificationTool()
            ]
        )

        # 初始化协调器智能体
        self.agent = OrchestratorAgent()

        # 初始化人机协作控制器
        self.hitl = HumanInTheLoop(auto_approve=True)

        # 对话历史
        self.conversation_history = []

        print("\n" + "="*60)
        print("🚀 Agent Builder 系统已就绪")
        print("="*60)

    def chat(self, user_input: str) -> str:
        """处理用户对话"""
        # 记录对话
        self.conversation_history.append({
            "role": "user",
            "content": user_input
        })

        # 处理请求
        response = self.agent.process_request(user_input)

        # 记录响应
        self.conversation_history.append({
            "role": "assistant",
            "content": response
        })

        return response

    def show_status(self):
        """显示系统状态"""
        print(f"\n{'='*60}")
        print("📊 系统状态")
        print('='*60)

        # 显示任务列表
        self.agent.todo_manager.display()

        # 显示记忆
        print(f"\n💾 记忆存储 ({len(self.agent.memory)} 项):")
        for key in self.agent.memory.keys():
            print(f"  - {key}")

        # 显示对话历史
        print(f"\n💬 对话历史 ({len(self.conversation_history)} 条消息)")

    def demo_mcp_calls(self):
        """演示 MCP 调用"""
        print(f"\n{'='*60}")
        print("🔌 MCP 服务器调用演示")
        print('='*60)

        # 列出工具
        response = self.mcp_server.handle_request("tools/list", {})
        print(f"\n📋 可用工具:")
        for tool in response["result"]:
            print(f"  - {tool['name']}: {tool['description']}")

        # 调用搜索工具
        response = self.mcp_server.handle_request(
            "tools/call",
            {"name": "search_web", "arguments": {"query": "LangGraph 教程"}}
        )
        print(f"\n🔍 搜索结果:")
        for i, result in enumerate(response["result"], 1):
            print(f"  {i}. {result['title']}")

# ========== 9. 主程序 ==========

if __name__ == "__main__":
    print("="*60)
    print("🎯 Agent Builder 完整示例")
    print("="*60)

    # 创建系统
    system = AgentBuilderSystem()

    # 演示 MCP 调用
    system.demo_mcp_calls()

    # 对话测试
    print(f"\n{'='*60}")
    print("💬 开始对话测试")
    print('='*60)

    # 测试 1: 研究任务
    response1 = system.chat("研究 LangGraph 的核心特性和最佳实践")
    print(f"\n🤖 响应:\n{response1[:500]}...")

    # 测试 2: 写作任务
    response2 = system.chat("创建一份关于 AI Agent 的简报")
    print(f"\n🤖 响应:\n{response2[:500]}...")

    # 显示系统状态
    system.show_status()

    # 演示人机协作
    print(f"\n{'='*60}")
    print("👥 Human-in-the-Loop 演示")
    print('='*60)

    hitl = HumanInTheLoop(auto_approve=False)
    result = hitl.request_approval(
        action="发送邮件",
        details={
            "to": "team@example.com",
            "subject": "研究报告已完成",
            "content": "请查收附件中的研究报告..."
        }
    )
    print(f"审批结果: {result}")

    # 最终输出
    print(f"\n{'='*60}")
    print("✅ 演示完成")
    print('='*60)

    print("""
📋 核心知识点回顾:

| 概念 | 说明 | 代码示例 |
|------|------|----------|
| **Agent Builder** | 无代码智能体构建平台 | LangSmith UI 操作 |
| **MCP 协议** | 工具服务器标准协议 | `langsmith-tool-server` |
| **@tool 装饰器** | 定义 MCP 工具 | `@tool def func()` |
| **子智能体** | 专门化任务处理 | `SubAgent.invoke(task)` |
| **任务管理** | 待办事项追踪 | `TodoListManager` |
| **Human-in-the-Loop** | 人工审批工作流 | `request_approval()` |
| **OAuth 认证** | 工具授权 | `auth_provider="google"` |
| **MCP 网关** | 聚合多个工具服务器 | `toolkit.toml` 配置 |

💡 提示：
1. Agent Builder 目前处于 Beta 阶段
2. 必须配置 ANTHROPIC_API_KEY 才能使用
3. Slack 集成免费，费用来自智能体执行
4. 可通过自定义 MCP 服务器扩展功能
""")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589

运行结果示例

============================================================
🎯 Agent Builder 完整示例
============================================================

🖥️ MCP 服务器 [research-toolkit] 已启动
   端点: http://localhost:8000/mcp
   工具: ['search_web', 'read_file', 'write_file', 'send_notification']
🤖 子智能体 [research-agent] 已创建
   工具: ['search_web', 'read_file']
🤖 子智能体 [writer-agent] 已创建
   工具: ['write_file', 'send_notification']

🎯 协调器智能体已初始化
   子智能体: ['research', 'writer']
   工具: ['search_web', 'read_file', 'write_file', 'send_notification']

============================================================
🚀 Agent Builder 系统已就绪
============================================================

============================================================
🔌 MCP 服务器调用演示
============================================================

📋 可用工具:
  - search_web: 搜索网络信息
  - read_file: 读取文件内容
  - write_file: 写入文件内容
  - send_notification: 发送用户通知

🔍 搜索完成: LangGraph 教程 (找到 3 条结果)

🔍 搜索结果:
  1. 关于 LangGraph 教程 的权威指南
  2. LangGraph 教程 入门教程
  3. LangGraph 教程 最新进展

============================================================
💬 开始对话测试
============================================================

============================================================
👤 用户请求: 研究 LangGraph 的核心特性和最佳实践
============================================================
📋 添加任务: 处理请求: 研究 LangGraph 的核心特性和最佳实践... (优先级: high)
✅ 任务状态更新: 处理请求: 研究 LangGraph 的核心特性和最佳实践... -> in_progress
📋 添加任务: 执行研究任务 (优先级: medium)

📤 委托任务给 [research]

📚 研究子智能体开始处理: 研究 LangGraph 的核心特性和最佳实践...
🔍 搜索完成: 研究 LangGraph 的核心特性和最佳实践 (找到 3 条结果)
✅ 任务状态更新: 执行研究任务 -> completed
💾 保存到记忆: research_研究 LangGraph 的核心特性
📋 添加任务: 生成研究报告 (优先级: medium)

📤 委托任务给 [writer]

✍️ 写作子智能体开始处理: 研究 LangGraph 的核心特性和最佳实践...
📝 写入文件: /reports/研究_LangGraph_的核心特性和最佳实践.md
🔔 通知 [slack]: 报告已生成: /reports/研究_LangGraph_的核心特性和最佳实践.md
✅ 任务状态更新: 生成研究报告 -> completed
✅ 任务状态更新: 处理请求: 研究 LangGraph 的核心特性和最佳实践... -> completed

============================================================
📊 系统状态
============================================================

📋 任务列表:
  ✅ 1. 处理请求: 研究 LangGraph 的核心特性和最佳实践...
  ✅ 2. 执行研究任务
  ✅ 3. 生成研究报告
  ✅ 4. 处理请求: 创建一份关于 AI Agent 的简报...
  ✅ 5. 执行写作任务

💾 记忆存储 (1 项):
  - research_研究 LangGraph 的核心特性

💬 对话历史 (4 条消息)

============================================================
👥 Human-in-the-Loop 演示
============================================================

⏸️ 等待人工审批:
   操作: 发送邮件
   详情: {'to': 'team@example.com', 'subject': '研究报告已完成', ...}
审批结果: pending

============================================================
✅ 演示完成
============================================================

扩展阅读

总结：Agent Builder 是 LangSmith 提供的革命性无代码智能体构建平台，它将复杂的 AI 智能体开发简化为自然语言对话。通过 MCP 协议，用户可以轻松集成内置工具或自定义工具服务器。无论是邮件自动化、Slack 机器人还是研究助手，Agent Builder 都能帮助你快速构建生产级的 AI 应用。

2.4.2 Agent Builder - 无代码智能体构建器 ​

目录 ​

概述 ​

什么是 Agent Builder? ​

架构概览 ​

与传统方式的对比 ​

核心概念 ​

1. Prompt（提示词） ​

2. Tools（工具） ​

3. Triggers（触发器） ​

4. Sub-agents（子智能体） ​

5. Memory（记忆系统） ​

6. Human-in-the-Loop（人机协作） ​

环境配置 ​

工作区密钥设置 ​

配置步骤 ​

必需的密钥 ​

可选的工具密钥 ​

内置工具 ​

通讯与消息 ​

Gmail ​

Slack ​

日历与项目管理 ​

Google Calendar ​

Linear ​

搜索与发现 ​

社交媒体 ​

LinkedIn ​

Twitter/X ​

实用工具 ​

认证方式 ​

MCP 框架 ​

什么是 MCP? ​

安装 ​

创建工具包 ​

定义工具 ​

运行服务器 ​

客户端调用 ​

OAuth 认证工具 ​

自定义认证 ​

MCP 网关配置 ​

Slack 应用集成 ​

安装步骤 ​

核心功能 ​

权限要求 ​

数据与隐私 ​

费用说明 ​

Deep Agents 底层框架 ​

核心设计原则 ​

内置工具 ​

中间件系统 ​

后端存储选项 ​

完整案例代码 ​

运行结果示例 ​

扩展阅读 ​

2.4.2 Agent Builder - 无代码智能体构建器

目录

概述

什么是 Agent Builder?

架构概览

与传统方式的对比

核心概念

1. Prompt（提示词）

2. Tools（工具）

3. Triggers（触发器）

4. Sub-agents（子智能体）

5. Memory（记忆系统）

6. Human-in-the-Loop（人机协作）

环境配置

工作区密钥设置

配置步骤

必需的密钥

可选的工具密钥

内置工具

通讯与消息

Gmail

Slack

日历与项目管理

Google Calendar

Linear

搜索与发现

社交媒体

LinkedIn

Twitter/X

实用工具

认证方式

MCP 框架

什么是 MCP?

安装

创建工具包

定义工具

运行服务器

客户端调用

OAuth 认证工具

自定义认证

MCP 网关配置

Slack 应用集成

安装步骤

核心功能

权限要求

数据与隐私

费用说明

Deep Agents 底层框架

核心设计原则

内置工具

中间件系统

后端存储选项

完整案例代码

运行结果示例

扩展阅读