开发

Hermes框架+DeepSeek V4实战：构建企业级智能助手的完整工程方案

北盛科技 阅读 663

项目背景

本文将用Hermes框架和DeepSeek V4-Pro，在两天内构建一个真实可用的企业IT支持助手，具备以下能力：

  1. 查询内部文档：回答技术FAQ、操作手册问题
  2. 创建工单：自动识别问题并创建JIRA工单
  3. 执行诊断命令：查询服务器状态、日志分析

一、环境配置

1.1 Hermes框架安装

# Hermes是专为企业LLM应用设计的Python框架pip install hermes-llm>=2.4.0pip install deepseek-sdk openai redis pymysql fastapi uvicorn# 项目结构mkdir it-assistant && cd it-assistanthermes 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.yamlllm: 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: 4096memory: 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_dbsecurity: # 敏感信息过滤 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.pyfrom hermes.skills import Skill, SkillParameterfrom jira import JIRAimport osclass 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.pyimport subprocessimport paramikofrom hermes.skills import Skill, SkillParameterclass 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.pyfrom fastapi import FastAPI, HTTPExceptionfrom hermes import HermesAssistantimport configapp = 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.ymlservices: 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天。关键在于：标准化的技能插件系统、配置驱动的对话管理，以及内置的安全防护机制。