调用 Expert 智能体

本教程将带你完成从零开始在外部应用中调用 Profy Expert 的全流程——从获取 Expert 标识到构建完整的流式聊天界面。

你将构建什么

一个可以调用 Profy Expert 的后端服务，并将 SSE 流式响应接入前端聊天界面。完成后，你的应用将具备：

调用任意已发布 Expert 的能力
多轮对话上下文维持
SSE 流式解析与错误处理
按 token 用量自动计费（METERED 模型）

Expert 是什么

Expert 是 Profy 平台上的 AI Agent 产品，由创作者在 Marketplace 上发布。每个 Expert 包含：

组成	说明
Persona	人设与行为风格定义
Tools	可调用的工具集（搜索、代码执行、文件操作等）
Memory	跨会话记忆，持续积累用户偏好与事实
Skills	可复用的结构化技能，定义 Agent 的行为模式

通过 Events API 的 /openapi/v1/events/invoke 端点，你的应用可以像调用一个 API 一样调用这些 Expert。

前置条件

在开始之前，确保你已具备：

Profy App：在 Studio 中创建并配置好 OAuth
OAuth Access Token：通过授权码流程获取（参见 SDK 快速开始）
目标 Expert Identifier：你要调用的 Expert 的唯一标识

POST /openapi/v1/events/invoke 需要 events:write OAuth scope。确保你的 App 在创建时申请了该权限。

Step 1: 找到 Expert Identifier

每个已发布的 Expert 都有一个唯一的 identifier（slug 格式），用于 API 调用。获取方式：

Marketplace 页面 — 打开 Expert 详情页，URL 中的最后一段路径即为 identifier，如 https://app.profy.cn/expert/data-analyst → data-analyst
Studio — 如果你是 Expert 的创作者，在编辑页面的基本信息中可以看到 identifier

建议将 Expert Identifier 存入环境变量或配置文件，避免硬编码。

Step 2: 单轮调用

调用 Expert 的核心是向 /openapi/v1/events/invoke 发送 POST 请求，返回 SSE 流。

const BASE_URL = "https://app.profy.cn";

async function invokeExpert(
  accessToken: string,
  expertIdentifier: string,
  message: string
): Promise<string> {
  const res = await fetch(`${BASE_URL}/openapi/v1/events/invoke`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${accessToken}`,
    },
    body: JSON.stringify({
      expert_identifier: expertIdentifier,
      message,
    }),
  });

  if (!res.ok) {
    throw new Error(`Invoke failed: ${res.status} ${res.statusText}`);
  }

  let fullText = "";
  const reader = res.body!.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    const chunk = decoder.decode(value, { stream: true });
    for (const line of chunk.split("\n")) {
      if (line.startsWith("data: ")) {
        const data = JSON.parse(line.slice(6));
        if (data.type === "text") {
          fullText += data.content;
        }
      }
    }
  }

  return fullText;
}

const answer = await invokeExpert(token.accessToken, "data-analyst", "分析一下最近的销售趋势");
console.log(answer);

import httpx

BASE_URL = "https://app.profy.cn"

async def invoke_expert(
    access_token: str,
    expert_identifier: str,
    message: str,
) -> str:
    full_text = ""

    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST",
            f"{BASE_URL}/openapi/v1/events/invoke",
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {access_token}",
            },
            json={
                "expert_identifier": expert_identifier,
                "message": message,
            },
        ) as response:
            response.raise_for_status()

            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    import json
                    data = json.loads(line[6:])
                    if data["type"] == "text":
                        full_text += data["content"]

    return full_text

import asyncio
answer = asyncio.run(
    invoke_expert(access_token, "data-analyst", "分析一下最近的销售趋势")
)
print(answer)

Step 3: 多轮对话

通过传递 session_id 参数，Expert 会在同一会话上下文中延续对话，保留之前的消息历史和记忆。

import { randomUUID } from "crypto";

const sessionId = randomUUID();

const answer1 = await invokeExpertWithSession(
  token.accessToken,
  "data-analyst",
  "帮我分析一下 Q1 的用户增长数据",
  sessionId
);

const answer2 = await invokeExpertWithSession(
  token.accessToken,
  "data-analyst",
  "和 Q4 相比有什么变化？",
  sessionId  // 同一个 session_id，Expert 记得上一轮的内容
);

async function invokeExpertWithSession(
  accessToken: string,
  expertIdentifier: string,
  message: string,
  sessionId: string
): Promise<string> {
  const res = await fetch(`${BASE_URL}/openapi/v1/events/invoke`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${accessToken}`,
    },
    body: JSON.stringify({
      expert_identifier: expertIdentifier,
      message,
      session_id: sessionId,
    }),
  });

  if (!res.ok) {
    throw new Error(`Invoke failed: ${res.status}`);
  }

  let fullText = "";
  const reader = res.body!.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    const chunk = decoder.decode(value, { stream: true });
    for (const line of chunk.split("\n")) {
      if (line.startsWith("data: ")) {
        const data = JSON.parse(line.slice(6));
        if (data.type === "text") {
          fullText += data.content;
        }
      }
    }
  }

  return fullText;
}

import uuid

session_id = str(uuid.uuid4())

answer1 = await invoke_expert_with_session(
    access_token, "data-analyst",
    "帮我分析一下 Q1 的用户增长数据",
    session_id,
)

answer2 = await invoke_expert_with_session(
    access_token, "data-analyst",
    "和 Q4 相比有什么变化？",
    session_id,  # 同一个 session_id，Expert 记得上一轮的内容
)

async def invoke_expert_with_session(
    access_token: str,
    expert_identifier: str,
    message: str,
    session_id: str,
) -> str:
    full_text = ""

    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST",
            f"{BASE_URL}/openapi/v1/events/invoke",
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {access_token}",
            },
            json={
                "expert_identifier": expert_identifier,
                "message": message,
                "session_id": session_id,
            },
        ) as response:
            response.raise_for_status()

            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    import json
                    data = json.loads(line[6:])
                    if data["type"] == "text":
                        full_text += data["content"]

    return full_text

session_id 由你的应用生成和管理。同一个 session_id 下的所有调用共享对话上下文。新的 session_id 会开启全新对话。

Step 4: 处理 SSE 流

Expert 的响应是标准 SSE（Server-Sent Events）流，包含以下事件类型：

事件	说明	数据字段
`text`	文本内容片段	`content: string`
`tool_call_chunk`	工具调用信息	`name: string`, `arguments: string`
`complete`	对话轮次结束	—
`error`	运行时错误	`message: string`, `code: string`

下面是一个通用的 SSE 流解析器：

interface ExpertEvent {
  type: "text" | "tool_call_chunk" | "complete" | "error";
  content?: string;
  name?: string;
  arguments?: string;
  message?: string;
  code?: string;
}

async function* streamExpertEvents(
  response: Response
): AsyncGenerator<ExpertEvent> {
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n");
    buffer = lines.pop() ?? "";

    for (const line of lines) {
      if (line.startsWith("data: ")) {
        yield JSON.parse(line.slice(6)) as ExpertEvent;
      }
    }
  }
}

const response = await fetch(`${BASE_URL}/openapi/v1/events/invoke`, {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Authorization: `Bearer ${accessToken}`,
  },
  body: JSON.stringify({
    expert_identifier: "data-analyst",
    message: "生成季度报告",
  }),
});

for await (const event of streamExpertEvents(response)) {
  switch (event.type) {
    case "text":
      process.stdout.write(event.content ?? "");
      break;
    case "tool_call_chunk":
      console.log(`\n[Tool] ${event.name}(${event.arguments})`);
      break;
    case "complete":
      console.log("\n--- 完成 ---");
      break;
    case "error":
      console.error(`\n[Error] ${event.code}: ${event.message}`);
      break;
  }
}

from dataclasses import dataclass
from typing import AsyncIterator, Optional
import json

@dataclass
class ExpertEvent:
    type: str
    content: Optional[str] = None
    name: Optional[str] = None
    arguments: Optional[str] = None
    message: Optional[str] = None
    code: Optional[str] = None

async def stream_expert_events(
    response: httpx.Response,
) -> AsyncIterator[ExpertEvent]:
    async for line in response.aiter_lines():
        if line.startswith("data: "):
            data = json.loads(line[6:])
            yield ExpertEvent(**data)

async with httpx.AsyncClient() as client:
    async with client.stream(
        "POST",
        f"{BASE_URL}/openapi/v1/events/invoke",
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {access_token}",
        },
        json={
            "expert_identifier": "data-analyst",
            "message": "生成季度报告",
        },
    ) as response:
        response.raise_for_status()

        async for event in stream_expert_events(response):
            match event.type:
                case "text":
                    print(event.content, end="", flush=True)
                case "tool_call_chunk":
                    print(f"\n[Tool] {event.name}({event.arguments})")
                case "complete":
                    print("\n--- 完成 ---")
                case "error":
                    print(f"\n[Error] {event.code}: {event.message}")

Step 5: 错误处理

invoke 端点可能返回以下错误，你的应用需要针对性处理：

HTTP 状态码	含义	处理方式
400	请求参数错误	检查 `expert_identifier` 和 `message` 字段
401	Token 无效或过期	刷新 Access Token 后重试
403	用户无权访问该 Expert	引导用户到 Marketplace 购买/订阅
404	Expert 不存在	检查 identifier 是否正确
402	用户积分不足	提示用户充值
502	Agent Runtime 错误	稍后重试，若持续则联系平台

async function safeInvoke(
  accessToken: string,
  expertIdentifier: string,
  message: string,
  sessionId?: string
): Promise<{ ok: true; text: string } | { ok: false; error: string }> {
  const res = await fetch(`${BASE_URL}/openapi/v1/events/invoke`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${accessToken}`,
    },
    body: JSON.stringify({
      expert_identifier: expertIdentifier,
      message,
      ...(sessionId && { session_id: sessionId }),
    }),
  });

  if (!res.ok) {
    switch (res.status) {
      case 401:
        return { ok: false, error: "TOKEN_EXPIRED" };
      case 402:
        return { ok: false, error: "INSUFFICIENT_BALANCE" };
      case 403:
        return { ok: false, error: "ACCESS_DENIED" };
      case 404:
        return { ok: false, error: "EXPERT_NOT_FOUND" };
      case 502:
        return { ok: false, error: "RUNTIME_ERROR" };
      default:
        return { ok: false, error: `UNKNOWN_${res.status}` };
    }
  }

  let text = "";
  for await (const event of streamExpertEvents(res)) {
    if (event.type === "text") text += event.content ?? "";
    if (event.type === "error") return { ok: false, error: event.message ?? "STREAM_ERROR" };
  }

  return { ok: true, text };
}

async def safe_invoke(
    access_token: str,
    expert_identifier: str,
    message: str,
    session_id: str | None = None,
) -> dict:
    payload = {
        "expert_identifier": expert_identifier,
        "message": message,
    }
    if session_id:
        payload["session_id"] = session_id

    async with httpx.AsyncClient() as client:
        try:
            async with client.stream(
                "POST",
                f"{BASE_URL}/openapi/v1/events/invoke",
                headers={
                    "Content-Type": "application/json",
                    "Authorization": f"Bearer {access_token}",
                },
                json=payload,
            ) as response:
                if response.status_code == 401:
                    return {"ok": False, "error": "TOKEN_EXPIRED"}
                if response.status_code == 402:
                    return {"ok": False, "error": "INSUFFICIENT_BALANCE"}
                if response.status_code == 403:
                    return {"ok": False, "error": "ACCESS_DENIED"}
                if response.status_code == 404:
                    return {"ok": False, "error": "EXPERT_NOT_FOUND"}
                if response.status_code == 502:
                    return {"ok": False, "error": "RUNTIME_ERROR"}

                response.raise_for_status()

                text = ""
                async for event in stream_expert_events(response):
                    if event.type == "text":
                        text += event.content or ""
                    if event.type == "error":
                        return {"ok": False, "error": event.message or "STREAM_ERROR"}

                return {"ok": True, "text": text}

        except httpx.HTTPStatusError:
            return {"ok": False, "error": f"UNKNOWN_{response.status_code}"}

403 通常意味着用户尚未购买该 Expert。对于 METERED 类型的 Expert，用户需要先在 Marketplace 完成订阅；对于 ONE_TIME 类型，用户需要完成一次性购买。

Step 6: 构建聊天界面

将 SSE 流接入前端聊天界面的最小示例：

"use client";

import { useState, useCallback } from "react";

interface Message {
  role: "user" | "assistant";
  content: string;
}

export function ExpertChat({ expertId }: { expertId: string }) {
  const [messages, setMessages] = useState<Message[]>([]);
  const [input, setInput] = useState("");
  const [loading, setLoading] = useState(false);
  const [sessionId] = useState(() => crypto.randomUUID());

  const send = useCallback(async () => {
    if (!input.trim() || loading) return;

    const userMsg: Message = { role: "user", content: input };
    setMessages((prev) => [...prev, userMsg]);
    setInput("");
    setLoading(true);

    const assistantMsg: Message = { role: "assistant", content: "" };
    setMessages((prev) => [...prev, assistantMsg]);

    try {
      const res = await fetch("/api/expert/invoke", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({
          expertId,
          message: input,
          sessionId,
        }),
      });

      const reader = res.body!.getReader();
      const decoder = new TextDecoder();

      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        const chunk = decoder.decode(value, { stream: true });
        for (const line of chunk.split("\n")) {
          if (!line.startsWith("data: ")) continue;
          const data = JSON.parse(line.slice(6));
          if (data.type === "text") {
            setMessages((prev) => {
              const updated = [...prev];
              const last = updated[updated.length - 1];
              updated[updated.length - 1] = {
                ...last,
                content: last.content + (data.content ?? ""),
              };
              return updated;
            });
          }
        }
      }
    } finally {
      setLoading(false);
    }
  }, [input, loading, expertId, sessionId]);

  return (
    <div>
      <div>
        {messages.map((msg, i) => (
          <div key={i} data-role={msg.role}>
            {msg.content}
          </div>
        ))}
      </div>
      <input
        value={input}
        onChange={(e) => setInput(e.target.value)}
        onKeyDown={(e) => e.key === "Enter" && send()}
        placeholder="输入消息..."
        disabled={loading}
      />
    </div>
  );
}

import asyncio
import uuid
import httpx
import json

BASE_URL = "https://app.profy.cn"

async def chat_loop(access_token: str, expert_identifier: str):
    session_id = str(uuid.uuid4())
    print(f"开始与 {expert_identifier} 对话（输入 quit 退出）\n")

    while True:
        user_input = input("你: ")
        if user_input.strip().lower() == "quit":
            break

        print("Expert: ", end="", flush=True)

        async with httpx.AsyncClient() as client:
            async with client.stream(
                "POST",
                f"{BASE_URL}/openapi/v1/events/invoke",
                headers={
                    "Content-Type": "application/json",
                    "Authorization": f"Bearer {access_token}",
                },
                json={
                    "expert_identifier": expert_identifier,
                    "message": user_input,
                    "session_id": session_id,
                },
                timeout=60.0,
            ) as response:
                response.raise_for_status()

                async for line in response.aiter_lines():
                    if not line.startswith("data: "):
                        continue
                    data = json.loads(line[6:])
                    if data["type"] == "text":
                        print(data.get("content", ""), end="", flush=True)
                    elif data["type"] == "error":
                        print(f"\n[错误] {data.get('message', '')}")
                    elif data["type"] == "complete":
                        print()

        print()

asyncio.run(chat_loop("your_access_token", "data-analyst"))

前端不直接调用 Profy API——请求经过你的后端代理（/api/expert/invoke），后端持有 OAuth Token 并转发 SSE 流。这避免了在前端暴露 Access Token。

完整示例

一个可运行的后端服务，将 Expert 调用封装为 API 端点：

import { Hono } from "hono";
import { stream } from "hono/streaming";

const app = new Hono();
const PROFY_BASE = "https://app.profy.cn";

app.post("/api/expert/invoke", async (c) => {
  const { expertId, message, sessionId } = await c.req.json();
  const accessToken = await getAccessTokenForUser(c);

  const upstream = await fetch(`${PROFY_BASE}/openapi/v1/events/invoke`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${accessToken}`,
    },
    body: JSON.stringify({
      expert_identifier: expertId,
      message,
      session_id: sessionId,
    }),
  });

  if (!upstream.ok) {
    return c.json({ error: `upstream_${upstream.status}` }, upstream.status as 400);
  }

  return stream(c, async (s) => {
    const reader = upstream.body!.getReader();
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      await s.write(value);
    }
  });
});

export default app;

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import httpx

app = FastAPI()
PROFY_BASE = "https://app.profy.cn"

@app.post("/api/expert/invoke")
async def invoke_expert(request: Request):
    body = await request.json()
    access_token = await get_access_token_for_user(request)

    async def proxy_stream():
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "POST",
                f"{PROFY_BASE}/openapi/v1/events/invoke",
                headers={
                    "Content-Type": "application/json",
                    "Authorization": f"Bearer {access_token}",
                },
                json={
                    "expert_identifier": body["expertId"],
                    "message": body["message"],
                    "session_id": body.get("sessionId"),
                },
            ) as response:
                response.raise_for_status()
                async for chunk in response.aiter_bytes():
                    yield chunk

    return StreamingResponse(
        proxy_stream(),
        media_type="text/event-stream",
    )

Expert 调用 vs AI 模型调用

Profy Events API 提供两种 AI 调用方式。根据你的场景选择合适的端点：

维度	Expert 调用 (`/events/invoke`)	AI 模型调用 (`/events/chat`)
调用对象	特定 Expert（含 persona + tools + memory）	通用 AI 模型（OpenAI 兼容）
请求格式	`{ expert_identifier, message }`	`{ model, messages }`
上下文管理	平台自动管理（`session_id`）	应用自行维护 `messages` 数组
工具调用	Expert 自带工具，平台自动编排	需要应用自行定义和处理
记忆	Expert 内置跨会话记忆	无记忆，每次调用独立
适用场景	领域专家任务（数据分析、客服、写作）	通用文本生成、翻译、摘要
计费	METERED（按 token）	METERED（按 token）
人设定制	创作者预定义，调用方无需配置	应用自行通过 system prompt 定义

如果你需要的是一个「开箱即用的领域专家」，用 Expert 调用；如果你需要的是「底层模型能力」并自行编排，用 AI 模型调用。两者可以在同一个应用中混合使用。

下一步

SDK 快速开始

安装 SDK、完成 OAuth 对接和首次事件上报

AI 模型调用

使用 OpenAI 兼容接口调用平台 AI 模型

应用接入教程

从零创建 App、配置 OAuth、上架市场

API 参考

Events Invoke 端点完整字段说明

​调用 Expert 智能体

​你将构建什么

​Expert 是什么

​前置条件

​Step 1: 找到 Expert Identifier

​Step 2: 单轮调用

​Step 3: 多轮对话

​Step 4: 处理 SSE 流

​Step 5: 错误处理

​Step 6: 构建聊天界面

​完整示例

​Expert 调用 vs AI 模型调用

​下一步

SDK 快速开始

AI 模型调用

应用接入教程

API 参考

调用 Expert 智能体

你将构建什么

Expert 是什么

前置条件

Step 1: 找到 Expert Identifier

Step 2: 单轮调用

Step 3: 多轮对话

Step 4: 处理 SSE 流

Step 5: 错误处理

Step 6: 构建聊天界面

完整示例

Expert 调用 vs AI 模型调用

下一步