上一篇 下一篇 分享链接 返回 返回顶部

从零接入 AI Agent API:接口调用、流式响应与源码实战指南

发布人:慈云数据-客服中心 发布时间:3小时前 阅读量:2

AI Agent API接口调用教程|附源码

随着大模型技术的快速发展,越来越多企业和开发者开始将 AI Agent(智能体) 集成到自己的系统中。相比传统的单轮问答模型,AI Agent 不仅可以理解用户问题,还能够根据任务目标进行规划、调用工具、访问外部接口、处理上下文,并最终完成相对复杂的自动化任务。

例如,一个 AI Agent 可以完成以下工作:

  • 自动分析用户需求;
  • 查询数据库或业务系统;
  • 调用天气、地图、搜索、支付等外部 API;
  • 生成结构化结果;
  • 根据执行结果继续下一步操作;
  • 在多轮对话中保持上下文记忆。

本文将以实战方式介绍 AI Agent API 接口调用方法,包括接口设计思路、请求参数说明、Python 调用示例、Node.js 调用示例、流式响应处理、工具调用思路以及完整源码示例。适合正在学习 AI Agent 开发、智能客服、自动化助手、企业内部知识库问答系统的开发者参考。


一、什么是 AI Agent API?

AI Agent API 可以理解为一个对外提供智能体能力的接口。开发者通过 HTTP 请求向 Agent 服务发送用户输入、上下文信息、工具参数等内容,Agent 服务收到请求后,会调用大语言模型进行推理,并根据配置决定是否调用外部工具,最终返回处理结果。

一个典型的 AI Agent API 调用流程如下:

用户输入
   ↓
业务系统调用 Agent API
   ↓
Agent 分析任务意图
   ↓
必要时调用外部工具/API
   ↓
整合结果
   ↓
返回最终答案

例如用户提问:

帮我查询今天北京天气,并给出是否适合户外跑步的建议。

普通大模型可能只能根据已有知识生成泛化回答,而 AI Agent 可以先调用天气 API 获取实时天气数据,再结合温度、空气质量、降雨概率等信息生成更准确的建议。


二、AI Agent API 常见应用场景

AI Agent API 并不局限于聊天机器人,它可以广泛应用于各种自动化业务场景。

1. 智能客服

将 Agent 接入客服系统后,可以自动回答用户常见问题,例如订单状态、物流信息、退款进度、产品说明等。如果涉及业务数据,Agent 可以调用订单系统、CRM 系统或工单系统进行查询。

2. 企业知识库问答

企业内部通常存在大量文档,如制度文件、技术文档、产品手册、合同模板等。通过 Agent API 可以构建知识库问答系统,员工只需要自然语言提问,系统即可返回相关答案。

3. 自动化办公助手

AI Agent 可以帮助用户完成日报生成、会议纪要整理、邮件撰写、日程安排、数据分析等任务,提升办公效率。

4. 数据分析助手

用户可以通过自然语言询问数据情况,例如:

请分析本月销售额相比上月下降的主要原因。

Agent 可以调用数据库查询数据,进行统计分析,并生成可读性较强的分析报告。

5. 代码开发助手

AI Agent 可以结合代码仓库、接口文档、错误日志等信息,帮助开发者定位问题、生成代码、解释异常、重构模块等。


三、接口调用前的准备工作

在调用 AI Agent API 之前,通常需要准备以下内容:

1. API Key

大多数 AI 服务都会通过 API Key 进行身份认证。你需要在服务平台创建应用并获取密钥。

示例:

API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx

为了安全起见,不建议将 API Key 直接写死在代码中,应优先使用环境变量或配置中心。

2. Agent ID

如果平台支持创建多个智能体,通常每个 Agent 都会有一个唯一标识,例如:

AGENT_ID=agent_123456

不同 Agent 可以拥有不同的人设、工具、知识库和工作流配置。

3. API 地址

假设我们的 Agent API 地址如下:

https://api.example.com/v1/agents/chat

实际项目中请替换为你所使用平台的真实地址。

4. 请求格式

通常 AI Agent API 会使用 JSON 作为请求体,Content-Type 为:

Content-Type: application/json

同时在请求头中携带认证信息:

Authorization: Bearer YOUR_API_KEY

四、AI Agent API 请求参数说明

下面是一个典型的 Agent API 请求示例:

{
  "agent_id": "agent_123456",
  "session_id": "user_001_session",
  "user_id": "user_001",
  "message": "帮我写一份关于AI Agent接口调用的教程",
  "stream": false,
  "temperature": 0.7,
  "max_tokens": 2000,
  "metadata": {
    "source": "web",
    "lang": "zh-CN"
  }
}

参数说明

参数名 类型 是否必填 说明
agent_id string 智能体 ID,用于指定调用哪个 Agent
session_id string 会话 ID,用于保持多轮对话上下文
user_id string 用户 ID,便于区分不同用户
message string 用户输入内容
stream boolean 是否开启流式输出
temperature number 控制生成内容随机性,值越高越发散
max_tokens number 最大输出长度
metadata object 附加参数,可用于业务追踪

其中比较重要的是 session_id。如果你希望 Agent 记住上下文,就需要在同一个用户的多轮对话中传入相同的 session_id


五、返回结果格式说明

非流式响应通常类似下面这样:

{
  "id": "resp_987654",
  "agent_id": "agent_123456",
  "session_id": "user_001_session",
  "status": "completed",
  "answer": "下面是一份AI Agent API接口调用教程……",
  "usage": {
    "prompt_tokens": 560,
    "completion_tokens": 1200,
    "total_tokens": 1760
  },
  "created_at": 1719999999
}

返回字段说明

字段名 类型 说明
id string 本次响应 ID
agent_id string 智能体 ID
session_id string 会话 ID
status string 执行状态,如 completed、failed
answer string Agent 生成的最终回答
usage object Token 消耗统计
created_at number 响应创建时间戳

如果 Agent 在执行过程中调用了工具,有些平台还会返回 tool_calls 字段,用于展示工具调用过程。

示例:

{
  "tool_calls": [
    {
      "name": "search_order",
      "arguments": {
        "order_id": "A10086"
      },
      "result": {
        "status": "已发货",
        "express_no": "SF123456789"
      }
    }
  ]
}

六、Python 调用 AI Agent API 示例

下面我们使用 Python 的 requests 库调用 Agent API。

1. 安装依赖

pip install requests python-dotenv

2. 创建 .env 文件

AI_AGENT_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx
AI_AGENT_API_URL=https://api.example.com/v1/agents/chat
AI_AGENT_ID=agent_123456

3. Python 源码示例

import os
import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("AI_AGENT_API_KEY")
API_URL = os.getenv("AI_AGENT_API_URL")
AGENT_ID = os.getenv("AI_AGENT_ID")


def call_ai_agent(message, user_id="user_001", session_id="session_001"):
    """
    调用 AI Agent API
    :param message: 用户输入内容
    :param user_id: 用户ID
    :param session_id: 会话ID
    :return: Agent 回复内容
    """

    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "agent_id": AGENT_ID,
        "user_id": user_id,
        "session_id": session_id,
        "message": message,
        "stream": False,
        "temperature": 0.7,
        "max_tokens": 2000
    }

    try:
        response = requests.post(
            API_URL,
            headers=headers,
            json=payload,
            timeout=60
        )

        response.raise_for_status()

        data = response.json()

        return data.get("answer", "")

    except requests.exceptions.Timeout:
        return "请求超时,请稍后重试。"

    except requests.exceptions.HTTPError as e:
        return f"HTTP错误:{e}"

    except requests.exceptions.RequestException as e:
        return f"请求异常:{e}"

    except Exception as e:
        return f"未知错误:{e}"


if __name__ == "__main__":
    question = input("请输入你的问题:")
    answer = call_ai_agent(question)
    print("\nAI Agent 回复:")
    print(answer)

4. 运行代码

python main.py

输入:

请帮我生成一份产品周报模板。

输出示例:

AI Agent 回复:
以下是一份产品周报模板……

七、Python 流式响应调用示例

在实际产品中,如果 Agent 生成内容较长,用户等待完整结果会比较久。此时可以使用 流式响应,让内容边生成边展示,类似 ChatGPT 的打字机效果。

流式响应请求参数

只需要将 stream 设置为 true

{
  "stream": true
}

Python 流式调用源码

import os
import json
import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("AI_AGENT_API_KEY")
API_URL = os.getenv("AI_AGENT_API_URL")
AGENT_ID = os.getenv("AI_AGENT_ID")


def stream_ai_agent(message, user_id="user_001", session_id="session_stream_001"):
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "agent_id": AGENT_ID,
        "user_id": user_id,
        "session_id": session_id,
        "message": message,
        "stream": True,
        "temperature": 0.7,
        "max_tokens": 2000
    }

    with requests.post(
        API_URL,
        headers=headers,
        json=payload,
        stream=True,
        timeout=120
    ) as response:
        response.raise_for_status()

        for line in response.iter_lines(decode_unicode=True):
            if not line:
                continue

            # 常见 SSE 格式为:data: {...}
            if line.startswith("data:"):
                line = line.replace("data:", "").strip()

            if line == "[DONE]":
                break

            try:
                chunk = json.loads(line)
                delta = chunk.get("delta", "")
                if delta:
                    print(delta, end="", flush=True)
            except json.JSONDecodeError:
                print(line, end="", flush=True)


if __name__ == "__main__":
    stream_ai_agent("请用中文介绍AI Agent API的调用方式。")

流式返回通常是 Server-Sent Events,格式可能如下:

data: {"delta":"AI"}
data: {"delta":" Agent"}
data: {"delta":" API"}
data: {"delta":" 可以通过"}
data: {"delta":" HTTP 请求调用"}
data: [DONE]

八、Node.js 调用 AI Agent API 示例

如果你的后端服务使用 Node.js,也可以通过 fetchaxios 调用 Agent API。

1. 初始化项目

mkdir ai-agent-demo
cd ai-agent-demo
npm init -y
npm install axios dotenv

2. 创建 .env 文件

AI_AGENT_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx
AI_AGENT_API_URL=https://api.example.com/v1/agents/chat
AI_AGENT_ID=agent_123456

3. Node.js 源码示例

创建 index.js

require("dotenv").config();
const axios = require("axios");

const API_KEY = process.env.AI_AGENT_API_KEY;
const API_URL = process.env.AI_AGENT_API_URL;
const AGENT_ID = process.env.AI_AGENT_ID;

async function callAiAgent(message, userId = "user_001", sessionId = "session_001") {
  try {
    const response = await axios.post(
      API_URL,
      {
        agent_id: AGENT_ID,
        user_id: userId,
        session_id: sessionId,
        message,
        stream: false,
        temperature: 0.7,
        max_tokens: 2000
      },
      {
        headers: {
          Authorization: `Bearer ${API_KEY}`,
          "Content-Type": "application/json"
        },
        timeout: 60000
      }
    );

    return response.data.answer;
  } catch (error) {
    if (error.response) {
      console.error("HTTP错误:", error.response.status, error.response.data);
    } else if (error.request) {
      console.error("请求发送失败:", error.message);
    } else {
      console.error("未知错误:", error.message);
    }

    return "AI Agent 调用失败,请稍后重试。";
  }
}

async function main() {
  const answer = await callAiAgent("请帮我写一份AI Agent接口调用教程大纲。");
  console.log("AI Agent 回复:");
  console.log(answer);
}

main();

4. 运行代码

node index.js

九、封装成后端接口供前端调用

在真实项目中,一般不建议前端直接调用 AI Agent API,因为这样会暴露 API Key。更安全的做法是由后端封装一层接口,前端请求自己的后端服务,再由后端调用 Agent API。

下面以 Node.js + Express 为例。

1. 安装依赖

npm install express cors axios dotenv

2. 后端服务源码

创建 server.js

require("dotenv").config();

const express = require("express");
const cors = require("cors");
const axios = require("axios");

const app = express();

app.use(cors());
app.use(express.json());

const API_KEY = process.env.AI_AGENT_API_KEY;
const API_URL = process.env.AI_AGENT_API_URL;
const AGENT_ID = process.env.AI_AGENT_ID;

app.post("/api/chat", async (req, res) => {
  const { message, userId, sessionId } = req.body;

  if (!message) {
    return res.status(400).json({
      error: "message不能为空"
    });
  }

  try {
    const response = await axios.post(
      API_URL,
      {
        agent_id: AGENT_ID,
        user_id: userId || "anonymous",
        session_id: sessionId || `session_${Date.now()}`,
        message,
        stream: false,
        temperature: 0.7,
        max_tokens: 2000
      },
      {
        headers: {
          Authorization: `Bearer ${API_KEY}`,
          "Content-Type": "application/json"
        }
      }
    );

    res.json({
      answer: response.data.answer,
      usage: response.data.usage || null
    });
  } catch (error) {
    console.error("Agent API调用失败:", error.message);

    res.status(500).json({
      error: "AI Agent调用失败"
    });
  }
});

app.listen(3000, () => {
  console.log("Server running at http://localhost:3000");
});

3. 前端调用示例




  
  AI Agent Chat Demo


  

AI Agent 对话示例


回复内容:



  


十、AI Agent 工具调用设计思路

AI Agent 的核心能力之一是调用工具。工具可以是任意外部函数或接口,例如:

  • 查询订单接口;
  • 查询天气接口;
  • 搜索接口;
  • 数据库查询接口;
  • 发送邮件接口;
  • 创建工单接口;
  • 调用企业内部系统接口。

工具定义示例

假设我们希望 Agent 能够查询订单状态,可以定义一个工具:

{
  "name": "query_order",
  "description": "根据订单号查询订单状态和物流信息",
  "parameters": {
    "type": "object",
    "properties": {
      "order_id": {
        "type": "string",
        "description": "订单号"
      }
    },
    "required": ["order_id"]
  }
}

当用户输入:

帮我查一下订单 A10086 现在到哪里了?

Agent 可以识别出需要调用 query_order 工具,并提取参数:

{
  "order_id": "A10086"
}

后端执行工具调用后返回结果:

{
  "order_id": "A10086",
  "status": "运输中",
  "express_company": "顺丰速运",
  "express_no": "SF123456789",
  "latest_trace": "快件已到达北京朝阳中转站"
}

最终 Agent 再将工具返回结果转成自然语言:

你的订单 A10086 当前状态为运输中,承运商是顺丰速运,运单号为 SF123456789。最新物流信息显示:快件已到达北京朝阳中转站,预计明天送达。

十一、完整 Python 版 Agent 客户端封装源码

为了方便在项目中复用,我们可以将 API 调用封装成一个类。

import os
import json
import requests


class AiAgentClient:
    def __init__(self, api_key, api_url, agent_id):
        self.api_key = api_key
        self.api_url = api_url
        self.agent_id = agent_id

    def chat(
        self,
        message,
        user_id="anonymous",
        session_id="default_session",
        temperature=0.7,
        max_tokens=2000
    ):
        headers = self._build_headers()

        payload = {
            "agent_id": self.agent_id,
            "user_id": user_id,
            "session_id": session_id,
            "message": message,
            "stream": False,
            "temperature": temperature,
            "max_tokens": max_tokens
        }

        response = requests.post(
            self.api_url,
            headers=headers,
            json=payload,
            timeout=60
        )

        response.raise_for_status()
        data = response.json()

        return data

    def stream_chat(
        self,
        message,
        user_id="anonymous",
        session_id="default_session",
        temperature=0.7,
        max_tokens=2000
    ):
        headers = self._build_headers()

        payload = {
            "agent_id": self.agent_id,
            "user_id": user_id,
            "session_id": session_id,
            "message": message,
            "stream": True,
            "temperature": temperature,
            "max_tokens": max_tokens
        }

        with requests.post(
            self.api_url,
            headers=headers,
            json=payload,
            stream=True,
            timeout=120
        ) as response:
            response.raise_for_status()

            for line in response.iter_lines(decode_unicode=True):
                if not line:
                    continue

                if line.startswith("data:"):
                    line = line.replace("data:", "").strip()

                if line == "[DONE]":
                    break

                try:
                    data = json.loads(line)
                    yield data.get("delta", "")
                except json.JSONDecodeError:
                    yield line

    def _build_headers(self):
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }


if __name__ == "__main__":
    client = AiAgentClient(
        api_key=os.getenv("AI_AGENT_API_KEY"),
        api_url=os.getenv("AI_AGENT_API_URL"),
        agent_id=os.getenv("AI_AGENT_ID")
    )

    result = client.chat(
        message="请生成一份AI Agent API接口调用教程。",
        user_id="user_001",
        session_id="session_001"
    )

    print(result.get("answer"))

    print("\n--- 流式输出示例 ---\n")

    for text in client.stream_chat(
        message="请简要说明AI Agent和普通聊天机器人的区别。",
        user_id="user_001",
        session_id="session_002"
    ):
        print(text, end="", flush=True)

这个封装类包含了两个核心方法:

  • chat():普通非流式调用;
  • stream_chat():流式调用。

在实际项目中,你还可以继续扩展日志记录、异常处理、重试机制、限流控制等功能。


十二、接口调用中的常见问题

1. API Key 暴露怎么办?

不要把 API Key 写在前端代码中,也不要提交到 Git 仓库。推荐做法是:

  • 使用环境变量;
  • 使用配置中心;
  • 后端代理调用;
  • 定期轮换密钥;
  • 设置密钥权限和调用额度。

2. 请求超时怎么办?

AI Agent 在执行复杂任务时可能需要调用多个工具,因此耗时会比普通问答更长。可以采用以下方案:

  • 增大后端请求超时时间;
  • 使用流式响应;
  • 对长任务采用异步任务队列;
  • 前端增加加载状态;
  • 对工具调用设置单独超时。

3. 多轮对话如何实现?

关键是传入固定的 session_id。同一个用户在同一次会话中使用相同 session_id,Agent 就可以基于上下文继续回答。

示例:

{
  "user_id": "user_001",
  "session_id": "chat_20240701_001",
  "message": "请帮我写一份项目计划"
}

下一轮继续使用同一个 session_id

{
  "user_id": "user_001",
  "session_id": "chat_20240701_001",
  "message": "请把时间周期缩短到两周"
}

4. 如何降低调用成本?

可以从以下几个方面优化:

  • 控制 max_tokens
  • 对重复问题做缓存;
  • 精简上下文;
  • 对知识库检索结果进行截断;
  • 根据任务复杂度选择不同模型;
  • 使用摘要压缩历史对话。

5. 如何提高回答稳定性?

如果你希望 Agent 输出更稳定,可以降低 temperature,例如设置为 0.20.3。如果希望内容更有创意,可以提高到 0.8 左右。


十三、生产环境最佳实践

在生产环境中调用 AI Agent API,不仅要能跑通,还需要关注稳定性、安全性和可维护性。

1. 增加日志追踪

建议记录以下信息:

  • 请求时间;
  • 用户 ID;
  • session_id;
  • 请求参数摘要;
  • 响应状态;
  • Token 使用量;
  • 错误信息;
  • 工具调用耗时。

但注意不要记录敏感信息,例如用户密码、身份证号、银行卡号等。

2. 增加重试机制

对于网络抖动、服务临时不可用等问题,可以增加有限次数的重试。例如最多重试 2 次,每次间隔 1 秒。

3. 做好限流控制

为了避免接口被恶意刷爆,可以对用户维度、IP 维度或应用维度进行限流,例如每分钟最多请求 30 次。

4. 敏感内容过滤

如果应用面向公众用户,需要对输入和输出进行安全过滤,避免生成违规内容或泄露隐私数据。

5. 设置降级方案

当 Agent 服务不可用时,可以返回固定提示,或者切换到 FAQ、人工客服、缓存结果等降级方案。


十四、总结

AI Agent API 是将智能体能力集成到业务系统中的关键入口。通过 API 调用,开发者可以快速构建智能客服、知识库问答、自动化办公助手、数据分析助手等应用。

本文从基础概念出发,介绍了 AI Agent API 的请求参数、返回格式、Python 调用方式、Node.js 调用方式、流式响应处理、后端接口封装、工具调用设计以及生产环境最佳实践,并提供了可直接参考的源码示例。

实际开发时,建议遵循以下原则:

  1. API Key 不暴露到前端
  2. 通过 session_id 管理多轮上下文
  3. 长文本生成优先使用流式响应
  4. 工具调用要设置超时和异常处理
  5. 生产环境必须加入日志、限流、重试和降级机制
  6. 根据业务场景合理控制成本和响应速度

掌握 AI Agent API 的调用方式后,你就可以将大模型智能体能力嵌入到自己的产品、网站、App 或企业系统中,实现更高效、更智能的自动化交互体验。

目录结构
全文