# 知行合翼agent

**Repository Path**: mingline-smallme/zxhy_agent

## Basic Information

- **Project Name**: 知行合翼agent
- **Description**: 知行合翼探索agent的代码仓库
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 3
- **Forks**: 1
- **Created**: 2026-01-13
- **Last Updated**: 2026-04-29

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 🚀 AgentScope 智能体系统 - 知行合翼 API 接口服务

> **目标**：构建一个基于 AgentScope 的智能体框架，支持多场景告警诊断任务，提供兼容 OpenAI 协议的 API 接口供“知行合翼”前端调用。

---

## 📌 1. 项目简介

本项目是一个面向企业级运维场景的智能体（Agent）系统，集成在业务平台「知行合翼」中。其核心功能是接收用户输入的告警信息或交互式追问，通过 **ReAct 框架 + 工具注册机制 + 权限控制 + 外部确认流** 实现自动化诊断推理。

系统支持三种操作模式：
1. 创建工具（由管理员完成）
2. 创建思维链语料（COT）（由专家编写）
3. 用户发起查询（触发智能体执行）

所有逻辑均封装为一个 **OpenAI 兼容 API**，供前端统一调用。

---

## 🧱 2. 整体架构设计

![告警智能体流程图](D:\Projects\ShangHaiAlertAgent\告警智能体流程图.png)

---

## 🗂️ 3. 目录结构设计

```bash
project-root/
├── app/                      # 主应用模块
│   ├── __init__.py
│   ├── router.py             # 请求路由逻辑
│   ├── agent_factory.py      # 智能体实例工厂
│   ├── tools/                # 工具注册与封装
│   │   ├── api_tool.py       # API 类型工具封装
│   │   ├── agent_tool.py     # Agent as Tool 封装
│   │   ├── xiaoda_workflow.py # 晓搭工作流封装
│   │   └── default_tools.py  # 默认工具集（如验证码）
│   ├── utils/
│   │   ├── permission_checker.py  # 权限校验器
│   │   ├── confirmation_manager.py # 外部确认管理器
│   │   └── file_loader.py        # JSON/TXT 文件加载器
│   ├── agents/               # ReActAgent 实例定义
│   │   └── react_agent.py
│   └── api_server.py         # OpenAI 兼容 API 服务入口
├── config/
│   ├── settings.py           # 配置文件（路径、密钥等）
│   └── constants.py          # 常量定义
├── data/
│   ├── tools/                # 存放工具元数据（JSON）
│   │   └── tool_*.json
│   └── cot/                  # 存放思维链语料（TXT）
│       └── cot_*.txt
├── tests/
│   ├── unit/
│   └── integration/
├── requirements.txt
├── README.md
└── run.sh                    # 启动脚本
```

---

## 🔧 4. 核心模块详解

### 4.1 Router（请求分发器）

负责判断用户请求类型，并决定如何处理：

```python
def route_request(user_id, session_id, query):
    if is_alert_query(query):
        return create_new_agent_instance(user_id, session_id, query)
    else:
        return get_last_agent_instance(user_id, session_id)
```

- `is_alert_query()`：通过关键词匹配或分类模型判断是否为告警信息。
- 支持三大业务场景：作业计划、主动巡检、故障诊治。

---

### 4.2 Agent Factory（智能体实例工厂）

根据用户输入创建对应的智能体实例。

#### 功能：
1. 加载对应 COT 文本作为 `sys_prompt`
2. 注册该 COT 中提到的所有工具
3. 初始化 `ReActAgent`

```python
class AgentFactory:
    def __init__(self, tool_dir, cot_dir):
        self.tool_dir = tool_dir
        self.cot_dir = cot_dir

    def create_agent(self, scenario, user_id):
        prompt = self.load_cot(scenario)
        toolkit = self.build_toolkit(scenario, user_id)
        agent = ReActAgent(
            name=f"{scenario}助手",
            sys_prompt=prompt,
            model=DashScopeChatModel(...),
            formatter=DashScopeChatFormatter(),
            toolkit=toolkit,
            memory=InMemoryMemory(),
        )
        return agent
```

---

### 4.3 工具注册与封装（Toolkit）

支持多种来源的工具：

| 来源 | 实现方式 |
|------|----------|
| API | `requests` 调用，封装成函数 |
| Agent as Tool | 使用 AgentScope 的 `AgentTool` |
| 晓搭工作流 | 通过 HTTP 或 SDK 调用，包装为函数 |
| 默认工具 | 如 `send_verification_code`, `verify_code` |

#### 示例：API 工具封装

```python
def query_pods_by_api(api_id: str, start_time: str) -> dict:
    url = f"https://api.example.com/pods/{api_id}"
    headers = {"Authorization": "Bearer XXX"}
    resp = requests.get(url, params={"start_time": start_time}, headers=headers)
    return resp.json()
```

#### 工具元数据格式（`tools/tool_query_pods.json`）

```json
{
  "name": "query_pods_by_api",
  "description": "查询指定API的Pod状态",
  "params": {
    "api_id": "string",
    "start_time": "string"
  },
  "permissions": {
    "roles": ["admin", "operator"],
    "orgs": ["ops", "dev"]
  },
  "requires_confirmation": false
}
```

---

### 4.4 权限与确认控制（中间件）

在工具执行前进行拦截处理。

#### 权限检查器

```python
class PermissionChecker:
    def __init__(self, tool_config_path):
        self.tool_configs = load_json_files(tool_config_path)

    def has_permission(self, user_role, user_org, tool_name):
        config = self.tool_configs.get(tool_name)
        if not config:
            return False
        roles = config.get("permissions", {}).get("roles", [])
        orgs = config.get("permissions", {}).get("orgs", [])
        return (user_role in roles) and (user_org in orgs)
```

#### 外部确认管理器

```python
class ConfirmationManager:
    def __init__(self, code_service):
        self.code_service = code_service
        self.attempts = {}

    def send_code(self, user_id, phone):
        code = generate_random_code()
        self.code_service.send_sms(phone, code)
        self.attempts[user_id] = {"code": code, "tries": 0}

    def verify_code(self, user_id, input_code):
        if user_id not in self.attempts:
            return False
        data = self.attempts[user_id]
        if data["tries"] >= 3:
            return False
        if input_code == data["code"]:
            del self.attempts[user_id]
            return True
        data["tries"] += 1
        return False
```

---

### 4.5 ReActAgent 扩展行为

在 `ReActAgent` 的 `action` 阶段插入自定义逻辑：

```python
class CustomReActAgent(ReActAgent):
    def _execute_action(self, action: Action):
        tool_name = action.name
        if not self.permission_checker.has_permission(self.user_role, self.user_org, tool_name):
            return {"error": "您没有使用此工具的权限。"}

        if self.confirmation_manager.requires_confirmation(tool_name):
            # 发送验证码并等待用户输入
            result = self._handle_confirmation(tool_name)
            if result["success"]:
                return super()._execute_action(action)
            else:
                return {"error": result["message"]}
        else:
            return super()._execute_action(action)
```

> ⚠️ 注意：由于 AgentScope 的 `ReActAgent` 不直接暴露 `action` 执行前钩子，建议通过 **覆盖 `_execute_action` 方法** 或 **继承类** 实现扩展。

---

## 💬 5. OpenAI 兼容 API 接口设计

### 请求示例

```http
POST /v1/chat/completions
Content-Type: application/json

{
  "model": "qwen3-32b",
  "messages": [
    {
      "role": "user",
      "content": "API成功率下降了"
    }
  ],
  "user_id": "u_123",
  "session_id": "s_456"
}
```

### 响应示例

```json
{
  "id": "chatcmpl-abc123",
  "object": "chat.completion",
  "created": 1700000000,
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "正在分析... 已提取到API_ID: api-xyz，时间范围: 2024-01-01T00:00:00Z..."
      }
    }
  ]
}
```

> ✅ 保持与 OpenAI ChatCompletion 接口完全兼容，便于前端对接。

---

## 🛠️ 6. 配置与部署

### 环境变量配置

```env
# config/settings.py
TOOL_DIR=./data/tools
COT_DIR=./data/cot
API_KEY=sk-your-dashscope-key
MODEL_NAME=qwen3-32b
PORT=8000
```

### 启动命令

```bash
uvicorn app.api_server:app --host 0.0.0.0 --port 8005
```

### Docker部署

```

docker build -t shanghai-alert-agent:1.0 .

docker save -o shanghai-alert-agent-1.0.tar shanghai-alert-agent:1.0

镜像文件：
shanghai-alert-agent-1.0.tar

创建容器：
docker run -itd -p 8005:8005 --name alert-agent shanghai-alert-agent:1.0 /bin/bash

进入容器：
docker exec -it <contain_id> /bin/bash

启动服务：
uvicorn app.api_server:app --host 0.0.0.0 --port 8005

测试指令(宿主机):
curl -X POST http://localhost:8005/v1/chat/completions -H "Content-Type: application/json" -d "{\"model\":\"qwen3-32b\",\"messages\":[{\"role\":\"user\",\"content\":\"API成功率告警:告警时间2025-12-25 08:00:05 API_ID: 12325\"}],\"user_id\":\"u_ope\",\"session_id\":\"s_456\"}
```
---

## 🔍 7. 扩展性设计

| 特性 | 实现方式 |
|------|---------|
| 新增工具类型 | 在 `tools/` 下新增模块，注册即可 |
| 新增业务场景 | 添加新的 COT 文件，修改 Router 判断规则 |
| 支持更多认证方式 | 替换 `confirmation_manager` 中的 `code_service` |
| 多租户支持 | 在权限检查中加入租户 ID 字段 |

---

## 📝 8. 待办事项（To-Do List）

| 任务 | 状态 | 备注 |
|------|------|------|
| 完成 Router 实现 | ✅ | 已初步设计 |
| 实现 AgentFactory | ✅ | 可复用 |
| 工具封装（API/Agent/晓搭） | 🟡 | 分步实现 |
| 权限校验模块 | 🟡 | 需对接用户中心 |
| 外部确认流程 | 🟡 | 包含验证码逻辑 |
| 自定义 ReActAgent 行为 | 🟡 | 关键点 |
| OpenAI 兼容 API 服务 | ✅ | 基于 FastAPI |
| 单元测试 | ❌ | 待补充 |
| 部署方案 | ❌ | 待 Docker 化 |

---

## 📚 9. 参考资料

- [AgentScope 官方文档](https://agentscope.readthedocs.io/)
- [DashScope API 文档](https://help.aliyun.com/document_detail/179327.html)
- [OpenAI API 规范](https://platform.openai.com/docs/api-reference/chat/create)

---