从零接入 AI Agent API:接口调用、流式响应与源码实战指南
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,也可以通过 fetch 或 axios 调用 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.2 或 0.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 调用方式、流式响应处理、后端接口封装、工具调用设计以及生产环境最佳实践,并提供了可直接参考的源码示例。
实际开发时,建议遵循以下原则:
- API Key 不暴露到前端;
- 通过 session_id 管理多轮上下文;
- 长文本生成优先使用流式响应;
- 工具调用要设置超时和异常处理;
- 生产环境必须加入日志、限流、重试和降级机制;
- 根据业务场景合理控制成本和响应速度。
掌握 AI Agent API 的调用方式后,你就可以将大模型智能体能力嵌入到自己的产品、网站、App 或企业系统中,实现更高效、更智能的自动化交互体验。