GuangYu/Agents
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

develop #5

Merged
nelson merged 6 commits from develop into main 2026-04-08 19:18:26 +08:00
Conversation 0 Commits 6 Files Changed 14 +982 -66
6 changed files with 982 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit c9f5724119 - Show all commits

681
API_DOC.md Normal file
View File

@@ -0,0 +1,681 @@
# 慧遇 Agent API 接口文档
> 基于 Google ADK `api_server` 提供的 REST API，供外部项目集成使用。
>
> **API 文档（Swagger UI）**：启动服务后访问 `http://<host>:<port>/docs`
---
## 目录
- [快速开始](#快速开始)
- [启动服务](#启动服务)
- [核心概念](#核心概念)
- [系统接口](#系统接口)
- [会话管理](#会话管理)
- [Agent 对话](#agent-对话)
- [SSE 流式对话](#sse-流式对话)
- [WebSocket 实时对话](#websocket-实时对话)
- [错误处理](#错误处理)
- [代码示例](#代码示例)
---
## 快速开始
```bash
# 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": "你好"}]
}
}'
```
---
## 启动服务
```bash
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`） | 无 |
### 会话存储配置
```bash
# 内存存储（重启丢失）
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` 格式：
```json
{
"role": "user",
"parts": [
{"text": "消息内容"}
]
}
```
`role` 可选值：`"user"` | `"model"`
---
## 系统接口
### 健康检查
```
GET /health
```
**响应：**
```json
{"status": "ok"}
```
### 版本信息
```
GET /version
```
**响应：**
```json
{
"version": "1.28.1",
"language": "python",
"language_version": "3.10.12"
}
```
### 列出可用应用
```
GET /list-apps?detailed=true
```
**查询参数：**
| 参数 | 类型 | 说明 |
|------|------|------|
| `detailed` | bool | 是否返回详细信息，默认 `false` |
**响应（detailed=false）：**
```json
["root_agent", "note_agent"]
```
**响应（detailed=true）：**
```json
{
"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
```
**请求体（均可选）：**
```json
{
"session_id": "my_custom_session_id",
"state": {"key": "value"}
}
```
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `session_id` | string | 否 | 自定义会话 ID，不传则自动生成 UUID |
| `state` | object | 否 | 会话初始状态 |
| `events` | array | 否 | 初始化事件列表 |
**响应：**
```json
{
"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
```
**响应：**
```json
[
{
"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}
```
**请求体：**
```json
{
"state_delta": {
"key_to_update": "new_value"
}
}
```
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `state_delta` | object | 是 | 要合并到会话状态的变更 |
**响应：** 更新后的 Session 对象。
---
## Agent 对话
### 运行 Agent（同步）
```
POST /run
```
**请求体：**
```json
{
"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[]` 事件数组
```json
[
{
"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 用量统计 |
**提取最终回复的逻辑：**
```python
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）：**
```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）：**
```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` | 服务器内部错误 |
---
## 错误处理
所有接口在出错时返回统一的错误格式：
```json
{"detail": "错误描述信息"}
```
| HTTP 状态码 | 说明 |
|-------------|------|
| `400` | 请求参数错误 |
| `404` | 会话 / 应用不存在 |
| `422` | 请求体验证失败（Pydantic 校验错误） |
| `500` | 服务器内部错误 |
---
## 代码示例
### Python 完整集成示例
```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 示例
```bash
# 创建会话
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>
```

5
agents/__init__.py
View File

@@ -3,3 +3,8 @@ Agent 模块
每个 Agent 是一个独立的子目录，包含 agent.py 和 __init__.py。 每个 Agent 是一个独立的子目录，包含 agent.py 和 __init__.py。
""" """
# 导出所有 Agent，供 adk web 服务注册
from .root_agent import agent as root_agent
__all__ = ["root_agent"]

8
agents/note_agent/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
from . import agent
__all__ = ["agent"]

77
agents/note_agent/agent.py Normal file
View File

@@ -0,0 +1,77 @@
import os
from base.agent import HuiYuBaseAgent
from google.adk.models.lite_llm import LiteLlm
from .note_formatter import NoteFormatter
# ============================================================
# MiniMax 模型配置（从 .env 文件读取）
# ============================================================
MINIMAX_API_KEY = os.getenv("MINIMAX_API_KEY")
MINIMAX_API_BASE = os.getenv("MINIMAX_API_BASE")
MINIMAX_MODEL = os.getenv("MINIMAX_MODEL")
# ============================================================
# 创建 LiteLlm 模型适配器
# ============================================================
model = LiteLlm(
model=MINIMAX_MODEL,
api_base=MINIMAX_API_BASE,
api_key=MINIMAX_API_KEY,
)
# ============================================================
# 笔记格式化器
# ============================================================
_formatter = NoteFormatter()
# ============================================================
# Tool: 生成文本笔记
# ============================================================
def generate_text_note(title: str, content: str) -> str:
"""将文本内容整理为 Markdown 笔记。
Args:
title: 笔记标题
content: 笔记正文内容
"""
return _formatter.format_to_markdown(title=title, text_content=content)
# ============================================================
# Tool: 生成图片笔记
# ============================================================
def generate_image_note(title: str, description: str) -> str:
"""将图片描述整理为 Markdown 笔记。
Args:
title: 笔记标题
description: 图片描述或相关文字说明
"""
return _formatter.format_to_markdown(title=title, text_content=description)
# ============================================================
# 定义 Agent（继承 HuiYuBaseAgent）
# ============================================================
note_agent = HuiYuBaseAgent(
name="note_agent",
model=model,
description="一个多模态笔记助手，能够将图片、文字等内容转换为结构化的 Markdown 笔记。",
instruction=(
"你是一个专业的笔记助手，帮助用户将各种内容整理成 Markdown 笔记。\n"
"你的能力包括：\n"
"1. 接收用户输入的文字内容，整理成结构化的笔记\n"
"2. 接收用户上传的图片，生成图片描述并整理为笔记\n"
"3. 将混合内容（文字+图片）整合为一份完整的笔记\n\n"
"工作流程：\n"
"- 当用户发送文字时，整理内容并调用 generate_text_note 生成笔记\n"
"- 当用户发送图片时，描述图片内容并调用 generate_image_note 生成笔记\n"
"- 当用户发送混合内容时，整合所有内容后生成笔记\n"
"- 笔记标题由你根据内容自动生成，要简洁准确\n"
"- 生成笔记后，直接将 Markdown 内容输出给用户\n"
"- 回复时使用中文"
),
tools=[generate_text_note, generate_image_note],
)

208
agents/note_agent/note_formatter.py Normal file
View File

@@ -0,0 +1,208 @@
"""
笔记格式化模块
将多模态内容（图片、文本等）转换为 Markdown 笔记格式。
支持可扩展的处理器架构，方便后续添加音频、视频、PDF 等格式。
"""
import base64
import hashlib
import logging
from abc import ABC, abstractmethod
from datetime import datetime
from pathlib import Path
from typing import Optional
logger = logging.getLogger("adk.agent")
# ============================================================
# 内容处理器基类
# ============================================================
class BaseContentHandler(ABC):
"""内容处理器基类，所有格式处理器继承此类。"""
@abstractmethod
def can_handle(self, mime_type: str) -> bool:
"""判断是否能处理该 MIME 类型"""
...
@abstractmethod
def process(self, data: bytes, mime_type: str, filename: str = "") -> str:
"""
处理内容并返回 Markdown 片段。
Args:
data: 原始字节数据
mime_type: MIME 类型
filename: 文件名（可选）
Returns:
Markdown 格式的字符串片段
"""
...
# ============================================================
# 图片处理器
# ============================================================
class ImageHandler(BaseContentHandler):
"""处理图片类型，保存到本地并生成 Markdown 图片引用。"""
SUPPORTED_TYPES = {"image/png", "image/jpeg", "image/jpg", "image/gif", "image/webp", "image/svg+xml"}
def can_handle(self, mime_type: str) -> bool:
return mime_type.lower() in self.SUPPORTED_TYPES
def process(self, data: bytes, mime_type: str, filename: str = "") -> str:
# 确保输出目录存在
NOTES_OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
# 生成文件名
if not filename:
ext = self._mime_to_ext(mime_type)
filename = f"image_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{hashlib.md5(data).hexdigest()[:8]}.{ext}"
filepath = NOTES_OUTPUT_DIR / filename
filepath.write_bytes(data)
logger.info("图片已保存: %s", filepath)
return f"![{filename}](notes/{filename})"
@staticmethod
def _mime_to_ext(mime_type: str) -> str:
return {
"image/png": "png",
"image/jpeg": "jpg",
"image/jpg": "jpg",
"image/gif": "gif",
"image/webp": "webp",
"image/svg+xml": "svg",
}.get(mime_type.lower(), "bin")
# ============================================================
# 文本处理器
# ============================================================
class TextHandler(BaseContentHandler):
"""处理纯文本内容。"""
SUPPORTED_TYPES = {"text/plain", "text/markdown", "text/csv"}
def can_handle(self, mime_type: str) -> bool:
return mime_type.lower() in self.SUPPORTED_TYPES
def process(self, data: bytes, mime_type: str, filename: str = "") -> str:
text = data.decode("utf-8", errors="replace")
if mime_type.lower() == "text/markdown":
return text # 已经是 Markdown，直接返回
return text # 纯文本直接返回
# ============================================================
# 占位处理器（用于尚未支持的格式）
# ============================================================
class PlaceholderHandler(BaseContentHandler):
"""占位处理器，为尚未支持的格式生成占位标记。"""
def can_handle(self, mime_type: str) -> bool:
return True # 兜底，处理所有未匹配的类型
def process(self, data: bytes, mime_type: str, filename: str = "") -> str:
name = filename or "unsupported_file"
return f"<!-- [{name}] 格式 {mime_type} 暂不支持，待后续扩展 -->"
# ============================================================
# 笔记格式化器
# ============================================================
class NoteFormatter:
"""
笔记格式化器，管理所有内容处理器。
用法：
formatter = NoteFormatter()
md_content = formatter.format_to_markdown(
title="会议笔记",
text_content="讨论了项目进度",
files=[(image_bytes, "image/png", "screenshot.png")],
)
"""
def __init__(self):
self._handlers: list[BaseContentHandler] = [
ImageHandler(),
TextHandler(),
PlaceholderHandler(), # 必须放最后，作为兜底
]
def add_handler(self, handler: BaseContentHandler):
"""注册新的内容处理器（插入到占位处理器之前）"""
# 移除末尾的 PlaceholderHandler
if self._handlers and isinstance(self._handlers[-1], PlaceholderHandler):
self._handlers.pop()
self._handlers.append(handler)
# 重新添加 PlaceholderHandler 到末尾
self._handlers.append(PlaceholderHandler())
def _find_handler(self, mime_type: str) -> BaseContentHandler:
for handler in self._handlers:
if handler.can_handle(mime_type):
return handler
return self._handlers[-1] # PlaceholderHandler
def format_to_markdown(
self,
title: str = "",
text_content: str = "",
files: Optional[list[tuple[bytes, str, str]]] = None,
) -> str:
"""
将多模态内容格式化为 Markdown 笔记。
Args:
title: 笔记标题
text_content: 文本内容
files: 文件列表，每项为 (data, mime_type, filename)
Returns:
完整的 Markdown 笔记字符串
"""
lines: list[str] = []
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 标题
if title:
lines.append(f"# {title}")
lines.append("")
else:
lines.append(f"# 笔记 - {now}")
lines.append("")
# 元信息
lines.append(f"> 创建时间：{now}")
lines.append("")
# 文本内容
if text_content:
lines.append("## 内容")
lines.append("")
lines.append(text_content)
lines.append("")
# 附件
if files:
lines.append("## 附件")
lines.append("")
for data, mime_type, filename in files:
handler = self._find_handler(mime_type)
try:
md_fragment = handler.process(data, mime_type, filename)
lines.append(md_fragment)
lines.append("")
except Exception as e:
logger.error("处理文件失败: %s, 错误: %s", filename, e)
lines.append(f"<!-- 文件 {filename} 处理失败: {e} -->")
lines.append("")
return "\n".join(lines)

69
agents/root_agent/agent.py
View File

@@ -1,11 +1,7 @@
import os import os
import asyncio
from base.agent import HuiYuBaseAgent from base.agent import HuiYuBaseAgent
from google.adk.models.lite_llm import LiteLlm from google.adk.models.lite_llm import LiteLlm
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types
# ============================================================ # ============================================================
# MiniMax 模型配置（从 .env 文件读取） # MiniMax 模型配置（从 .env 文件读取）
@@ -27,67 +23,8 @@ model = LiteLlm(
# 定义 Agent（继承 HuiYuBaseAgent，自动获得防注入 + 日志能力） # 定义 Agent（继承 HuiYuBaseAgent，自动获得防注入 + 日志能力）
# ============================================================ # ============================================================
root_agent = HuiYuBaseAgent( root_agent = HuiYuBaseAgent(
name="minimax_agent", name="huiyu_agent",
model=model, model=model,
description="一个使用 MiniMax 模型的智能助手，能够回答用户的各种问题。", description="一个智能助手，能够回答用户的各种问题。",
instruction="你是一个乐于助人的中文 AI 助手，由 MiniMax 模型驱动。请用中文回答用户的问题，回答要准确、简洁、友好。", instruction="你是一个乐于助人的中文 AI 助手，你的名字叫小慧，请用中文回答用户的问题，回答要准确、简洁、友好。",
) )
# ============================================================
# 运行入口（用于 adk run / adk web）
# ============================================================
APP_NAME = "慧遇agent app"
USER_ID = "user_001"
SESSION_ID = "session_001"
async def main():
"""主函数：创建会话并运行 Agent 对话"""
# 1. 创建会话服务
session_service = InMemorySessionService()
# 2. 创建会话
session = await session_service.create_session(
app_name=APP_NAME,
user_id=USER_ID,
session_id=SESSION_ID,
)
print(f"会话已创建: App='{APP_NAME}', User='{USER_ID}', Session='{SESSION_ID}'")
# 3. 创建 Runner
runner = Runner(
agent=root_agent,
app_name=APP_NAME,
session_service=session_service,
)
print(f"Runner 已创建，Agent: '{runner.agent.name}'")
# 4. 交互式对话循环
print("\n===== MiniMax Agent 对话 (输入 'exit' 退出) =====")
while True:
query = input("\n[user]: ").strip()
if query.lower() in ("exit", "quit", "退出"):
print("再见！")
break
if not query:
continue
# 构造 ADK 消息
content = types.Content(role="user", parts=[types.Part(text=query)])
# 调用 Agent
final_response_text = ""
async for event in runner.run_async(
user_id=USER_ID,
session_id=SESSION_ID,
new_message=content,
):
if event.is_final_response():
if event.content and event.content.parts:
final_response_text = event.content.parts[0].text
print(f"\n[agent]: {final_response_text}")
if __name__ == "__main__":
asyncio.run(main())