GuangYu/Agents

Fork 0

Files

nelson 85604454a0 Focus L2 agent and update V7.1 knowledge base

2026-06-13 15:43:24 +08:00

16 KiB

Raw Permalink Blame History

慧遇 Agent API 接口文档

基于 Google ADK api_server 提供的 REST API，供外部项目集成使用。

API 文档（Swagger UI）：启动服务后访问 http://<host>:<port>/docs

快速开始

# 1. 启动服务
cd agents && adk api_server . --auto_create_session --port 8000

# 2. 发送对话
curl -X POST http://localhost:8000/run \
  -H "Content-Type: application/json" \
  -d '{
    "app_name": "root_agent",
    "user_id": "user_001",
    "session_id": "my_session_001",
    "new_message": {
      "role": "user",
      "parts": [{"text": "你好"}]
    }
  }'

启动服务

adk api_server <agents_dir> [选项]

常用参数

参数	说明	默认值
`--host`	监听地址	`127.0.0.1`
`--port`	监听端口	`8000`
`--auto_create_session`	调用 `/run` 时自动创建不存在的会话	关闭
`--session_service_uri`	会话存储后端	本地 SQLite
`--allow_origins`	CORS 允许的源	`*`
`--url_prefix`	URL 路径前缀（如 `/api/v1`）	无

会话存储配置

# 内存存储（重启丢失）
adk api_server . --session_service_uri memory://

# SQLite 持久化
adk api_server . --session_service_uri sqlite:///path/to/sessions.db

# PostgreSQL
adk api_server . --session_service_uri postgresql://user:pass@host:5432/db

核心概念

三维会话标识

每次 API 调用需要通过三个维度定位会话：

维度	说明	示例
`app_name`	Agent 应用名称（目录名）	`root_agent`
`user_id`	用户唯一标识（由调用方管理）	`user_001`
`session_id`	会话唯一标识	`sess_abc123`

消息格式（Content）

所有对话消息使用统一的 Content 格式：

{
  "role": "user",
  "parts": [
    {"text": "消息内容"}
  ]
}

role 可选值："user" | "model"

系统接口

健康检查

GET /health

响应：

{"status": "ok"}

版本信息

GET /version

响应：

{
  "version": "1.28.1",
  "language": "python",
  "language_version": "3.10.12"
}

列出可用应用

GET /list-apps?detailed=true

查询参数：

参数	类型	说明
`detailed`	bool	是否返回详细信息，默认 `false`

响应（detailed=false）：

["root_agent", "l2_training_camp_agent"]

响应（detailed=true）：

{
  "apps": [
    {
      "name": "root_agent",
      "root_agent_name": "huiyu_agent",
      "description": "一个智能助手，能够回答用户的各种问题。",
      "language": "python",
      "is_computer_use": false
    }
  ]
}

会话管理

所有会话接口的基础路径：

/apps/{app_name}/users/{user_id}/sessions

创建会话

POST /apps/{app_name}/users/{user_id}/sessions

请求体（均可选）：

{
  "session_id": "my_custom_session_id",
  "state": {"key": "value"}
}

字段	类型	必填	说明
`session_id`	string	否	自定义会话 ID，不传则自动生成 UUID
`state`	object	否	会话初始状态
`events`	array	否	初始化事件列表

响应：

{
  "id": "c470eb5f-6a66-4005-8427-7d3d9d178171",
  "appName": "root_agent",
  "userId": "user_001",
  "state": {},
  "events": [],
  "lastUpdateTime": 1775556907.896
}

获取会话详情

GET /apps/{app_name}/users/{user_id}/sessions/{session_id}

响应： 同创建会话的响应格式。

错误： 404 会话不存在

列出用户的所有会话

GET /apps/{app_name}/users/{user_id}/sessions

响应：

[
  {
    "id": "c470eb5f-6a66-4005-8427-7d3d9d178171",
    "appName": "root_agent",
    "userId": "user_001",
    "state": {},
    "events": [],
    "lastUpdateTime": 1775556907.896
  }
]

删除会话

DELETE /apps/{app_name}/users/{user_id}/sessions/{session_id}

响应： 204 No Content

更新会话状态

PATCH /apps/{app_name}/users/{user_id}/sessions/{session_id}

请求体：

{
  "state_delta": {
    "key_to_update": "new_value"
  }
}

字段	类型	必填	说明
`state_delta`	object	是	要合并到会话状态的变更

响应： 更新后的 Session 对象。

Agent 对话

运行 Agent（同步）

POST /run

请求体：

{
  "app_name": "root_agent",
  "user_id": "user_001",
  "session_id": "c470eb5f-6a66-4005-8427-7d3d9d178171",
  "new_message": {
    "role": "user",
    "parts": [{"text": "你好，请介绍一下你自己"}]
  }
}

字段	类型	必填	说明
`app_name`	string	是	应用名称
`user_id`	string	是	用户 ID
`session_id`	string	是	会话 ID
`new_message`	Content	否*	新消息（与 `invocation_id` 至少提供一个）
`streaming`	bool	否	是否流式，默认 `false`
`state_delta`	object	否	状态变更
`invocation_id`	string	否	用于恢复被中断的函数调用
`function_call_event_id`	string	否	用于恢复长时间运行的函数

*如果未启用 --auto_create_session，session_id 对应的会话必须已存在，否则返回 404。

响应： Event[] 事件数组

[
  {
    "id": "852208bf-c883-4c44-99ef-68303e0ef4be",
    "author": "huiyu_agent",
    "content": {
      "role": "model",
      "parts": [
        {"text": "思考过程...", "thought": true},
        {"text": "你好！我是小慧，一个乐于助人的智能助手。"}
      ]
    },
    "partial": false,
    "finishReason": "STOP",
    "usageMetadata": {
      "promptTokenCount": 113,
      "candidatesTokenCount": 44,
      "totalTokenCount": 157
    },
    "timestamp": 1775556943.259,
    "invocationId": "e-90d3bd10-59e4-4403-a5d4-2a822dff0bd0"
  }
]

Event 关键字段说明：

字段	说明
`content.parts[].thought`	为 `true` 时表示该 part 是思考过程，非最终回复
`content.parts[].text`	文本内容
`finishReason`	结束原因：`STOP`（正常完成）、`SAFETY`（安全拦截）等
`usageMetadata`	Token 用量统计

提取最终回复的逻辑：

for event in response:
    parts = event.get("content", {}).get("parts", [])
    for part in parts:
        if part.get("text") and not part.get("thought"):
            final_reply = part["text"]

SSE 流式对话

运行 Agent（SSE 流式）

POST /run_sse

请求体： 与 /run 完全相同。

响应： text/event-stream

每个 SSE 事件格式：

data: {"id": "...", "content": {"role": "model", "parts": [{"text": "你"}]}, "partial": true}

data: {"id": "...", "content": {"role": "model", "parts": [{"text": "好"}]}, "partial": true}

data: {"id": "...", "content": {"role": "model", "parts": [{"text": "！"}]}, "partial": false, "finishReason": "STOP"}

错误事件格式：

data: {"error": "Session not found"}

客户端示例（Python）：

import requests

response = requests.post(
    "http://localhost:8000/run_sse",
    json={
        "app_name": "root_agent",
        "user_id": "user_001",
        "session_id": "my_session",
        "new_message": {"role": "user", "parts": [{"text": "你好"}]},
        "streaming": True,
    },
    stream=True,
)

for line in response.iter_lines():
    if line.startswith(b"data: "):
        data = json.loads(line[6:])
        if "error" in data:
            print(f"错误: {data['error']}")
            break
        parts = data.get("content", {}).get("parts", [])
        for part in parts:
            if part.get("text") and not part.get("thought"):
                print(part["text"], end="", flush=True)

客户端示例（JavaScript）：

const response = await fetch("http://localhost:8000/run_sse", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    app_name: "root_agent",
    user_id: "user_001",
    session_id: "my_session",
    new_message: { role: "user", parts: [{ text: "你好" }] },
    streaming: true,
  }),
});

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

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  const text = decoder.decode(value);
  for (const line of text.split("\n")) {
    if (line.startsWith("data: ")) {
      const data = JSON.parse(line.slice(6));
      if (data.error) { console.error(data.error); break; }
      const parts = data.content?.parts || [];
      for (const part of parts) {
        if (part.text && !part.thought) {
          process.stdout.write(part.text);
        }
      }
    }
  }
}

WebSocket 实时对话

连接地址

ws://<host>:<port>/run_live?app_name=root_agent&user_id=user_001&session_id=my_session

查询参数：

参数	类型	必填	说明
`app_name`	string	是	应用名称
`user_id`	string	是	用户 ID
`session_id`	string	是	会话 ID
`modalities`	string[]	否	模态类型，默认 `["AUDIO"]`，可选 `TEXT`、`AUDIO`
`proactive_audio`	bool	否	是否启用主动音频
`enable_affective_dialog`	bool	否	是否启用情感对话
`enable_session_resumption`	bool	否	是否启用会话恢复

通信协议： 双向发送 JSON 文本消息，每条消息为 Event 对象的 JSON 序列化。

WebSocket 关闭码：

关闭码	说明
`1002`	会话不存在
`1008`	Origin 不被允许
`1011`	服务器内部错误

错误处理

所有接口在出错时返回统一的错误格式：

{"detail": "错误描述信息"}

HTTP 状态码	说明
`400`	请求参数错误
`404`	会话 / 应用不存在
`422`	请求体验证失败（Pydantic 校验错误）
`500`	服务器内部错误

代码示例

Python 完整集成示例

import requests

BASE_URL = "http://localhost:8000"
APP_NAME = "root_agent"


class HuiYuAgentClient:
    """慧遇 Agent API 客户端"""

    def __init__(self, base_url: str = BASE_URL, app_name: str = APP_NAME):
        self.base_url = base_url
        self.app_name = app_name

    def health_check(self) -> bool:
        """健康检查"""
        resp = requests.get(f"{self.base_url}/health")
        return resp.json().get("status") == "ok"

    def create_session(self, user_id: str, session_id: str = None) -> dict:
        """创建会话"""
        body = {}
        if session_id:
            body["session_id"] = session_id
        resp = requests.post(
            f"{self.base_url}/apps/{self.app_name}/users/{user_id}/sessions",
            json=body,
        )
        resp.raise_for_status()
        return resp.json()

    def list_sessions(self, user_id: str) -> list:
        """列出用户的所有会话"""
        resp = requests.get(
            f"{self.base_url}/apps/{self.app_name}/users/{user_id}/sessions",
        )
        resp.raise_for_status()
        return resp.json()

    def delete_session(self, user_id: str, session_id: str):
        """删除会话"""
        resp = requests.delete(
            f"{self.base_url}/apps/{self.app_name}/users/{user_id}/sessions/{session_id}",
        )
        resp.raise_for_status()

    def chat(self, user_id: str, session_id: str, message: str) -> str:
        """
        发送消息并获取回复（同步）

        Args:
            user_id: 用户 ID
            session_id: 会话 ID
            message: 用户消息文本

        Returns:
            Agent 的最终回复文本
        """
        resp = requests.post(
            f"{self.base_url}/run",
            json={
                "app_name": self.app_name,
                "user_id": user_id,
                "session_id": session_id,
                "new_message": {
                    "role": "user",
                    "parts": [{"text": message}],
                },
            },
        )
        resp.raise_for_status()
        events = resp.json()

        # 提取最终回复（跳过 thought 部分）
        for event in events:
            parts = event.get("content", {}).get("parts", [])
            for part in parts:
                if part.get("text") and not part.get("thought"):
                    return part["text"]

        return ""

    def chat_stream(self, user_id: str, session_id: str, message: str):
        """
        发送消息并流式获取回复（SSE）

        Yields:
            每次 yield 一个文本片段
        """
        resp = requests.post(
            f"{self.base_url}/run_sse",
            json={
                "app_name": self.app_name,
                "user_id": user_id,
                "session_id": session_id,
                "new_message": {
                    "role": "user",
                    "parts": [{"text": message}],
                },
                "streaming": True,
            },
            stream=True,
        )
        resp.raise_for_status()

        for line in resp.iter_lines():
            if line and line.startswith(b"data: "):
                import json
                data = json.loads(line[6:])
                if "error" in data:
                    raise Exception(data["error"])
                parts = data.get("content", {}).get("parts", [])
                for part in parts:
                    if part.get("text") and not part.get("thought"):
                        yield part["text"]


# ===== 使用示例 =====
if __name__ == "__main__":
    client = HuiYuAgentClient()

    # 1. 健康检查
    assert client.health_check(), "服务未就绪"
    print("✅ 服务正常")

    # 2. 创建会话
    session = client.create_session("user_001")
    session_id = session["id"]
    print(f"✅ 会话已创建: {session_id}")

    # 3. 同步对话
    reply = client.chat("user_001", session_id, "你好，我叫小明")
    print(f"👤 用户: 你好，我叫小明")
    print(f"🤖 Agent: {reply}")

    # 4. 上下文记忆测试
    reply = client.chat("user_001", session_id, "我叫什么名字？")
    print(f"👤 用户: 我叫什么名字？")
    print(f"🤖 Agent: {reply}")

    # 5. 流式对话
    print("👤 用户: 请用一句话介绍你自己")
    print("🤖 Agent: ", end="", flush=True)
    for chunk in client.chat_stream("user_001", session_id, "请用一句话介绍你自己"):
        print(chunk, end="", flush=True)
    print()

cURL 示例

# 创建会话
curl -s -X POST http://localhost:8000/apps/root_agent/users/user_001/sessions \
  -H "Content-Type: application/json" \
  -d '{}'

# 发送对话（使用返回的 session_id）
curl -s -X POST http://localhost:8000/run \
  -H "Content-Type: application/json" \
  -d '{
    "app_name": "root_agent",
    "user_id": "user_001",
    "session_id": "<替换为实际session_id>",
    "new_message": {"role": "user", "parts": [{"text": "你好"}]}
  }'

# 查看会话列表
curl -s http://localhost:8000/apps/root_agent/users/user_001/sessions

# 删除会话
curl -s -X DELETE http://localhost:8000/apps/root_agent/users/user_001/sessions/<session_id>

16 KiB Raw Permalink Blame History Unescape Escape

慧遇 Agent API 接口文档

目录

快速开始

启动服务

常用参数

会话存储配置

核心概念

三维会话标识

消息格式（Content）

系统接口

健康检查

版本信息

列出可用应用

会话管理

创建会话

获取会话详情

列出用户的所有会话

删除会话

更新会话状态

Agent 对话

运行 Agent（同步）

SSE 流式对话

运行 Agent（SSE 流式）

WebSocket 实时对话

连接地址

错误处理

代码示例

Python 完整集成示例

cURL 示例

16 KiB

Raw Permalink Blame History