调用专家智能体

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

你将构建什么

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

使用 Profy SDK 调用任意已发布专家
多轮对话上下文维持（sessionId）
SSE 流式解析与结构化事件处理
完善的错误处理和重试策略

专家是什么

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

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

通过 SDK 的 agents.run() / agents.runStream() 方法，你的应用可以像调用函数一样调用这些专家。

前置条件

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

Profy API Key：在 Platform Console 创建的 sk-pro- 开头的 API Key
目标专家 Identifier：你要调用的专家的唯一标识

专家调用使用你的应用自己的 API Key（sk-pro-）鉴权，不需要用户的 OAuth Token。API Key 持有你应用的身份，平台按 Key 维度统计用量和计费。

Step 1: 找到专家 Identifier

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

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

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

Step 2: 单轮调用

使用 SDK 的 agents.run() 方法一次性获取完整响应。

import { Profy } from "@profy-ai/sdk";

const client = new Profy({ apiKey: process.env.PROFY_API_KEY! });

const result = await client.agents.run("data-analyst", "分析一下最近的销售趋势");

console.log(result.text);
console.log(`Model: ${result.model}, Tokens: ${result.usage.totalTokens}`);
if (result.toolCalls.length > 0) {
  console.log("Tool calls:", result.toolCalls);
}

from profy import Profy

async with Profy(api_key="sk-pro-...") as client:
    result = await client.agents.run("data-analyst", "分析一下最近的销售趋势")

    print(result.text)
    print(f"Model: {result.model}, Tokens: {result.usage.total_tokens}")
    if result.tool_calls:
        print("Tool calls:", result.tool_calls)

agents.run() 内部消费完整个 SSE 流，返回聚合后的结果：

字段	类型	说明
`text`	`string`	专家完整的文本输出
`model`	`string?`	实际使用的模型
`sessionId`	`string?`	会话 ID（用于多轮对话）
`toolCalls`	`ToolCall[]`	专家调用的工具列表
`usage`	`TokenUsage`	Token 用量统计
`error`	`string?`	运行时错误信息
`elapsedMs` / `elapsed_sec`	`number`	耗时

Step 3: 多轮对话

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

const sessionId = crypto.randomUUID();

const answer1 = await client.agents.run(
  "data-analyst",
  "帮我分析一下 Q1 的用户增长数据",
  { sessionId },
);
console.log(answer1.text);

const answer2 = await client.agents.run(
  "data-analyst",
  "和 Q4 相比有什么变化？",
  { sessionId },  // 同一个 sessionId，专家记得上一轮的内容
);
console.log(answer2.text);

import uuid

session_id = str(uuid.uuid4())

answer1 = await client.agents.run(
    "data-analyst",
    "帮我分析一下 Q1 的用户增长数据",
    session_id=session_id,
)
print(answer1.text)

answer2 = await client.agents.run(
    "data-analyst",
    "和 Q4 相比有什么变化？",
    session_id=session_id,  # 同一个 session_id，专家记得上一轮的内容
)
print(answer2.text)

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

Step 4: 流式输出

使用 agents.runStream() 获取实时 SSE 事件流，适合构建打字机效果的聊天界面。

for await (const chunk of client.agents.runStream("data-analyst", "生成季度报告")) {
  switch (chunk.type) {
    case "output.text.delta":
      process.stdout.write(chunk.content ?? "");
      break;
    case "tool_call.created":
      console.log(`\n[Tool] ${chunk.toolName}`);
      break;
    case "tool_call.completed":
      console.log("[Tool completed]");
      break;
    case "run.completed":
      console.log("\n--- 完成 ---");
      break;
    case "run.failed":
      console.error(`\n[Error] ${chunk.message}`);
      break;
  }
}

async for chunk in await client.agents.run_stream("data-analyst", "生成季度报告"):
    match chunk.type:
        case "output.text.delta":
            print(chunk.content, end="", flush=True)
        case "tool_call.created":
            print(f"\n[Tool] {chunk.tool_name}")
        case "tool_call.completed":
            print("[Tool completed]")
        case "run.completed":
            print("\n--- 完成 ---")
        case "run.failed":
            print(f"\n[Error] {chunk.message}")

SSE 事件类型

事件	说明	关键字段
`run.created`	运行已创建	—
`run.in_progress`	运行进行中	`model_name`, `session_id`
`output.text.delta`	文本内容片段	`content`
`tool_call.created`	工具调用开始	`tool_name`
`tool_call.completed`	工具调用完成	—
`run.completed`	运行结束	—
`run.failed`	运行失败	`message`
`usage`	Token 用量	`prompt_tokens`, `completion_tokens`, `total_tokens`

Step 5: 错误处理

SDK 根据 HTTP 状态码抛出具体异常，你的应用可以针对性处理：

异常类	HTTP 状态码	含义	处理方式
`InvalidRequestError`	400	请求参数错误	检查 expert identifier 和 message
`AuthenticationError`	401	API Key 无效	检查 Key 是否正确、未过期
`InsufficientBalanceError`	402	余额不足	提示充值，不重试
`PermissionDeniedError`	403	无权访问该专家	引导购买/订阅
`RateLimitError`	429	请求频率过高	指数退避重试

import {
  Profy,
  AuthenticationError,
  InsufficientBalanceError,
  RateLimitError,
} from "@profy-ai/sdk";

try {
  const result = await client.agents.run("data-analyst", "分析数据");
  console.log(result.text);
} catch (error) {
  if (error instanceof AuthenticationError) {
    console.error("API Key 无效，请检查配置");
  } else if (error instanceof InsufficientBalanceError) {
    console.error("余额不足，请充值");
  } else if (error instanceof RateLimitError) {
    console.error("请求过于频繁，稍后重试");
  } else {
    throw error;
  }
}

from profy import (
    Profy,
    AuthenticationError,
    InsufficientBalanceError,
    RateLimitError,
)

try:
    result = await client.agents.run("data-analyst", "分析数据")
    print(result.text)
except AuthenticationError:
    print("API Key 无效，请检查配置")
except InsufficientBalanceError:
    print("余额不足，请充值")
except RateLimitError:
    print("请求过于频繁，稍后重试")

收到 InsufficientBalanceError（402）时不要重试——余额不足不会因为重试而改变。向用户展示充值入口。

Step 6: 构建聊天界面

将流式输出接入前端聊天界面的示例。前端通过你的后端代理调用，后端持有 API Key：

"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 === "output.text.delta") {
            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
from profy import Profy

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

    async with Profy(api_key=api_key) as client:
        while True:
            user_input = input("你: ")
            if user_input.strip().lower() == "quit":
                break

            print("专家: ", end="", flush=True)
            async for chunk in await client.agents.run_stream(
                expert_identifier, user_input, session_id=session_id,
            ):
                if chunk.type == "output.text.delta":
                    print(chunk.content or "", end="", flush=True)
                elif chunk.type == "run.failed":
                    print(f"\n[错误] {chunk.message}")
                elif chunk.type == "run.completed":
                    print()
            print()

asyncio.run(chat_loop("sk-pro-...", "data-analyst"))

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

完整后端示例

将专家调用封装为 API 端点，前端直接消费 SSE 流：

import { Hono } from "hono";
import { stream } from "hono/streaming";
import { Profy } from "@profy-ai/sdk";

const app = new Hono();
const client = new Profy({ apiKey: process.env.PROFY_API_KEY! });

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

  return stream(c, async (s) => {
    for await (const chunk of client.agents.runStream(expertId, message, { sessionId })) {
      s.write(`data: ${JSON.stringify(chunk)}\n\n`);
    }
  }, "text/event-stream");
});

export default app;

import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from profy import Profy

app = FastAPI()
client = Profy(api_key="sk-pro-...")

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

    async def proxy_stream():
        async for chunk in await client.agents.run_stream(
            body["expertId"], body["message"],
            session_id=body.get("sessionId"),
        ):
            yield f"data: {json.dumps({'type': chunk.type, 'content': chunk.content})}\n\n"

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

专家调用 vs AI 模型调用

Profy SDK 提供两种 AI 调用方式。根据场景选择：

维度	专家调用 (`agents.run`)	AI 模型调用 (`chat.completions.create`)
调用对象	特定专家（含 persona + tools + memory）	通用 AI 模型
API 路径	`/v1/agents/run`	`/v1/chat/completions`
上下文管理	平台自动管理（`sessionId`）	应用自行维护 `messages` 数组
工具调用	专家自带工具，平台自动编排	需要应用自行定义和处理
记忆	专家内置跨会话记忆	无记忆，每次调用独立
适用场景	领域专家任务（数据分析、客服、写作）	通用文本生成、翻译、摘要

const response = await client.chat.completions.create({
  model: "deepseek-chat",
  messages: [
    { role: "system", content: "你是一个翻译助手。" },
    { role: "user", content: "Translate to English: 今天天气不错" },
  ],
  temperature: 0.3,
});
console.log(response.text);

response = await client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是一个翻译助手。"},
        {"role": "user", "content": "Translate to English: 今天天气不错"},
    ],
    temperature=0.3,
)
print(response.text)

如果你需要的是一个「开箱即用的领域专家」，用 agents.run；如果你需要的是「底层模型能力」并自行编排，用 chat.completions.create。两者使用同一个 Profy 客户端。

下一步

SDK 快速开始

安装 SDK、创建 API Key 和首次调用

按 Token 计费的 AI 工具

使用 chat.completions 构建按量计费的 AI 应用

按次计费实战

使用 ProfyApp report_event 实现按次扣费

Token 管理

OAuth Token 持久化与并发刷新最佳实践

开始使用

认证与密钥

API 能力指南

SDK

集成

Cookbook

参考

调用专家智能体

你将构建什么

专家是什么

前置条件

Step 1: 找到专家 Identifier

Step 2: 单轮调用

Step 3: 多轮对话

Step 4: 流式输出

SSE 事件类型

Step 5: 错误处理

Step 6: 构建聊天界面

完整后端示例

专家调用 vs AI 模型调用

下一步

SDK 快速开始

按 Token 计费的 AI 工具

按次计费实战

Token 管理

​你将构建什么

​专家是什么

​前置条件

​Step 1: 找到专家 Identifier

​Step 2: 单轮调用

​Step 3: 多轮对话

​Step 4: 流式输出

​SSE 事件类型

​Step 5: 错误处理

​Step 6: 构建聊天界面

​完整后端示例

​专家调用 vs AI 模型调用

​下一步

SDK 快速开始

按 Token 计费的 AI 工具

按次计费实战

Token 管理

你将构建什么

专家是什么

前置条件

Step 1: 找到专家 Identifier

Step 2: 单轮调用

Step 3: 多轮对话

Step 4: 流式输出

SSE 事件类型

Step 5: 错误处理

Step 6: 构建聊天界面

完整后端示例

专家调用 vs AI 模型调用

下一步