Skip to main content
本指南提供了开始使用 Drasi 工具的快速概览。有关所有 Drasi 功能、参数和配置的详细说明，请访问 Drasi 文档 以及 langchain_drasi 代码库。

​
概述

Drasi 是一个变更检测平台，能够轻松高效地检测和响应数据库中的变更。LangChain-Drasi 集成通过将外部数据变更与工作流执行相连接，构建出响应式、事件驱动的 AI Agent。这使得 Agent 能够发现、订阅并响应实时查询更新，从而将外部数据变更与 Agentic 工作流相衔接。Drasi 持续查询以流式方式实时推送更新，触发 Agent 状态转换、修改记忆或动态控制工作流执行——将静态 Agent 转化为长期运行的环境感知响应系统。

​
详情

类包可序列化JS 支持下载量版本
DrasiToollangchain-drasi❌❌PyPI - DownloadsPyPI - Version

​
功能

  • 查询发现 - 自动识别可用的 Drasi 查询
  • 实时订阅 - 监控持续查询更新
  • 通知处理器 - 六种内置处理器，适用于不同场景
    • Console（控制台）
    • Logging（日志）
    • Memory（内存）
    • Buffer（缓冲区）
    • LangChain Memory（LangChain 记忆）
    • LangGraph Memory（LangGraph 记忆）
  • 自定义处理器 - 扩展基础处理器以实现特定领域逻辑

​
设置

要使用 Drasi 工具，您需要运行 Drasi 和 Drasi MCP 服务器。

​
前置条件

​
凭证（可选）

如果您的 Drasi MCP 服务器需要身份验证，可以通过 Bearer token 或其他认证方式配置请求头：
Configure authentication
from langchain_drasi import MCPConnectionConfig

config = MCPConnectionConfig(
    server_url="http://localhost:8083",
    headers={"Authorization": "Bearer your-token"},
    timeout=30.0
)

​
安装

Drasi 工具位于 langchain-drasi 包中： 
pip install -U langchain-drasi

​
实例化

现在可以实例化 Drasi 工具。您需要配置 MCP 连接，并可选择性地添加通知处理器来处理实时更新：
Initialize tool instance
from langchain_drasi import create_drasi_tool, MCPConnectionConfig, ConsoleHandler

# Configure connection to Drasi MCP server
config = MCPConnectionConfig(
    server_url="http://localhost:8083",
    timeout=30.0
)

# Create a notification handler
handler = ConsoleHandler()

# Create the tool
tool = create_drasi_tool(
    mcp_config=config,
    notification_handlers=[handler]
)

​
调用

​
直接调用

以下是直接调用工具的简单示例。
Call tool
# Discover available queries
queries = await tool.discover_queries()
# Returns: [QueryInfo, QueryInfo, ...]

# Subscribe to a specific query
await tool.subscribe("hot-freezers")
# Notifications routed to registered handlers

# Read current results from a query
result = await tool.read_query("active-orders")
# Returns: QueryResult with current data

​
作为 ToolCall 调用

我们也可以使用模型生成的 ToolCall 调用工具，此时将返回一个 ToolMessage。

​
在 Agent 中使用

我们可以在 LangGraph Agent 中使用 Drasi 工具，构建响应式事件驱动工作流。为此，我们需要一个支持工具调用的模型。
Agent with tool
from langchain_anthropic import ChatAnthropic
from langchain.agents import create_agent

# Initialize the model
model = ChatAnthropic(model="claude-sonnet-4-6")

# Create agent with Drasi tool
agent = create_agent(model, [tool])

# Run the agent
result = agent.invoke(
    {"messages": [{"role": "user", "content": "What queries are available?"}]}
)

print(result["messages"][-1].content)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Subscribe to the customer-orders query"}]}
)

print(result["messages"][-1].content)

​
通知处理器

Drasi 的核心功能之一是内置的通知处理器，用于处理实时查询结果的变更。您可以使用这些处理器针对数据变更采取特定操作。

​
内置处理器

ConsoleHandler - 将格式化的通知输出到标准输出： 
from langchain_drasi import ConsoleHandler

handler = ConsoleHandler()
LoggingHandler - 使用 Python 的 logging 框架记录通知： 
from langchain_drasi import LoggingHandler
import logging

handler = LoggingHandler(
    logger_name="drasi.notifications",
    log_level=logging.INFO
)
MemoryHandler - 将通知存储在内存中，支持可选过滤： 
from langchain_drasi import MemoryHandler

handler = MemoryHandler(max_size=100)

# Retrieve notifications
all_notifs = handler.get_all()
freezer_notifs = handler.get_by_query("hot-freezers")
added_events = handler.get_by_type("added")
BufferHandler - 用于顺序处理的 FIFO 队列： 当工作流正在处理其他任务时，该处理器可用于缓冲传入的变更通知；之后可在工作流中使用循环来消费缓冲区中的通知。 
from langchain_drasi import BufferHandler

handler = BufferHandler(max_size=100)
# Later, consume notifications
notification = handler.consume()  # Remove and return next notification
notification = handler.peek()     # View next notification without removing
LangGraphMemoryHandler - 将更新直接注入 LangGraph 检查点： 
from langchain_drasi import LangGraphMemoryHandler
from langgraph.checkpoint.memory import MemorySaver

checkpoint_manager = MemorySaver()
handler = LangGraphMemoryHandler(
    checkpointer=checkpoint_manager,
    thread_id="your-thread-id"
)

​
自定义处理器

您可以通过扩展 BaseDrasiNotificationHandler 创建自定义处理器： 
from langchain_drasi import BaseDrasiNotificationHandler

class CustomHandler(BaseDrasiNotificationHandler):
    def on_result_added(self, query_name: str, added_data: dict):
        # Handle new results
        print(f"New result in {query_name}: {added_data}")

    def on_result_updated(self, query_name: str, updated_data: dict):
        # Handle updated results
        print(f"Updated result in {query_name}: {updated_data}")

    def on_result_deleted(self, query_name: str, deleted_data: dict):
        # Handle deleted results
        print(f"Deleted result in {query_name}: {deleted_data}")

handler = CustomHandler()
tool = create_drasi_tool(
    mcp_config=config,
    notification_handlers=[handler]
)

​
示例

  • 交互式聊天：一个使用 Drasi 进行实时内存更新的聊天应用程序。
  • 终结者游戏：一个利用 Drasi 实现动态 NPC 行为的游戏。

​
使用场景

Drasi 特别适用于构建需要响应实时数据变更的环境感知 Agent。一些典型使用场景包括：
  • AI 副驾驶 - 监控并响应系统事件的助手
  • AI 游戏玩家 - 能够适应游戏内事件的 NPC
  • 物联网监控 - 处理传感器数据流的 Agent
  • 客户支持 - 响应工单更新或客户操作的机器人
  • DevOps 助手 - 监控基础设施变更的工具
  • 协同编辑 - 响应文档或代码变更的系统

​
API 参考

有关所有 Drasi 功能和配置的详细文档，请访问 API 参考。
在 GitHub 上编辑此页面 或 提交问题。
将这些文档连接到 Claude、VSCode 等，通过 MCP 获取实时解答。