开发
软件开发相关知识
Hermes框架+DeepSeek V4实战：构建企业级智能助手的完整工程方案
Hermes框架+DeepSeek V4实战：构建企业级智能助手的完整工程方案
项目背景
本文将用Hermes框架和DeepSeek V4-Pro，在两天内构建一个真实可用的企业IT支持助手，具备以下能力：
- 查询内部文档：回答技术FAQ、操作手册问题
- 创建工单：自动识别问题并创建JIRA工单
- 执行诊断命令：查询服务器状态、日志分析
一、环境配置
1.1 Hermes框架安装
# Hermes是专为企业LLM应用设计的Python框架
pip install hermes-llm>=2.4.0
pip install deepseek-sdk openai redis pymysql fastapi uvicorn
# 项目结构
mkdir it-assistant && cd it-assistant
hermes init --template enterprise-assistant
项目结构：
it-assistant/
├── main.py # 入口文件
├── config.yaml # 配置文件
├── skills/ # 技能插件目录
│ ├── jira_skill.py
│ ├── server_check.py
│ └── knowledge_base.py
├── knowledge/ # 知识库文档
│ ├── it-faq.md
│ └── server-manual.md
├── prompts/ # 提示词模板
│ └── system_prompt.j2
└── docker-compose.yml
1.2 配置文件
# config.yaml
llm:
provider: deepseek
model: deepseek-v4-pro
api_key: ${DEEPSEEK_API_KEY}
base_url: https://api.deepseek.com/v1
# 利用V4-Pro的100万Token超长上下文
max_context_tokens: 800000
context_strategy: sliding_window # 超出时滑动窗口
temperature: 0.3
max_tokens: 4096
memory:
type: redis
host: localhost
port: 6379
session_ttl: 3600 # 会话保留1小时
max_history: 50 # 最多50轮对话
knowledge_base:
type: vector_store
embedding_model: BAAI/bge-m3
store: chroma
persist_directory: ./chroma_db
security:
# 敏感信息过滤
pii_filter: true
# 输出内容审核
content_filter: true
# 工具调用白名单
allowed_tools:
- query_knowledge_base
- create_jira_ticket
- get_server_status
- get_log_summary
# 禁止的操作
blocked_patterns:
- "rm -rf"
- "DROP TABLE"
- "DELETE FROM"
二、System Prompt设计
{# prompts/system_prompt.j2 #}
你是{{ company_name }}的IT支持智能助手，由DeepSeek V4提供支持。
## 你的角色
- 友好、专业的IT技术支持专员
- 能够快速诊断常见技术问题
- 在需要时创建工单进行跟进
## 可用工具
1. **查询知识库** - 搜索内部技术文档、FAQ和操作手册
2. **创建JIRA工单** - 当问题需要人工处理时
3. **查询服务器状态** - 检查服务器和应用的健康状态
4. **获取日志摘要** - 分析指定服务的最近日志
## 行为准则
1. 先查询知识库，尽量自助解决问题
2. 如果问题已有标准处理流程，引导用户按流程操作
3. 涉及生产环境变更时，必须先创建工单
4. 不确定时，明确告知并升级到人工支持
5. 记录所有对话，便于后续分析
## 当前用户信息
- 姓名：{{ user_name }}
- 部门：{{ user_department }}
- 权限级别：{{ user_permission_level }}
今天是 {{ today }}，请基于最新信息提供帮助。
三、技能插件开发
3.1 JIRA工单技能
# skills/jira_skill.py
from hermes.skills import Skill, SkillParameter
from jira import JIRA
import os
class JiraTicketSkill(Skill):
"""创建JIRA工单技能"""
name = "create_jira_ticket"
description = "当用户的问题需要人工处理或IT团队跟进时，创建JIRA工单"
parameters = [
SkillParameter(
name="summary",
type="string",
description="工单标题，简洁描述问题",
required=True
),
SkillParameter(
name="description",
type="string",
description="详细问题描述，包括复现步骤",
required=True
),
SkillParameter(
name="priority",
type="string",
description="优先级：Critical/High/Medium/Low",
required=False,
default="Medium"
),
SkillParameter(
name="category",
type="string",
description="问题类别：Hardware/Software/Network/Access/Other",
required=True
)
]
def __init__(self):
self.jira = JIRA(
server=os.environ["JIRA_SERVER"],
basic_auth=(
os.environ["JIRA_USER"],
os.environ["JIRA_TOKEN"]
)
)
def execute(
self,
summary: str,
description: str,
priority: str = "Medium",
category: str = "Other",
context: dict = None
) -> dict:
# 从上下文获取用户信息
user_name = context.get("user_name", "Unknown")
user_dept = context.get("user_department", "Unknown")
# 创建工单
issue = self.jira.create_issue(
project="IT",
summary=f"[{category}] {summary}",
description=f"""
*提交人:* {user_name} ({user_dept})
*提交时间:* {datetime.now().strftime('%Y-%m-%d %H:%M')}
*对话记录:* 已附加
---
{description}
""",
issuetype={"name": "IT Support"},
priority={"name": priority},
labels=[category, "ai-generated"]
)
return {
"success": True,
"ticket_id": issue.key,
"url": f"{os.environ['JIRA_SERVER']}/browse/{issue.key}",
"message": f"工单已创建：{issue.key}"
}
3.2 服务器状态查询技能
# skills/server_check.py
import subprocess
import paramiko
from hermes.skills import Skill, SkillParameter
class ServerCheckSkill(Skill):
"""安全的服务器状态查询技能（只读操作）"""
name = "get_server_status"
description = "查询指定服务器或应用的运行状态"
# 白名单：只允许查询这些主机
ALLOWED_HOSTS = {
"prod-web-01": "192.168.1.101",
"prod-db-01": "192.168.1.102",
"staging-web-01": "192.168.1.201"
}
# 只允许这些安全的只读命令
ALLOWED_COMMANDS = {
"cpu_usage": "top -bn1 | grep 'Cpu(s)' | awk '{print $2}'",
"memory_usage": "free -h | awk '/^Mem:/{print $3\"/\"$2}'",
"disk_usage": "df -h | grep -v tmpfs",
"process_count": "ps aux | wc -l",
"uptime": "uptime",
"service_status": "systemctl status {service_name} --no-pager"
}
parameters = [
SkillParameter(
name="host",
type="string",
description="服务器名称（如：prod-web-01）",
required=True
),
SkillParameter(
name="check_type",
type="string",
description="检查类型：cpu_usage/memory_usage/disk_usage/uptime/service_status",
required=True
),
SkillParameter(
name="service_name",
type="string",
description="当check_type为service_status时，指定服务名",
required=False
)
]
def execute(self, host: str, check_type: str, service_name: str = None, **kwargs) -> dict:
# 严格的白名单检查
if host not in self.ALLOWED_HOSTS:
return {
"success": False,
"error": f"服务器 {host} 不在允许查询的列表中"
}
if check_type not in self.ALLOWED_COMMANDS:
return {
"success": False,
"error": f"不支持的检查类型：{check_type}"
}
# 构建命令
cmd = self.ALLOWED_COMMANDS[check_type]
if service_name and "{service_name}" in cmd:
# 过滤服务名中的特殊字符
safe_service_name = re.sub(r'[^a-zA-Z0-9_.-]', '', service_name)
cmd = cmd.format(service_name=safe_service_name)
# 执行SSH命令
try:
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(
self.ALLOWED_HOSTS[host],
username="it-bot",
key_filename="/home/it-bot/.ssh/id_rsa"
)
_, stdout, stderr = ssh.exec_command(cmd, timeout=10)
result = stdout.read().decode().strip()
error = stderr.read().decode().strip()
ssh.close()
return {
"success": True,
"host": host,
"check_type": check_type,
"result": result,
"error": error if error else None
}
except Exception as e:
return {"success": False, "error": str(e)}
四、FastAPI服务入口
# main.py
from fastapi import FastAPI, HTTPException
from hermes import HermesAssistant
import config
app = FastAPI(title="IT Support Assistant")
assistant = HermesAssistant(
config_path="config.yaml",
skills_directory="skills/",
prompt_template="prompts/system_prompt.j2"
)
@app.post("/chat")
async def chat(request: ChatRequest):
"""处理用户消息"""
try:
response = await assistant.chat(
session_id=request.session_id,
message=request.message,
user_context={
"user_name": request.user_name,
"user_department": request.user_department,
"user_permission_level": request.permission_level,
"today": datetime.now().strftime('%Y年%m月%d日')
}
)
return {
"reply": response.text,
"tools_used": response.tools_called,
"session_id": request.session_id
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
五、部署配置
# docker-compose.yml
services:
it-assistant:
build: .
ports: ["8080:8080"]
environment:
DEEPSEEK_API_KEY: ${DEEPSEEK_API_KEY}
JIRA_SERVER: https://your-company.atlassian.net
JIRA_USER: ${JIRA_USER}
JIRA_TOKEN: ${JIRA_TOKEN}
volumes:
- ./knowledge:/app/knowledge
- ./chroma_db:/app/chroma_db
redis:
image: redis:7-alpine
command: redis-server --maxmemory 512mb
nginx:
image: nginx:alpine
ports: ["443:443"]
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
- ./certs:/etc/ssl/certs
通过Hermes框架和DeepSeek V4的组合，企业级AI助手的开发周期从传统的数周压缩到了2天。关键在于：标准化的技能插件系统、配置驱动的对话管理，以及内置的安全防护机制。