Agnost 数据摄取

将数据摄取至 Agnost AI 以支持分析、监控与洞察的完整指南。涵盖用于追踪 AI 交互的 Conversation SDK，以及用于 Model Context Protocol（MCP）服务器分析的 MCP SDK。

官方文档： https://docs.agnost.ai
API 端点： https://api.agnost.ai
仪表板： https://app.agnost.ai

重要提示：如何应用本技能

在实施 Agnost 数据摄取前，请严格遵循以下优先级顺序：

1. 明确使用场景：是对话追踪（AI 聊天机器人、智能体），还是 MCP 服务器分析
2. 查阅 SDK 参考文档：参考 references/ 目录下的详细 API 文档
3. 复用提供的代码示例：将其作为开发起点
4. 引用参考资料：在解释实现细节时注明对应参考来源

快速参考

SDK 包

API 端点

Conversation SDK（AI 应用推荐）

当构建需追踪用户交互、输入输出及性能指标的 AI 应用、聊天机器人或智能体时，请使用 Conversation SDK。

Python 安装与初始化

# 安装
pip install agnost
# 或
uv add agnost

# 基础初始化
import agnost

# 使用组织 ID（来自仪表板）初始化
agnost.init("your-org-id")

TypeScript/Node.js 安装与初始化

// 安装
npm install agnostai
// 或
pnpm add agnostai

// 基础初始化
import * as agnost from "agnostai";

// 使用组织 ID（来自仪表板）初始化
agnost.init("your-org-id");

核心方法

1. init(org_id, config?) — 初始化 SDK

必须在调用任何追踪方法前执行。

Python

import agnost

# 基础初始化
agnost.init("your-org-id")

# 带配置初始化
agnost.init(
    "your-org-id",
    endpoint="https://api.agnost.ai",  # 自定义端点（可选）
    debug=True                          # 启用调试日志
)

TypeScript

import * as agnost from "agnostai";

// 基础初始化
agnost.init("your-org-id");

// 带配置初始化
agnost.init("your-org-id", {
  endpoint: "https://api.agnost.ai",  // 自定义端点（可选）
  debug: true                          // 启用调试日志
});

2. begin() + end() — 追踪交互（推荐方式）

采用 begin/end 模式可实现自动延迟计算，并使代码更清晰。

Python

import agnost

agnost.init("your-org-id")

# 开始追踪一次交互
interaction = agnost.begin(
    user_id="user_123",
    agent_name="weather-agent",
    input="What's the weather in NYC?",
    conversation_id="conv_456",  # 可选：用于聚合相关事件
    properties={"model": "gpt-4"}  # 可选：自定义元数据
)

# ... 此处执行您的 AI 处理逻辑 ...
response = call_your_ai_model(interaction.input)

# 完成交互（延迟自动计算）
interaction.end(
    output=response,
    success=True  # 若调用失败则设为 False
)

TypeScript

import * as agnost from "agnostai";

agnost.init("your-org-id");

// 开始追踪一次交互
const interaction = agnost.begin({
  userId: "user_123",
  agentName: "weather-agent",
  input: "What's the weather in NYC?",
  conversationId: "conv_456",  // 可选：用于聚合相关事件
  properties: { model: "gpt-4" }  // 可选：自定义元数据
});

// ... 此处执行您的 AI 处理逻辑 ...
const response = await callYourAIModel(interaction.input);

// 完成交互（延迟自动计算）
interaction.end(response);  // 或 interaction.end(response, true) 表示成功

3. track() — 单次调用追踪

适用于所有数据已就绪、无需 begin/end 分阶段处理的场景。

Python

import agnost

agnost.init("your-org-id")

agnost.track(
    user_id="user_123",
    input="What's the weather?",
    output="The weather is sunny with 72°F.",
    agent_name="weather-agent",
    conversation_id="conv_456",  # 可选
    success=True,
    latency=150,  # 毫秒
    properties={"model": "gpt-4", "tokens": 42}
)

4. identify() — 用户信息增强

将用户元数据与用户 ID 关联，以支持更丰富的分析能力。

Python

import agnost

agnost.init("your-org-id")

agnost.identify("user_123", {
    "name": "John Doe",
    "email": "john@example.com",
    "plan": "premium",
    "company": "Acme Inc"
})

TypeScript

import * as agnost from "agnostai";

agnost.init("your-org-id");

agnost.identify("user_123", {
  name: "John Doe",
  email: "john@example.com",
  plan: "premium",
  company: "Acme Inc"
});

5. flush() 与 shutdown() — 资源管理

Python

import agnost

# 手动刷新待发送事件
agnost.flush()

# 清洁关闭（自动刷新并关闭连接）
agnost.shutdown()

TypeScript

import * as agnost from "agnostai";

// 手动刷新待发送事件
await agnost.flush();

// 清洁关闭（自动刷新并关闭连接）
await agnost.shutdown();

Interaction 对象方法

使用 begin() 时，您将获得一个 Interaction 对象，其提供如下方法：

示例：动态构建输入（Python）

interaction = agnost.begin(
    user_id="user_123",
    agent_name="my-agent"
)

# 从多个来源组合输入
interaction.set_input("Combined user query: " + user_input)
interaction.set_property("source", "chat-widget")
interaction.set_properties({"model": "gpt-4", "version": "v2"})

# 处理并完成
response = process_query(interaction.input)
interaction.end(output=response)

MCP 服务器分析

针对 Model Context Protocol（MCP）服务器的追踪，请使用 MCP SDK。

Python（FastMCP）

from mcp.server.fastmcp import FastMCP
from agnost_mcp import track, config

# 创建 FastMCP 服务器
mcp = FastMCP("my-mcp-server")

# 添加您的工具
@mcp.tool()
def my_tool(param: str) -> str:
    return f"Result: {param}"

# 启用追踪
track(mcp, "your-org-id", config(
    endpoint="https://api.agnost.ai",
    disable_input=False,   # 追踪输入参数
    disable_output=False   # 追踪输出结果
))

# 启动服务器
mcp.run()

TypeScript（MCP SDK）

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { trackMCP } from "agnost";

// 创建 MCP 服务器
const server = new Server({
  name: "my-mcp-server",
  version: "1.0.0"
}, {
  capabilities: { tools: {} }
});

// 启用追踪
trackMCP(server, "your-org-id", {
  endpoint: "https://api.agnost.ai",
  disableInput: false,
  disableOutput: false
});

// 启动服务器
const transport = new StdioServerTransport();
await server.connect(transport);

Go（mcp-go）

package main

import (
    "github.com/agnostai/agnost-go/agnost"
    "github.com/mark3labs/mcp-go/server"
)

func main() {
    s := server.NewMCPServer("my-server", "1.0.0")

    // 添加工具...

    // 启用追踪
    agnost.Track(s, "your-org-id", &agnost.Config{
        DisableInput:  false,
        DisableOutput: false,
        BatchSize:     10,
        LogLevel:      "info",
    })

    server.ServeStdio(s)
}

API 参考（直连 HTTP）

适用于无需 SDK、需直接调用 API 的场景。

创建会话

curl -X POST https://api.agnost.ai/api/v1/capture-session 
  -H "Content-Type: application/json" 
  -H "X-Org-Id: your-org-id" 
  -d '{
    "session_id": "unique-session-id",
    "client_config": "my-app",
    "connection_type": "http",
    "ip": "",
    "user_data": {
      "user_id": "user_123",
      "email": "user@example.com"
    },
    "tools": ["tool1", "tool2"]
  }'

捕获事件

curl -X POST https://api.agnost.ai/api/v1/capture-event 
  -H "Content-Type: application/json" 
  -H "X-Org-Id: your-org-id" 
  -d '{
    "session_id": "unique-session-id",
    "primitive_type": "tool",
    "primitive_name": "weather_lookup",
    "latency": 150,
    "success": true,
    "args": "{"city": "NYC"}",
    "result": "{"temp": 72}",
    "metadata": {
      "model": "gpt-4",
      "tokens": "42"
    }
  }'

数据结构

会话请求

{
  "session_id": "字符串（UUID 或自定义 ID）",
  "client_config": "字符串（应用标识符）",
  "connection_type": "字符串（http/stdio/sse）",
  "ip": "字符串（可选）",
  "user_data": {
    "user_id": "字符串",
    "...": "任意其他用户字段"
  },
  "tools": ["数组", "包含", "工具", "名称"]
}

事件请求

{
  "session_id": "字符串（须匹配已有会话）",
  "primitive_type": "字符串（tool/resource/prompt）",
  "primitive_name": "字符串（原始操作名称）",
  "latency": "整数（毫秒）",
  "success": "布尔值",
  "args": "字符串（JSON 编码的输入）",
  "result": "字符串（JSON 编码的输出）",
  "checkpoints": [
    {
      "name": "字符串",
      "timestamp": "整数（距开始时间的毫秒数）",
      "metadata": {}
    }
  ],
  "metadata": {
    "key": "value 对"
  }
}

配置选项

Python Conversation SDK

agnost.init(
    "your-org-id",
    endpoint="https://api.agnost.ai",  # API 端点
    debug=False                         # 启用调试日志
)

TypeScript Conversation SDK

interface ConversationConfig {
  endpoint?: string;  // API 端点（默认：https://api.agnost.ai）
  debug?: boolean;    // 启用调试日志（默认：false）
}

agnost.init("your-org-id", { endpoint: "...", debug: true });

Python MCP SDK（FastMCP）

from agnost_mcp import track, config

track(server, "your-org-id", config(
    endpoint="https://api.agnost.ai",
    disable_input=False,   # 不追踪输入参数
    disable_output=False   # 不追踪输出结果
))

TypeScript MCP SDK

import { trackMCP, createConfig } from "agnost";

const cfg = createConfig({
  endpoint: "https://api.agnost.ai",
  disableInput: false,
  disableOutput: false
});

trackMCP(server, "your-org-id", cfg);

Go MCP SDK

type Config struct {
    Endpoint         string        // 默认："https://api.agnost.ai"
    DisableInput     bool          // 默认：false
    DisableOutput    bool          // 默认：false
    BatchSize        int           // 默认：5
    MaxRetries       int           // 默认：3
    RetryDelay       time.Duration // 默认：1s
    RequestTimeout   time.Duration // 默认：5s
    LogLevel         string        // "debug", "info", "warning", "error"
    Identify         IdentifyFunc  // 可选：用户识别函数
}

最佳实践

1. 总是在应用启动时尽早初始化

# 在应用启动时执行
import agnost
agnost.init("your-org-id")

2. 使用 begin/end 获取准确延迟

# 自动计算处理耗时
interaction = agnost.begin(user_id="u1", agent_name="agent")
# ... 处理中 ...
interaction.end(output=result)

3. 使用 conversation_id 聚合相关事件

# 同一聊天会话中的全部事件
conversation_id = f"chat_{session_id}"
interaction = agnost.begin(
    user_id="u1",
    conversation_id=conversation_id,
    agent_name="chatbot"
)

4. 健壮地处理错误

interaction = agnost.begin(user_id="u1", agent_name="agent")
try:
    result = process_request()
    interaction.end(output=result, success=True)
except Exception as e:
    interaction.end(output=str(e), success=False)

5. 清洁关闭

import atexit
import agnost

atexit.register(agnost.shutdown)

适用场景

当您遇到以下任一情况时，应启用本技能：

• Agnost 数据摄取的实现
• AI 对话追踪功能的搭建
• MCP 服务器分析的集成
• 事件/会话捕获 API 的使用
• SDK 初始化相关问题
• 延迟追踪需求
• 用户身份识别与信息增强

附加资源

• Python SDK 参考文档: references/python-sdk.md
• TypeScript SDK 参考文档: references/typescript-sdk.md
• API 参考文档: references/api-reference.md
• 仪表板: https://app.agnost.ai
• Discord 社区: https://discord.gg/agnost