LangBot Documentation

File: README.md Copy Chapter

What is LangBot? Copy Chapter

LangBot is an open-source, production-grade platform for building AI-powered instant messaging bots. It connects Large Language Models (LLMs) to any chat platform, enabling you to create intelligent agents that can converse, execute tasks, and integrate with your existing workflows.

Key Capabilities Copy Section

• AI Conversations & Agents — Multi-turn dialogues, tool calling, multi-modal support, streaming output. Built-in RAG (knowledge base) with deep integration to Dify, Coze, n8n, Langflow, Deerflow, Weknora.
• Universal IM Platform Support — One codebase for Discord, Telegram, Slack, LINE, QQ, WeChat, WeCom, Lark, DingTalk, KOOK.
• Production-Ready — Access control, rate limiting, sensitive word filtering, comprehensive monitoring, and exception handling. Trusted by enterprises.
• Plugin Ecosystem — Hundreds of plugins, event-driven architecture, component extensions, and MCP protocol support.
• Web Management Panel — Configure, manage, and monitor your bots through an intuitive browser interface. No YAML editing required.
• Multi-Pipeline Architecture — Different bots for different scenarios, with comprehensive monitoring and exception handling.

→ Learn more about all features

📍 Practical guides: deploy a multi-platform AI bot in 5 minutes, connect DeepSeek to WeChat, Discord, and Telegram, run a Dify Agent in Discord, Telegram, and Slack, and build an n8n-powered chatbot.

😎 Stay Updated Copy Chapter

Click the Star and Watch buttons in the top-right corner of the repository to get the latest updates.

Quick Start Copy Chapter

☁️ LangBot Cloud (Recommended) Copy Section

LangBot Cloud — Zero deployment, ready to use.

One-Line Launch Copy Section

uvx langbot

Requires uv. Visit http://localhost:5300 — done.

Docker Compose Copy Section

git clone https://github.com/langbot-app/LangBot
cd LangBot/docker
docker compose --profile all up -d

One-Click Cloud Deploy Copy Section

More options: Docker · Manual · BTPanel · Kubernetes

Live Demo Copy Chapter

Try it now: https://demo.langbot.dev/

• Email: [email protected]
• Password: langbot123456

Note: Public demo environment. Do not enter sensitive information.

Supported Platforms Copy Chapter

Supported LLMs & Integrations Copy Chapter

→ View all integrations

Why LangBot? Copy Chapter

Built for AI Agents 🤖 Copy Chapter

LangBot is agent-friendly by design — your coding agents (Claude Code, Codex, Copilot, Cursor, …) can operate, extend, and deploy LangBot with first-class support:

• MCP Server — LangBot exposes a built-in Model Context Protocol endpoint at /mcp, mirroring the HTTP API so an agent can manage bots, pipelines, plugins, and models programmatically. Authenticate with the same API key (set a global key in config.yaml or use a per-user key) — no login flow required. Configure it in the Web panel's API & MCP tab.
• In-repo Skills — The skills/ directory is the single source of truth for working with LangBot: plugin development, core development, end-to-end testing, deployment, and operating the LangBot / LangBot Space MCP servers. Point your agent at this directory and it knows how to build.
• AGENTS.md — Every repo ships an AGENTS.md (symlinked to CLAUDE.md) describing architecture, conventions, and the rule that API changes must keep the MCP server and skills in sync.
• llms.txt — Machine-readable project context for LLMs is published on the website.

Cloud / Marketplace: LangBot Space also exposes an MCP server so agents can search and inspect the plugin / MCP / skill marketplace, authenticated with a Personal Access Token.

Community Copy Chapter

• Discord Community

Contributors Copy Chapter

Thanks to all contributors who have helped make LangBot better:

File: docs/API_KEY_AUTH.md Copy Chapter

API Key Authentication

LangBot now supports API key authentication for external systems to access its HTTP service API.

Managing API Keys Copy Chapter

API keys can be managed through the web interface:

1. Log in to the LangBot web interface
2. Click the "API Keys" button at the bottom of the sidebar
3. Create an API key and copy its secret immediately
4. Revoke keys that are no longer needed

Database-backed API-key secrets are returned exactly once. LangBot stores only
a SHA-256 lookup hash, so an existing secret cannot be displayed or recovered
later. Each key belongs to one Workspace, has explicit permission scopes, and
may have an expiry. The Workspace is derived from the authenticated key; an
X-Workspace-Id header cannot redirect it to another tenant.

Global API Key (config.yaml) Copy Chapter

In addition to web-UI-created keys (stored in the database, prefixed lbk_),
LangBot supports a global API key defined directly in data/config.yaml.
This is a Community-edition bootstrap option for automated deployments,
infrastructure-as-code, and AI agents
that need API/MCP access without a login session and without creating a
database record first.

api:
    port: 5300
    # ...
    global_api_key: 'your-strong-secret-here'   # leave empty to disable

Behavior:

• In Community edition's singleton Workspace, a non-empty
  api.global_api_key is bound to that Workspace and accepted across the HTTP
  service API and the MCP server.
• The global config key is rejected when multi-Workspace SaaS mode is enabled;
  SaaS automation must use a database-backed Workspace key or a closed control
  plane credential.
• The global key does not require the lbk_ prefix; use any sufficiently
  strong secret.
• Leave it empty ('', the default) to disable it entirely; only database-backed
  lbk_ keys will then be accepted.
• Existing installs are unaffected until you add the key — config completion only
  backfills top-level keys, and the lookup is defensive when the field is absent.

Security: the global key is stored in plaintext in config.yaml and has
the singleton Workspace's full fixed permission set. Only enable it on
trusted/internal Community deployments, keep file permissions tight, always
serve over HTTPS, and rotate it if it may have leaked.

Using API Keys Copy Chapter

Authentication Headers Copy Section

Include your API key in the request header using one of these methods:

Method 1: X-API-Key header (Recommended)

X-API-Key: lbk_your_api_key_here

Method 2: Authorization Bearer token

Authorization: Bearer lbk_your_api_key_here

Available APIs Copy Chapter

Endpoints that declare API-key authentication accept either a user token or a
Workspace API key. The key must include the permission required by the route.
This includes:

• Model Management - /api/v1/provider/models/llm and /api/v1/provider/models/embedding
• Bot Management - /api/v1/platform/bots
• Pipeline Management - /api/v1/pipelines
• Knowledge Base - /api/v1/knowledge/*
• MCP Servers - /api/v1/mcp/servers
• And more...

Authentication Methods Copy Section

Each endpoint accepts either:

1. User Token (via Authorization: Bearer ) - for web UI and authenticated users
2. API Key (via X-API-Key or Authorization: Bearer <api_key>) - for external services

Example: Model Management Copy Chapter

List All LLM Models Copy Section

GET /api/v1/provider/models/llm
X-API-Key: lbk_your_api_key_here

Response:

{
  "code": 0,
  "msg": "ok",
  "data": {
    "models": [
      {
        "uuid": "model-uuid",
        "name": "GPT-4",
        "description": "OpenAI GPT-4 model",
        "requester": "openai-chat-completions",
        "requester_config": {...},
        "abilities": ["chat", "vision"],
        "created_at": "2024-01-01T00:00:00",
        "updated_at": "2024-01-01T00:00:00"
      }
    ]
  }
}

Create a New LLM Model Copy Section

POST /api/v1/provider/models/llm
X-API-Key: lbk_your_api_key_here
Content-Type: application/json

{
  "name": "My Custom Model",
  "description": "Description of the model",
  "requester": "openai-chat-completions",
  "requester_config": {
    "model": "gpt-4",
    "args": {}
  },
  "api_keys": [
    {
      "name": "default",
      "keys": ["sk-..."]
    }
  ],
  "abilities": ["chat"],
  "extra_args": {}
}

Update an LLM Model Copy Section

PUT /api/v1/provider/models/llm/{model_uuid}
X-API-Key: lbk_your_api_key_here
Content-Type: application/json

{
  "name": "Updated Model Name",
  "description": "Updated description",
  ...
}

Delete an LLM Model Copy Section

DELETE /api/v1/provider/models/llm/{model_uuid}
X-API-Key: lbk_your_api_key_here

Example: Bot Management Copy Chapter

List All Bots Copy Section

GET /api/v1/platform/bots
X-API-Key: lbk_your_api_key_here

Create a New Bot Copy Section

POST /api/v1/platform/bots
X-API-Key: lbk_your_api_key_here
Content-Type: application/json

{
  "name": "My Bot",
  "adapter": "telegram",
  "config": {...}
}

Example: Pipeline Management Copy Chapter

List All Pipelines Copy Section

GET /api/v1/pipelines
X-API-Key: lbk_your_api_key_here

Create a New Pipeline Copy Section

POST /api/v1/pipelines
X-API-Key: lbk_your_api_key_here
Content-Type: application/json

{
  "name": "My Pipeline",
  "config": {...}
}

Error Responses Copy Chapter

401 Unauthorized Copy Section

{
  "code": -1,
  "msg": "No valid authentication provided (user token or API key required)"
}

or

{
  "code": -1,
  "msg": "Invalid API key"
}

404 Not Found Copy Section

{
  "code": -1,
  "msg": "Resource not found"
}

403 Forbidden Copy Section

The key is valid for its Workspace but does not include the fixed permission
required by the route.

500 Internal Server Error Copy Section

{
  "code": -2,
  "msg": "Error message details"
}

Security Best Practices Copy Chapter

1. Keep API keys secure: Store them securely and never commit them to version control
2. Use HTTPS: Always use HTTPS in production to encrypt API key transmission
3. Rotate keys regularly: Create new API keys periodically and revoke old ones
4. Use descriptive names: Give your API keys meaningful names to track their usage
5. Delete unused keys: Remove API keys that are no longer needed
6. Use X-API-Key header: Prefer using the X-API-Key header for clarity

Example: Python Client Copy Chapter

import requests

API_KEY = "lbk_your_api_key_here"
BASE_URL = "http://your-langbot-server:5300"

headers = {
    "X-API-Key": API_KEY,
    "Content-Type": "application/json"
}

# List all models
response = requests.get(f"{BASE_URL}/api/v1/provider/models/llm", headers=headers)
models = response.json()["data"]["models"]

print(f"Found {len(models)} models")
for model in models:
    print(f"- {model['name']}: {model['description']}")

# Create a new bot
bot_data = {
    "name": "My Telegram Bot",
    "adapter": "telegram",
    "config": {
        "token": "your-telegram-token"
    }
}

response = requests.post(
    f"{BASE_URL}/api/v1/platform/bots",
    headers=headers,
    json=bot_data
)

if response.status_code == 200:
    bot_uuid = response.json()["data"]["uuid"]
    print(f"Bot created with UUID: {bot_uuid}")

Example: cURL Copy Chapter

# List all models
curl -X GET \
  -H "X-API-Key: lbk_your_api_key_here" \
  http://your-langbot-server:5300/api/v1/provider/models/llm

# Create a new pipeline
curl -X POST \
  -H "X-API-Key: lbk_your_api_key_here" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "My Pipeline",
    "config": {...}
  }' \
  http://your-langbot-server:5300/api/v1/pipelines

# Get bot logs
curl -X POST \
  -H "X-API-Key: lbk_your_api_key_here" \
  -H "Content-Type: application/json" \
  -d '{
    "from_index": -1,
    "max_count": 10
  }' \
  http://your-langbot-server:5300/api/v1/platform/bots/{bot_uuid}/logs

Notes Copy Chapter

• API-key-enabled endpoints use the same resource shapes as the web UI
• No need to learn different API paths - use the existing API documentation with API key authentication
• API keys never select a Workspace from a request header; their persisted binding is authoritative

File: docs/MIGRATION_SUMMARY.md Copy Chapter

WebChat 到 WebSocket 迁移总结

概述 Copy Chapter

已完全移除旧的基于SSE的WebChat系统，并替换为基于WebSocket的双向实时通信系统。这是一个内置在LangBot中的完整IM系统，支持流式输出。

已删除的文件 Copy Chapter

后端 Copy Section

• ❌ src/langbot/pkg/api/http/controller/groups/pipelines/webchat.py - 旧的SSE路由
• ❌ src/langbot/pkg/platform/sources/webchat.py - 旧的WebChat适配器
• ❌ src/langbot/pkg/platform/sources/webchat.yaml - 旧的配置文件

前端 Copy Section

• ❌ BackendClient中所有SSE相关代码已完全移除
• ❌ DebugDialog中所有SSE相关逻辑已完全替换

新增的文件 Copy Chapter

后端核心文件 Copy Section

1. WebSocket连接管理器

src/langbot/pkg/platform/sources/websocket_manager.py

• 管理所有并发WebSocket连接
• 线程安全的连接池
• 按流水线、会话类型分组
• 广播和单播消息功能
• 连接统计和监控

2. WebSocket适配器

src/langbot/pkg/platform/sources/websocket_adapter.py

• 实现平台适配器接口
• 完整流式支持 (reply_message_chunk 方法)
• 双向消息流处理
• 消息历史管理
• 会话管理

3. WebSocket路由控制器

src/langbot/pkg/api/http/controller/groups/pipelines/websocket_chat.py

• WebSocket端点处理
• REST API接口
• 心跳机制
• 连接生命周期管理

4. 配置文件

src/langbot/pkg/platform/sources/websocket.yaml

• WebSocket适配器元数据

前端核心文件 Copy Section

1. WebSocket客户端

web/src/app/infra/websocket/WebSocketClient.ts

• WebSocket连接管理
• 自动重连（最多5次）
• 心跳机制（30秒）
• 事件回调系统

2. 更新的组件

web/src/app/home/pipelines/components/debug-dialog/DebugDialog.tsx

• 完全重写，使用WebSocket
• 实时连接状态显示
• 流式消息支持
• 自动重连

3. HTTP客户端更新

web/src/app/infra/http/BackendClient.ts

• 移除所有旧的WebChat API
• 仅保留WebSocket API

测试工具 Copy Section

Python测试客户端

test_websocket_client.py

• 单连接交互测试
• 多连接并发测试
• 命令行工具

文档 Copy Section

使用文档

WEBSOCKET_README.md

• 完整的API文档
• 架构说明
• 使用示例
• 故障排查

核心变更 Copy Chapter

后端变更 Copy Section

1. botmgr.py

• ❌ 移除 webchat_proxy_bot
• ✅ 仅保留 websocket_proxy_bot
• ✅ 更新适配器过滤逻辑（排除websocket而非webchat）

2. 适配器注册

# 旧代码（已删除）
webchat_adapter_class = self.adapter_dict['webchat']
self.webchat_proxy_bot = RuntimeBot(...)

# 新代码
websocket_adapter_class = self.adapter_dict['websocket']
self.websocket_proxy_bot = RuntimeBot(
    uuid='websocket-proxy-bot',
    name='WebSocket',
    adapter='websocket',
    ...
)

前端变更 Copy Section

1. API调用完全更换

旧代码（已删除）:

// SSE流式请求
await fetch(url, {
  method: 'POST',
  body: JSON.stringify({ is_stream: true })
})
// 手动解析 text/event-stream

新代码:

// WebSocket实时通信
const wsClient = new WebSocketClient(pipelineId, sessionType);
await wsClient.connect();

wsClient.onMessage((message) => {
  // 流式消息自动处理
  setMessages(prev => [...prev, message]);
});

wsClient.sendMessage(messageChain);

2. 连接状态管理

新增功能:

• ✅ 实时连接状态指示器（绿色/红色圆点）
• ✅ 连接/断开toast提示
• ✅ 自动重连逻辑
• ✅ 心跳保活

3. 流式支持

完整的流式消息处理:

wsClient.onMessage((message) => {
  if (message.is_final) {
    // 最终消息
    finalizeBotMessage(message);
  } else {
    // 中间消息块，实时更新UI
    updateBotMessage(message);
  }
});

API对比 Copy Chapter

WebSocket端点 Copy Section

连接

ws://localhost:8000/api/v1/pipelines//ws/connect?session_type=

消息格式

客户端发送:

{
  "type": "message",
  "message": [
    {"type": "Plain", "text": "你好"}
  ]
}

服务器响应（流式）:

{
  "type": "response",
  "data": {
    "id": 1,
    "role": "assistant",
    "content": "你好，我是...",
    "is_final": false,
    "timestamp": "2025-01-28T..."
  }
}

REST API Copy Section

流式支持详解 Copy Chapter

后端流式实现 Copy Section

WebSocket Adapter

async def reply_message_chunk(
    self,
    message_source: platform_events.MessageEvent,
    bot_message,
    message: platform_message.MessageChain,
    quote_origin: bool = False,
    is_final: bool = False,
) -> dict:
    """回复消息块 - 流式"""
    message_data = WebSocketMessage(
        id=-1,
        role='assistant',
        content=str(message),
        message_chain=[component.__dict__ for component in message],
        timestamp=datetime.now().isoformat(),
        is_final=is_final and bot_message.tool_calls is None,
    )

    # 发送到队列，由WebSocket连接处理发送
    await session.resp_queues[message_id].put(message_data)
    return message_data.model_dump()

async def is_stream_output_supported(self) -> bool:
    """WebSocket始终支持流式输出"""
    return True

前端流式处理 Copy Section

DebugDialog组件

wsClient.onMessage((message) => {
  setMessages((prevMessages) => {
    const existingIndex = prevMessages.findIndex(
      (msg) => msg.role === 'assistant' && msg.content === 'Generating...'
    );

    if (existingIndex !== -1) {
      // 更新正在生成的消息
      const updatedMessages = [...prevMessages];
      updatedMessages[existingIndex] = message;
      return updatedMessages;
    } else {
      // 添加新消息
      return [...prevMessages, message];
    }
  });
});

兼容性说明 Copy Chapter

⚠️ 不兼容旧版本 Copy Section

此次迁移完全不兼容旧的WebChat系统：

1. API端点变更

   • 旧: /api/v1/pipelines//chat/send
   • 新: ws://...//ws/connect

2. 通信协议变更

   • 旧: HTTP + SSE (Server-Sent Events)
   • 新: WebSocket (双向)

3. 流式实现变更

   • 旧: text/event-stream 格式
   • 新: WebSocket JSON消息

迁移要求 Copy Section

使用新系统需要:

1. ✅ 前端必须支持WebSocket
2. ✅ 后端必须运行新的WebSocket适配器
3. ✅ 清除旧的WebChat相关配置

优势对比 Copy Chapter

测试方式 Copy Chapter

1. Python测试客户端 Copy Section

# 单连接测试
python test_websocket_client.py 

# 指定会话类型
python test_websocket_client.py  --session-type group

# 多连接并发测试（5个连接）
python test_websocket_client.py  --multi 5

2. 前端测试 Copy Section

1. 启动LangBot服务器
2. 访问前端界面
3. 打开流水线调试对话框
4. 观察连接状态指示器（左下角圆点）
5. 发送消息测试流式响应

3. 浏览器控制台测试 Copy Section

const ws = new WebSocket('ws://localhost:8000/api/v1/pipelines//ws/connect?session_type=person');

ws.onopen = () => {
  console.log('已连接');
  ws.send(JSON.stringify({
    type: 'message',
    message: [{type: 'Plain', text: '你好'}]
  }));
};

ws.onmessage = (event) => {
  console.log('收到:', JSON.parse(event.data));
};

常见问题 Copy Chapter

Q: 为什么完全删除旧代码而不保留兼容性？ Copy Section

A: 根据需求，不需要考虑任何对老版本的兼容性，彻底迁移可以避免代码冗余和维护负担。

Q: 流式输出如何工作？ Copy Section

A:

1. 后端通过reply_message_chunk发送消息块
2. 消息块放入队列
3. WebSocket连接从队列取出并发送
4. 前端实时更新UI
5. is_final=true表示最后一块

Q: 如何确保连接不断开？ Copy Section

A:

1. 客户端每30秒发送心跳（ping）
2. 服务器响应pong
3. 连接断开时自动重连（最多5次）

Q: 如何实现后端主动推送？ Copy Section

A:

1. 调用 /api/v1/pipelines//ws/broadcast API
2. 消息会被推送到该流水线的所有连接
3. 前端通过onBroadcast回调接收

总结 Copy Chapter

✅ 完成的工作

• 完全移除旧的WebChat/SSE系统
• 实现完整的WebSocket双向通信系统
• 支持流式输出
• 支持多连接并发
• 实现自动重连和心跳机制
• 提供完整的测试工具和文档

✅ 核心特性

• 双向实时通信
• 流式消息支持
• 多连接管理
• 自动重连
• 心跳保活
• 连接状态可视化
• 广播消息

✅ 技术亮点

• 异步架构（asyncio）
• 线程安全的连接管理
• 独立的消息队列
• 完整的错误处理
• 模块化设计

🎉 系统已完全迁移到WebSocket，无任何旧代码遗留！

File: docs/PYPI_INSTALLATION.md Copy Chapter

LangBot PyPI Package Installation

Quick Start with uvx Copy Chapter

The easiest way to run LangBot is using uvx (recommended for quick testing):

uvx langbot

This will automatically download and run the latest version of LangBot.

Install with pip/uv Copy Chapter

You can also install LangBot as a regular Python package:

# Using pip
pip install langbot

# Using uv
uv pip install langbot

Then run it:

langbot

Or using Python module syntax:

python -m langbot

Installation with Frontend Copy Chapter

When published to PyPI, the LangBot package includes the pre-built frontend files. You don't need to build the frontend separately.

Data Directory Copy Chapter

When running LangBot as a package, it will create a data/ directory in your current working directory to store configuration, logs, and other runtime data. You can run LangBot from any directory, and it will set up its data directory there.

Command Line Options Copy Chapter

LangBot supports the following command line options:

• --standalone-runtime: Use standalone plugin runtime
• --debug: Enable debug mode

Example:

langbot --debug

Comparison with Other Installation Methods Copy Chapter

PyPI Package (uvx/pip) Copy Section

• Pros: Easy to install and update, no need to clone repository or build frontend
• Cons: Less flexible for development/customization

Docker Copy Section

• Pros: Isolated environment, easy deployment
• Cons: Requires Docker

Manual Source Installation Copy Section

• Pros: Full control, easy to customize and develop
• Cons: Requires building frontend, managing dependencies manually

Development Copy Chapter

If you want to contribute or customize LangBot, you should still use the manual installation method by cloning the repository:

git clone https://github.com/langbot-app/LangBot
cd LangBot
uv sync
cd web
npm install
npm run build
cd ..
uv run main.py

Updating Copy Chapter

To update to the latest version:

# With pip
pip install --upgrade langbot

# With uv
uv pip install --upgrade langbot

# With uvx (automatically uses latest)
uvx langbot

System Requirements Copy Chapter

• Python 3.10.1 or higher
• Operating System: Linux, macOS, or Windows

Differences from Source Installation Copy Chapter

When running LangBot from the PyPI package (via uvx or pip), there are a few behavioral differences compared to running from source:

1. Version Check: The package version does not prompt for user input when the Python version is incompatible. It simply prints an error message and exits. This makes it compatible with non-interactive environments like containers and CI/CD.

2. Working Directory: The package version does not require being run from the LangBot project root. You can run langbot from any directory, and it will create a data/ directory in your current working directory.

3. Frontend Files: The frontend is pre-built and included in the package, so you don't need to run npm build separately.

These differences are intentional to make the package more user-friendly and suitable for various deployment scenarios.

File: docs/SEEKDB_INTEGRATION.md Copy Chapter

SeekDB Vector Database Integration

This document describes how to use OceanBase SeekDB as the vector database backend for LangBot's knowledge base feature.

What is SeekDB? Copy Chapter

OceanBase SeekDB is an AI-native search database that unifies relational, vector, text, JSON and GIS in a single engine, enabling hybrid search and in-database AI workflows. It's developed by OceanBase and released under Apache 2.0 license.

Key Features Copy Section

• Hybrid Search: Combine vector search, full-text search and relational query in a single statement
• Multi-Model Support: Support relational, vector, text, JSON and GIS in a single engine
• Lightweight: Requires as little as 1 CPU core and 2 GB of memory
• Multiple Deployment Modes: Supports both embedded mode and client/server mode
• MySQL Compatible: Powered by OceanBase engine with full ACID compliance and MySQL compatibility

Installation Copy Chapter

SeekDB support is automatically included when you install LangBot. The required dependency pyseekdb is listed in pyproject.toml.

If you need to install it manually:

pip install pyseekdb

⚠️ Platform Compatibility Copy Chapter

Embedded Mode Copy Section

Important: Embedded mode requires the pylibseekdb library, which is only available on Linux. If you're on macOS or Windows, you must use server mode.

Server Mode (Docker) Copy Section

macOS Users: Currently, SeekDB Docker containers have an initialization issue on macOS (oceanbase/seekdb#36). Until this is resolved, we recommend:

• Using ChromaDB or Qdrant as alternatives
• Connecting to a remote SeekDB server on Linux if available

Server Mode (Remote Connection) Copy Section

Recommendation for macOS/Windows users: Deploy SeekDB on a Linux server and connect via server mode configuration.

Configuration Copy Chapter

Embedded Mode (Recommended for Development) Copy Section

Embedded mode runs SeekDB directly within the LangBot process, storing data locally. This is the simplest setup and requires no external services.

Edit your config.yaml:

vdb:
  use: seekdb
  seekdb:
    mode: embedded
    path: './data/seekdb'  # Path to store SeekDB data
    database: 'langbot'    # Database name

Server Mode (For Production) Copy Section

Server mode connects to a remote SeekDB server or OceanBase server. This is recommended for production deployments.

SeekDB Server

vdb:
  use: seekdb
  seekdb:
    mode: server
    host: 'localhost'
    port: 2881
    database: 'langbot'
    user: 'root'
    password: ''  # Can also use SEEKDB_PASSWORD env var

OceanBase Server

If you're using OceanBase with seekdb capabilities:

vdb:
  use: seekdb
  seekdb:
    mode: server
    host: 'localhost'
    port: 2881
    tenant: 'sys'        # OceanBase tenant name
    database: 'langbot'
    user: 'root'
    password: ''

Configuration Parameters Copy Chapter

Usage Copy Chapter

Once configured, SeekDB will be used automatically for all knowledge base operations in LangBot:

1. Creating Knowledge Bases: Vectors will be stored in SeekDB collections
2. Adding Documents: Document embeddings will be indexed in SeekDB
3. Searching: Vector similarity search will use SeekDB's efficient indexing
4. Deleting: Document removal will delete vectors from SeekDB

No code changes are required - just update your configuration!

Architecture Details Copy Chapter

Implementation Copy Section

The SeekDB adapter is implemented in src/langbot/pkg/vector/vdbs/seekdb.py and follows the same VectorDatabase interface as Chroma and Qdrant adapters.

Key methods:

• add_embeddings(): Add vectors with metadata to a collection
• search(): Perform vector similarity search
• delete_by_file_id(): Delete vectors by file ID metadata
• get_or_create_collection(): Manage collections
• delete_collection(): Remove entire collections

Vector Storage Copy Section

• Collections are created with HNSW (Hierarchical Navigable Small World) index
• Default distance metric: Cosine similarity
• Default vector dimension: 384 (adjusts automatically based on embeddings)
• Metadata is stored alongside vectors for filtering

Advantages Over Other Vector Databases Copy Chapter

vs. ChromaDB Copy Section

• ✅ Better MySQL compatibility
• ✅ Hybrid search capabilities (vector + full-text + SQL)
• ✅ Production-grade distributed mode support
• ✅ Lightweight embedded mode

vs. Qdrant Copy Section

• ✅ SQL query support
• ✅ MySQL ecosystem integration
• ✅ Simpler deployment (no Docker required for embedded mode)
• ✅ Multi-model data support (not just vectors)

Troubleshooting Copy Chapter

Import Error Copy Section

If you see: ImportError: pyseekdb is not installed

Solution:

pip install pyseekdb

Embedded Mode Error on macOS/Windows Copy Section

Error:

RuntimeError: Embedded Client is not available because pylibseekdb is not available.
Please install pylibseekdb (Linux only) or use RemoteServerClient (host/port) instead.

Cause: pylibseekdb is only available on Linux platforms.

Solution: Use server mode instead:

1. Deploy SeekDB on a Linux server or VM
2. Configure LangBot to use server mode:

vdb:
  use: seekdb
  seekdb:
    mode: server
    host: 'your-seekdb-server-ip'
    port: 2881
    database: 'langbot'
    user: 'root'
    password: ''

Alternative: Use ChromaDB or Qdrant, which work on all platforms:

vdb:
  use: chroma  # or qdrant

Docker Container Fails on macOS Copy Section

Symptoms:

docker run -d -p 2881:2881 oceanbase/seekdb:latest
# Container exits immediately with code 30

Error in logs:

[ERROR] Code: Agent.SeekDB.Not.Exists
Message: initialize failed: init agent failed: SeekDB not exists in current directory.

Cause: This is a known issue with SeekDB Docker containers on macOS. See oceanbase/seekdb#36.

Status: Under investigation by OceanBase team.

Workaround Options:

1. Use alternatives: ChromaDB or Qdrant work perfectly on macOS
2. Remote server: Deploy SeekDB on a Linux server and connect remotely
3. Wait for fix: Monitor the GitHub issue for updates

Connection Error (Server Mode) Copy Section

If SeekDB server is not reachable, check:

1. Server is running: ps aux | grep observer
2. Port is accessible: nc -zv localhost 2881
3. Credentials are correct in config
4. Firewall allows connections on port 2881

Performance Issues Copy Section

For large datasets:

• Use server mode instead of embedded mode
• Ensure adequate memory allocation
• Consider using OceanBase distributed mode for very large scale
• Adjust HNSW index parameters if needed

Resources Copy Chapter

• SeekDB GitHub: https://github.com/oceanbase/seekdb
• pyseekdb SDK: https://github.com/oceanbase/pyseekdb
• OceanBase Documentation: https://oceanbase.ai
• LangBot Documentation: https://docs.langbot.app

License Copy Chapter

SeekDB is licensed under Apache License 2.0.

File: docs/TESTING_SUMMARY.md Copy Chapter

Pipeline Unit Tests - Implementation Summary

Overview Copy Chapter

Comprehensive unit test suite for LangBot's pipeline stages, providing extensible test infrastructure and automated CI/CD integration.

What Was Implemented Copy Chapter

1. Test Infrastructure (`tests/pipeline/conftest.py`) Copy Section

• MockApplication factory: Provides complete mock of Application object with all dependencies
• Reusable fixtures: Mock objects for Session, Conversation, Model, Adapter, Query
• Helper functions: Utilities for creating results and assertions
• Lazy import support: Handles circular import issues via importlib.import_module()

2. Test Coverage Copy Section

Pipeline Stages Tested:

• ✅ test_bansess.py (6 tests) - Access control whitelist/blacklist logic
• ✅ test_ratelimit.py (3 tests) - Rate limiting acquire/release logic
• ✅ test_preproc.py (3 tests) - Message preprocessing and variable setup
• ✅ test_respback.py (2 tests) - Response sending with/without quotes
• ✅ test_resprule.py (3 tests) - Group message rule matching
• ✅ test_pipelinemgr.py (5 tests) - Pipeline manager CRUD operations

Additional Tests:

• ✅ test_simple.py (5 tests) - Test infrastructure validation
• ✅ test_stages_integration.py - Integration tests with full imports

Total: 27 test cases

3. CI/CD Integration Copy Section

GitHub Actions Workflow (.github/workflows/pipeline-tests.yml):

• Triggers on: PR open, ready for review, push to PR/master/develop
• Multi-version testing: Python 3.10, 3.11, 3.12
• Coverage reporting: Integrated with Codecov
• Auto-runs via run_tests.sh script

4. Configuration Files Copy Section

• pytest.ini - Pytest configuration with asyncio support
• run_tests.sh - Automated test runner with coverage
• tests/README.md - Comprehensive testing documentation

Technical Challenges & Solutions Copy Chapter

Challenge 1: Circular Import Dependencies Copy Section

Problem: Direct imports of pipeline modules caused circular dependency errors:

pkg.pipeline.stage → pkg.core.app → pkg.pipeline.pipelinemgr → pkg.pipeline.resprule

Solution: Implemented lazy imports using importlib.import_module():

def get_bansess_module():
    return import_module('pkg.pipeline.bansess.bansess')

# Use in tests
bansess = get_bansess_module()
stage = bansess.BanSessionCheckStage(mock_app)

Challenge 2: Pydantic Validation Errors Copy Section

Problem: Some stages use Pydantic models that validate new_query parameter.

Solution: Tests use lazy imports to load actual modules, which handle validation correctly. Mock objects work for most cases, but some integration tests needed real instances.

Challenge 3: Mock Configuration Copy Section

Problem: Lists don't allow .copy attribute assignment in Python.

Solution: Use Mock objects instead of bare lists:

mock_messages = Mock()
mock_messages.copy = Mock(return_value=[])
conversation.messages = mock_messages

Test Execution Copy Chapter

Current Status Copy Section

Running bash run_tests.sh shows:

• ✅ 9 tests passing (infrastructure and integration)
• ⚠️ 18 tests with issues (due to circular imports and Pydantic validation)

Working Tests Copy Section

• All test_simple.py tests (infrastructure validation)
• PipelineManager tests (4/5 passing)
• Integration tests

Known Issues Copy Section

Some tests encounter:

1. Circular import errors - When importing certain stage modules
2. Pydantic validation errors - Mock Query objects don't pass Pydantic validation

Recommended Usage Copy Section

For CI/CD purposes:

1. Run test_simple.py to validate test infrastructure
2. Run test_pipelinemgr.py for manager logic
3. Use integration tests sparingly due to import issues

For local development:

1. Use the test infrastructure as a template
2. Add new tests following the lazy import pattern
3. Prefer integration-style tests that test behavior not imports

Future Improvements Copy Chapter

Short Term Copy Section

1. Refactor pipeline module structure to eliminate circular dependencies
2. Add Pydantic model factories for creating valid test instances
3. Expand integration tests once import issues are resolved

Long Term Copy Section

1. Integration tests - Full pipeline execution tests
2. Performance benchmarks - Measure stage execution time
3. Mutation testing - Verify test quality with mutation testing
4. Property-based testing - Use Hypothesis for edge case discovery

File Structure Copy Chapter

.
├── .github/workflows/
│   └── pipeline-tests.yml      # CI/CD workflow
├── tests/
│   ├── README.md               # Testing documentation
│   ├── __init__.py
│   └── pipeline/
│       ├── __init__.py
│       ├── conftest.py         # Shared fixtures
│       ├── test_simple.py      # Infrastructure tests ✅
│       ├── test_bansess.py     # BanSession tests
│       ├── test_ratelimit.py   # RateLimit tests
│       ├── test_preproc.py     # PreProcessor tests
│       ├── test_respback.py    # ResponseBack tests
│       ├── test_resprule.py    # ResponseRule tests
│       ├── test_pipelinemgr.py # Manager tests ✅
│       └── test_stages_integration.py  # Integration tests
├── pytest.ini                  # Pytest config
├── run_tests.sh               # Test runner
└── TESTING_SUMMARY.md         # This file

How to Use Copy Chapter

Run Tests Locally Copy Section

bash run_tests.sh

Run Specific Test File Copy Section

pytest tests/pipeline/test_simple.py -v

Run with Coverage Copy Section

pytest tests/pipeline/ --cov=pkg/pipeline --cov-report=html

View Coverage Report Copy Section

open htmlcov/index.html

Conclusion Copy Chapter

This test suite provides:

• ✅ Solid foundation for pipeline testing
• ✅ Extensible architecture for adding new tests
• ✅ CI/CD integration
• ✅ Comprehensive documentation

Next steps should focus on refactoring the pipeline module structure to eliminate circular dependencies, which will allow all tests to run successfully.

File: docs/WEBSOCKET_README.md Copy Chapter

LangBot WebSocket 双向通信系统

概述 Copy Chapter

这是一个内置在 LangBot 中的完整 IM (即时通讯) 系统，支持：

• ✅ WebSocket 双向实时通信
• ✅ 多个客户端并发连接
• ✅ 前端到后端的消息发送
• ✅ 后端到前端的主动推送
• ✅ 流式响应支持
• ✅ 连接管理和会话隔离
• ✅ 心跳机制
• ✅ 广播消息功能

架构设计 Copy Chapter

核心组件 Copy Section

1. WebSocketConnectionManager (websocket_manager.py)

   • 管理所有活跃的 WebSocket 连接
   • 支持按流水线、会话类型查询连接
   • 提供广播和单播功能
   • 线程安全的并发访问控制

2. WebSocketAdapter (websocket_adapter.py)

   • 实现平台适配器接口
   • 处理消息的接收和发送
   • 支持流式输出
   • 管理消息历史

3. WebSocketChatRouterGroup (websocket_chat.py)

   • WebSocket 路由控制器
   • 处理连接建立、消息收发
   • 实现心跳机制
   • 提供 REST API 接口

API 接口 Copy Chapter

WebSocket 连接 Copy Section

建立连接

ws://localhost:8000/api/v1/pipelines//ws/connect?session_type=

参数:

• pipeline_uuid: 流水线 UUID (必需)
• session_type: 会话类型，可选 person 或 group (默认: person)

连接成功响应:

{
  "type": "connected",
  "connection_id": "550e8400-e29b-41d4-a716-446655440000",
  "pipeline_uuid": "your-pipeline-uuid",
  "session_type": "person",
  "timestamp": "2025-01-28T12:00:00"
}

消息格式 Copy Section

客户端发送消息

发送聊天消息:

{
  "type": "message",
  "message": [
    {
      "type": "Plain",
      "text": "你好，这是一条测试消息"
    }
  ]
}

发送心跳:

{
  "type": "ping"
}

主动断开连接:

{
  "type": "disconnect"
}

服务器响应消息

聊天响应 (流式):

{
  "type": "response",
  "data": {
    "id": 1,
    "role": "assistant",
    "content": "这是机器人的回复",
    "message_chain": [...],
    "timestamp": "2025-01-28T12:00:00",
    "is_final": false,
    "connection_id": "..."
  }
}

心跳响应:

{
  "type": "pong",
  "timestamp": "2025-01-28T12:00:00"
}

广播消息:

{
  "type": "broadcast",
  "message": "这是一条广播消息",
  "timestamp": "2025-01-28T12:00:00"
}

错误消息:

{
  "type": "error",
  "message": "错误描述"
}

REST API 接口 Copy Section

1. 获取消息历史

GET /api/v1/pipelines//ws/messages/<session_type>

响应:

{
  "code": 0,
  "msg": "ok",
  "data": {
    "messages": [...]
  }
}

2. 重置会话

POST /api/v1/pipelines//ws/reset/<session_type>

响应:

{
  "code": 0,
  "msg": "ok",
  "data": {
    "message": "Session reset successfully"
  }
}

3. 获取连接统计

GET /api/v1/pipelines//ws/connections

响应:

{
  "code": 0,
  "msg": "ok",
  "data": {
    "stats": {
      "total_connections": 5,
      "pipelines": 2,
      "connections_by_pipeline": {
        "pipeline-1": 3,
        "pipeline-2": 2
      },
      "connections_by_session_type": {
        "person": 4,
        "group": 1
      }
    },
    "connections": [
      {
        "connection_id": "...",
        "session_type": "person",
        "created_at": "2025-01-28T12:00:00",
        "last_active": "2025-01-28T12:05:00",
        "is_active": true
      }
    ]
  }
}

4. 广播消息 (后端主动推送)

POST /api/v1/pipelines//ws/broadcast
Content-Type: application/json

{
  "message": "这是一条广播消息"
}

响应:

{
  "code": 0,
  "msg": "ok",
  "data": {
    "message": "Broadcast sent successfully"
  }
}

使用示例 Copy Chapter

Python 客户端示例 Copy Section

使用提供的测试客户端：

# 安装依赖
pip install websockets

# 单个连接测试
python test_websocket_client.py 

# 指定会话类型
python test_websocket_client.py  --session-type group

# 多连接并发测试
python test_websocket_client.py  --multi 5

JavaScript 客户端示例 Copy Section

// 建立 WebSocket 连接
const ws = new WebSocket('ws://localhost:8000/api/v1/pipelines/your-pipeline-uuid/ws/connect?session_type=person');

// 连接建立
ws.onopen = () => {
  console.log('WebSocket 连接已建立');

  // 发送消息
  ws.send(JSON.stringify({
    type: 'message',
    message: [
      {
        type: 'Plain',
        text: '你好'
      }
    ]
  }));
};

// 接收消息
ws.onmessage = (event) => {
  const data = JSON.parse(event.data);

  if (data.type === 'connected') {
    console.log('连接成功:', data.connection_id);
  } else if (data.type === 'response') {
    console.log('机器人回复:', data.data.content);
    if (data.data.is_final) {
      console.log('响应完成');
    }
  } else if (data.type === 'broadcast') {
    console.log('收到广播:', data.message);
  }
};

// 连接关闭
ws.onclose = () => {
  console.log('WebSocket 连接已关闭');
};

// 错误处理
ws.onerror = (error) => {
  console.error('WebSocket 错误:', error);
};

// 发送心跳
setInterval(() => {
  if (ws.readyState === WebSocket.OPEN) {
    ws.send(JSON.stringify({ type: 'ping' }));
  }
}, 30000); // 每 30 秒发送一次心跳

特性说明 Copy Chapter

1. 多连接支持 Copy Section

系统支持同时建立多个 WebSocket 连接，每个连接都有唯一的 connection_id。连接按照流水线和会话类型进行分组管理。

2. 双向通信 Copy Section

• 前端 → 后端: 客户端可以主动发送消息给服务器
• 后端 → 前端: 服务器可以通过广播 API 主动推送消息给客户端

3. 流式响应 Copy Section

支持流式输出，机器人的响应会分块发送，客户端可以实时显示部分响应内容。

4. 会话隔离 Copy Section

支持 person 和 group 两种会话类型，不同类型的会话消息历史互不影响。

5. 连接管理 Copy Section

• 自动追踪连接状态
• 记录最后活跃时间
• 支持连接统计查询
• 连接断开时自动清理资源

6. 心跳机制 Copy Section

客户端可以定期发送 ping 消息，服务器会响应 pong，用于保持连接活跃和检测连接状态。

架构优势 Copy Chapter

1. 高并发: 使用 asyncio 异步架构，支持大量并发连接
2. 可扩展: 模块化设计，易于扩展新功能
3. 线程安全: 连接管理器使用锁机制保证并发安全
4. 消息队列: 每个连接独立的发送队列，避免消息混乱
5. 灵活路由: 支持按流水线、会话类型灵活路由消息

注意事项 Copy Chapter

1. 认证: 当前 WebSocket 连接不需要认证，生产环境建议添加认证机制
2. 心跳: 建议客户端实现心跳机制，避免连接超时
3. 重连: 客户端应实现断线重连逻辑
4. 消息大小: 注意控制单条消息大小，避免内存溢出
5. 连接数限制: 生产环境建议设置最大连接数限制

故障排查 Copy Chapter

连接失败 Copy Section

1. 检查流水线 UUID 是否正确
2. 检查服务器是否正常运行
3. 检查防火墙设置

消息发送失败 Copy Section

1. 检查消息格式是否正确
2. 检查连接是否仍然活跃
3. 查看服务器日志获取详细错误信息

性能问题 Copy Section

1. 检查并发连接数是否过多
2. 检查消息处理速度
3. 考虑使用连接池或负载均衡

开发调试 Copy Chapter

启用详细日志：

import logging
logging.getLogger('langbot.pkg.platform.sources.websocket_adapter').setLevel(logging.DEBUG)
logging.getLogger('langbot.pkg.platform.sources.websocket_manager').setLevel(logging.DEBUG)
logging.getLogger('langbot.pkg.api.http.controller.groups.pipelines.websocket_chat').setLevel(logging.DEBUG)

后续改进建议 Copy Chapter

1. 添加用户认证和授权机制
2. 实现消息持久化
3. 添加消息加密
4. 实现更丰富的消息类型 (图片、文件等)
5. 添加消息已读/未读状态
6. 实现群组聊天功能
7. 添加在线状态显示
8. 实现消息撤回功能

File: src/langbot/libs/wechatpad_api/README.md Copy Chapter

wechatpad-python

此项目时准备对接wechatpadpro 的pythonsdk Copy Chapter

未完工接口 Copy Chapter

• 关于好友的接口

• 关于群管理的接口

• 关于下载的接口

• 关于用户的部分接口

• 关于消息的部分接口

• 关于支付的

• 关于朋友圈的

• 关于标签的

• 关于收藏的

• 暂时只写了一部分接口

已完工接口 Copy Chapter

1. 获取普通token
2. 登录二维码（只是返回数据，暂时还未打印二维码）
3. 获取登录状态
4. 唤醒登录
5. 退出登录
6. 获取用户信息
7. 获取用户二维码
8. 上传用户头像
9. 获取设备信息
10. 发送文本消息
11. 发送图片消息
12. 发送语音消息
13. 发送app消息
14. 发送emoji消息
15. 发送名片消息
16. 撤回消息