hermes agent的安装和使用
hermes agent完全可以不从公网访问它，但是它必须能访问公网。
hermes agent的Docker方式安装：
首次安装（镜像5GB多挺大，使用的Debian 13 linux）：
# setenforce 0
# dnf upgrade crun
# mkdir /root/hermes_agent/global_agent
# docker run -it –rm -v /root/hermes_agent/global_agent:/opt/data nousresearch/hermes-agent:v2026.5.7 setup
表示在docker里面执行：/opt/hermes/.venv/bin/hermes setup命令。
配置和飞书的chat id（群聊才能用）: global_agent
再次以网关方式启动：
# docker run -d –name hermes_global_agent –restart unless-stopped -v /root/hermes_agent/global_agent:/opt/data -p 8642:8642 -e TZ=Asia/Shanghai nousresearch/hermes-agent:v2026.5.7 gateway run
飞书机器人配置：必须以websocket长连接方式，事件配置: 添加im.message.receive_v1事件, 回调配置: 添加card.action.trigger回调。在飞书群组里不能用”@机器人”方式使用”/cron”等斜杠类的命令。
根据用户指示临时生成的python文件、结果文件都在容器里面的/tmp/目录下，stop容器再start还会保留，不会清除掉。
升级：
# docker stop hermes_global_agent
# docker rm hermes_global_agent
# docker pull nousresearch/hermes-agent:v2026.5.7
然后执行上面的网关方式启动即可(不用再setup初始化)。
把/opt/hermes/目录映射到host，方便修改代码：
注意：必须是先从容器里拷贝出来，再重建容器时映射进去，如果首次建容器直接映射的话会导致容器起不来。
# mkdir /root/hermes_agent/global_agent_opt_hermes
# chown -R 10000:10000 /root/hermes_agent/global_agent_opt_hermes
# rm -rf /root/hermes_agent/global_agent_opt_hermes
# docker cp hermes_global_agent:/opt/hermes /root/hermes_agent/global_agent_opt_hermes
# docker stop hermes_global_agent
# docker rm hermes_global_agent
# docker run -d –name hermes_global_agent –restart unless-stopped -v /root/hermes_agent/global_agent:/opt/data -v /root/hermes_agent/global_agent_opt_hermes/hermes:/opt/hermes -p 8642:8642 -e TZ=Asia/Shanghai nousresearch/hermes-agent:v2026.5.7 gateway run
进入docker里面root用户的bash：
# docker exec -it hermes_global_agent /bin/bash
在docker里面增加编辑vi命令(有内置nano代替)和配置时区、路径等：
# apt-get install -y vim
# vi /etc/bash.bashrc
…
alias ll=’ls -lF’
export TZ=Asia/Shanghai
export PATH=$PATH:/opt/hermes/.venv/bin/
重启容器，切换到hermes用户执行命令：
# docker restart hermes_global_agent
# docker exec -u hermes -it hermes_global_agent /bin/bash
在docker里面执行命令：
$ date
$ /opt/hermes/.venv/bin/hermes –version
$ hermes –version
Hermes Agent v0.11.0 (2026.4.23)
Project: /opt/hermes
Python: 3.13.5
OpenAI SDK: 2.32.0
$ hermes chat
$ exit
hermes agent的直接服务器方式安装：
阿里云上搞一台ECS主机(ubuntu 24.x)来安装，并且挂载一个阿里云nas目录到ubuntu的 /home/ 目录。
安装：
# ln /usr/bin/python3 /usr/bin/python
# apt install ripgrep ffmpeg
# apt install build-essential python3-dev libffi-dev
# apt install playwright chromium
# adduser hermes
# visudo
…
hermes ALL=(ALL) NOPASSWD: ALL
…
# su – hermes
$ wgethttps://hermes-agent.nousresearch.com/install.sh
$ chmod +x install.sh
$ ./install.sh –branch v2026.6.19 –skip-setup
安装后的路径：
Config: /home/hermes/.hermes/config.yaml
API Keys: /home/hermes/.hermes/.env
Data: /home/hermes/.hermes/cron/, sessions/, logs/
Code: /home/hermes/.hermes/hermes-agent
$ exit 因为bashrc被改所以需要重新进入bash。
$ hermes version
Hermes Agent v0.13.0 (2026.5.7)
Project: /home/hermes/.hermes/hermes-agent
Python: 3.11.15
初始化配置：
$ hermes setup
根据提示建立初始化配置
作为service启动gateway，sudo到root用户执行：
$ hermes gateway -h
$ sudo /home/hermes/.local/bin/hermes gateway install –system –run-as-user hermes
会建立这个service：/etc/systemd/system/hermes-gateway.service，然后启动：
$ sudo /home/hermes/.local/bin/hermes gateway start –system
$ sudo /home/hermes/.local/bin/hermes gateway status –system
$ sudo systemctl start hermes-gateway
$ sudo systemctl stop hermes-gateway
$ sudo systemctl status hermes-gateway
后续测试中要用相同python版本执行的话(VENV方式)：
$ source .hermes/hermes-agent/venv/bin/activate
$ python -V
升级（只能升级到main分支的最新位置）：
$ hermes update -h
hermes update –check
//hermes update –backup 需要比较久，接受风险不用也可以
hermes update
tail -f ~/.hermes/logs/update.log
hermes –version
hermes config check
hermes setup –migrate
hermes doctor
//sudo systemctl restart hermes-gateway 会自动重启
hermes gateway status
使用：
hermes -h
初始化命令：
hermes setup -h
hermes setup Re-run the full wizard
hermes setup model Change model/provider
hermes setup terminal Change terminal backend
hermes setup gateway Configure messaging
hermes setup tools Configure tool providers
hermes config View current settings
hermes config edit Open config in your editor
hermes config set <key> <value> Set a specific value
配置文件位置：
/opt/data/config.yaml： 对应主机的/root/hermes_agent/global_agent/config.yaml文件（参考docker启动命令的-v参数）
/opt/data/.env
启动命令：
hermes 默认chat模式启动
hermes chat 以chat模式启动，进入后/help可以看到内置的交互命令，/exit命令退出。
hermes gateway 以网关方式启动和飞书等交互
hermes doctor 检查问题模式启动
配置log级别（DEBUG级别可以看到和模型的交互，打印会很多，谨慎）：
修改config.yaml:
…
logging:
level: DEBUG
使用多模型：
其实使用openrouter直接作为主模型api key，然后让openrouter自行选更好。
$ hermes fallback -h
$ hermes fallback add
(○) Configure auxiliary models… 这项是修改辅助模型，选其他项则是选fallback模型。
使用多profile建立多agent对应到多个飞书机器人：
注意：飞书机器人不能重复用，一个agent对应一个。
$ hermes profile -h
$ hermes profile list
$ hermes profile create test_1
在profiles/test_1目录下建立了新的profile，后续可以用test_1直接替换hermes命令使用该profile，比如：
$ which test_1
/home/hermes/.local/bin/test_1
$ test_1 setup
$ test_1 chat
$ test_1 gateway run
$ test_1 gateway install 同样可以作为系统服务安装自动启动
$ test_1 gateway start
$ test_1 gateway status 通过systemctl 是看不到该服务的
$ test_1 gateway stop
$ test_1 gateway uninstall
其实test_1是个脚本，实际执行的是hermes -p test_1带后面的参数。
可以导入导出：
$ hermes gateway list
$ hermes profile export test_1
$ hermes profile import test_1
编辑SOUL.md文件定义灵魂：
# vi SOUL.md
你是一个专业的游戏行业从业者。
你是一个专业的数据分析师。
你是一个可以利用各种API操作游戏投放和变现的专业人员。
你绝对不会执行bigquery、mysql数据库的任何删除和更新操作，只做查询。
…
定时任务（保存在/opt/data/cron/jobs.json文件里）：
注意：可以在hermes chat里面直接建立，也可以在飞书单聊里面发消息建立，飞书中斜杠命令和chat里面的斜杠命令是不一样的，飞书只是子集。
# docker exec -u hermes -it d360b120dc3b /bin/bash
$ hermes chat
/cron
/cron list
/cron add “every 2h” “Check server status” [–skill blogwatcher]
/cron edit <job_id> 。。。
/cron pause <job_id>
/cron resume <job_id>
/cron run <job_id>
/cron remove <job_id>
是否在对话中显示调用工具类进度配置：
$ vi config.yaml
…
display:
tool_progress: verbose # all/off/new/verbose
发送定时任务的消息给飞书的指定接收人：
注意：接收人必须能被识别，比如已经与机器人对话过的人员或者机器人所在群组。
直接在hermes chat或者飞书单聊里面输入类似消息即可： 每分钟向“测试ai群聊功能” 群组发送”你好”消息
直接在hermes chat或者飞书单聊里面输入类似消息即可： 每10分钟向“测试ai群聊功能” 群组的”张三”发送”你好”消息
在回复带有回复消息的消息时，不要自动创建飞书话题：
$ vi gateway/platforms/feishu.py
…
# thread_id = getattr(message, “thread_id”, None) or getattr(message, “root_id”, None) or None
thread_id = getattr(message, “thread_id”, None) or None
reply_to_message_id = (
找到skill接入：
从这里可以找到一些官方的直接安装，安装后的目录在skills/下面
https://hermes-agent.nousresearch.com/docs/skills
$ hermes skills install google-sheets –force
$ hermes skills list
$ ls -l /opt/data/skills/google-sheets/
$ vi /opt/data/.env
…
MATON_API_KEY=v2.8hhhhAAAAAA
$ hermes chat
> 看下这个表里广告的总价格：https://docs.google.com/spreadsheets/d/1U9vnxvPUCs0qCKQy1IQ68Ei0o5rIPcZg879ir82SVek/edit?gid=0#gid=0
编写自己的SKILL：
$ mkdir skills/feishu_button_text_interactive
$ vi skills/feishu_button_text_interactive/SKILL.md
—
name: feishu_interactive
description: “飞书机器人回复的消息的尾部加入按钮并接受按钮反馈”
version: 1.0.0
author: Zhang san
license: MIT
metadata:
hermes:
tags: [Feishu, 消息, 按钮]
related_skills: []
—
# Agent回复的飞书消息尾部加入按钮并接受按钮反馈
AI Agent每次回复飞书消息时，在飞书尾部加入按钮，在用户点击按钮给出反馈时，保存反馈信息。
## 先决条件
– 收到飞书请求AI执行某项功能的消息
– AI需要回复飞书消息
## 执行
– Agent回复消息的时候加入一个按钮，按钮内容为”满意”
– 用户在飞书聊天点击”满意”时，Agent在log里输出 用户id、时间、反馈满意
找到mcp server接入：
配置bigquery的官方mcp（不再好用）：
在有glcoud的host主机上执行即可(不用在docker容器里面)，启用bigquery项目的mcp server：
# gcloud beta services mcp enable bigquery.googleapis.com –project=data-collect
# gcloud beta services mcp disable bigquery.googleapis.com –project=data-collect
启用后，回来修改配置，这类修改都可以1分钟内直接生效，不用重启docker：
$ vi config.yaml
…
mcp_servers:
bigquery:
url: https://bigquery.googleapis.com/mcp
env:
GOOGLE_CLOUD_PROJECT: data-collect
GOOGLE_APPLICATION_CREDENTIALS: /opt/data/common_config/data-collect-11111111.json
编写自己的MCP Server：
$ mkdir skills/mcp_test/
$ vi skills/mcp_test/my_ip.py
# /// script
# dependencies = [
# “fastmcp”,
# “mcp”,
# “httpx”,
# ]
# ///
from mcp.server.fastmcp import FastMCP
import httpx
mcp = FastMCP(“我的公网IP和地理位置”)
@mcp.tool()
async def get_ip_info() -> str:
“””
获取当前出口 IP 的地理位置信息
当用户输入我的出口ip、我的公网ip、我的外网ip的时候调用该工具
参数：
无
“””
async with httpx.AsyncClient() as client:
resp = await client.get(“https://ipapi.co/json/“)
data = resp.json()
return f”IP: {data.get(‘ip’)}, 城市: {data.get(‘city’)}, 运营商: {data.get(‘org’)}”
if __name__ == “__main__”:
mcp.run(transport=”stdio”)
先用uvx测试一下：
$ uvx –with mcp python /opt/data/skills/maxshu_test/mcp_test/my_ip.py
先输入json格式初始化服务：
{“jsonrpc”: “2.0”, “id”: 1, “method”: “initialize”, “params”: {“capabilities”: {}, “clientInfo”: {“name”: “test-client”, “version”: “1.0”}, “protocolVersion”: “2024-11-05”}}
返回：
{“jsonrpc”:”2.0″,”id”:1,”result”:{“protocolVersion”:”2024-11-05″,”capabilities”:{“experimental”:{},”prompts”:{“listChanged”:false},”resources”:{“subscribe”:false,”listChanged”:false},”tools”:{“listChanged”:false}},”serverInfo”:{“name”:”我的公网IP和地理位置”,”version”:”1.27.0″}}}
再输入对tool的调用：
{“jsonrpc”: “2.0”, “id”: 2, “method”: “tools/call”, “params”: {“name”: “get_ip_info”, “arguments”: {}}}
返回：
{“jsonrpc”:”2.0″,”id”:2,”result”:{“content”:[{“type”:”text”,”text”:”IP: 123.123.123.123, 城市: 测试城市, 运营商: 测试运营商”}],”structuredContent”:{“result”:”IP: 123.123.123.123, 城市: 测试城市, 运营商: 测试运营商”},”isError”:false}}
然后配置到hermes：
$ vi config.yaml
…
my_ip_position:
command: “uvx”
args:
– “–with”
– “mcp”
– “python”
– “/opt/data/skills/mcp_test/my_ip.py”
env:
TESTA: not_use
enabled: true
测试一下：
$ hermes mcp list
$ hermes mcp test my_ip_position
输出mcp里面的tool工具名和注释表示正常可以使用了。
$ hermes chat
> 我的公网ip
第一次吃可能要指定使用该mcp服务，然后交互几次，后续能看到调用了这个mcp服务。
限制飞书交互中能用的斜杠类指令：
只能修改源码，修改 hermes_cli/commands.py 里面的 COMMAND_REGISTRY 常量定义，不让聊天交互的命令增加 “cli_only=True” 即可，例子：
CommandDef(“history”, “Show conversation history”, “Session”, cli_only=True),
如果只让聊天交互的话，设置 “gateway_only=True” 即可（一般不用），例子：
CommandDef(“deny”, “Deny a pending dangerous command”, “Session”, gateway_only=True),
当然每个skill也会变成一个斜杠指令，这个一般不用限制。
把对话进度存入log文件，而不是从聊天上打印出来：
$ hermes-agent/gateway/platforms/feishu.py
…
if not self._client:
return SendResult(success=False, error=”Not connected”)
chat_info = await self.get_chat_info(chat_id, None)
sender_profile=chat_info.get(“sender_profile”) or {“user_id”: None}
logger.info(f’发送：chat_id: {chat_id}, user_id: {sender_profile[“user_id”]}, reply_to: {repr(reply_to)}, metadata: {repr(metadata)}, content: \n{content}’)
for handler in logger.handlers:
handler.flush()
formatted = self.format_message(content)
chunks = self.truncate_message(formatted, self.MAX_MESSAGE_LENGTH)
…
async def get_chat_info(self, chat_id: str, sender_profile: Dict[str, Optional[str]]) -> Dict[str, Any]:
“””Return real chat metadata from Feishu when available.”””
…
“type”: self._map_chat_type(raw_chat_type),
“raw_type”: raw_chat_type or None,
“sender_profile”: sender_profile,
… 所有4个这种地方：
sender_profile = await self._resolve_sender_profile(user_id_obj)
chat_info = await self.get_chat_info(chat_id, sender_profile)
…
和飞书对话消息的尾部加入自定义按钮之类：
记得在.env文件中加入 MY_ACTION_1_LOG_FILE=/opt/data/logs/MY_ACTION_1_LOG_FILE.log 的环境变量值
只能修改源码：
$ vi gateway/platforms/feishu.py
…
_APPROVAL_LABEL_MAP: Dict[str, str] = {
“once”: “Approved once”,
“session”: “Approved for session”,
“always”: “Approved permanently”,
“deny”: “Denied”,
}
# my_action_1: two-button interaction card — button labels and value keys
_MY_ACTION_1_LABEL_MAP: Dict[str, str] = {
“btn_yes”: “👍赞”,
“btn_no”: “👎踩”,
}
_MY_ACTION_1_LOG_FILE_ENV = “MY_ACTION_1_LOG_FILE” # env var to override log path
_FEISHU_BOT_MSG_TRACK_SIZE = 512 # LRU size for tracking sent message IDs
…
hermes_action = action_value.get(“hermes_action”) if isinstance(action_value, dict) else None
my_action_1 = action_value.get(“my_action_1”) if isinstance(action_value, dict) else None
if hermes_action:
return self._handle_approval_card_action(event=event, action_value=action_value, loop=loop)
if my_action_1:
self._submit_on_loop(loop, self._handle_my_action_1_card_action(event=event, action_value=action_value))
if P2CardActionTriggerResponse is None:
return None
response = P2CardActionTriggerResponse()
if CallBackCard is not None:
card = CallBackCard()
card.type = “raw”
card.data = self._build_my_action_1_resolved_card(btn_key=my_action_1)
response.card = card
return response
self._submit_on_loop(loop, self._handle_card_action_event(data))
…
async def send_my_action_1(
self,
chat_id: str,
ai_output: str,
title: str = “请选择操作”,
metadata: Optional[Dict[str, Any]] = None,
) -> “SendResult”:
“””Send an interactive card with two buttons for my_action_1.
The buttons carry “my_action_1“ in their value dict so that
“_on_card_action_trigger“ can intercept them and record the
chat_id, timestamp, ai_output and the button choice.
“””
if not self._client:
from gateway.platforms.base import SendResult # local import to avoid circular
return SendResult(success=False, error=”Not connected”)
try:
action_id = str(uuid.uuid4())
# ai_output embedded in value so the handler can recover it without a
# separate state store; truncated to avoid oversized card payloads.
ai_output_preview = ai_output[:500] + “…” if len(ai_output) > 500 else ai_output
def _btn(label: str, btn_key: str, btn_type: str = “default”) -> dict:
return {
“tag”: “button”,
“text”: {“tag”: “plain_text”, “content”: label},
“type”: btn_type,
“element_id”: “btn_”+btn_key,
“value”: {
“my_action_1”: btn_key,
“action_id”: action_id,
“chat_id”: chat_id,
“ai_output”: ai_output_preview,
},
}
content_preview = ai_output[:1000] + “…” if len(ai_output) > 1000 else ai_output
card = {
“schema”: “2.0”,
# “config”: {“wide_screen_mode”: True},
# “header”: {
# “title”: {“content”: title, “tag”: “plain_text”},
# “template”: “blue”,
# },
“body”: {
“direction”: “horizontal”,
“elements”: [
# {
# “tag”: “markdown”,
# “content”: content_preview,
# },
# {
# “tag”: “action”,
# “actions”: [
_btn(_MY_ACTION_1_LABEL_MAP[“btn_yes”], “btn_yes”, “primary”),
_btn(_MY_ACTION_1_LABEL_MAP[“btn_no”], “btn_no”, “danger”),
# ],
# },
],
}
}
payload = json.dumps(card, ensure_ascii=False)
response = await self._feishu_send_with_retry(
chat_id=chat_id,
msg_type=”interactive”,
payload=payload,
reply_to=None,
metadata=metadata,
)
return self._finalize_send_result(response, “send_my_action_1 failed”)
except Exception as exc:
logger.warning(“[Feishu] send_my_action_1 failed: %s”, exc)
from gateway.platforms.base import SendResult # local import to avoid circular
return SendResult(success=False, error=str(exc))
@staticmethod
def _build_my_action_1_resolved_card(*, btn_key: str) -> Dict[str, Any]:
“””Build raw card JSON shown after a my_action_1 button is clicked.”””
label = _MY_ACTION_1_LABEL_MAP.get(btn_key, btn_key)
icon = “✅” if btn_key == “btn_yes” else “❌”
# color = “green” if btn_key == “btn_yes” else “red”
return {
“schema”: “2.0”,
# “config”: {“wide_screen_mode”: True},
# “header”: {
# “title”: {“content”: f”{icon} {label}”, “tag”: “plain_text”},
# “template”: color,
# },
“body”: {
“elements”: [
# {
# “tag”: “button”,
# “element_id”: “btn_”+btn_key,
# “text”: {“tag”: “plain_text”, “content”: “已记录”},
# “disabled”: True,
# },
{
“tag”: “markdown”,
“content”: f”{icon} 已记录您的选择：**{label}**”,
},
],
}
}
async def _handle_my_action_1_card_action(
self, *, event: Any, action_value: Dict[str, Any]
) -> None:
“””Persist my_action_1 button-click data to a JSONL log file.
Saved fields:
– “chat_id“ — the originating chat
– “timestamp“ — ISO-8601 UTC time of the click
– “ai_output“ — the AI output embedded in the card value
– “btn_key“ — which button was clicked (“btn_yes“ / “btn_no“)
– “action_id“ — unique ID linking click to the original card send
– “operator_open_id“ — open_id of the user who clicked
“””
btn_key = action_value.get(“my_action_1”, “”)
action_id = action_value.get(“action_id”, “”)
chat_id = action_value.get(“chat_id”, “”)
ai_output = action_value.get(“ai_output”, “”)
operator = getattr(event, “operator”, None)
operator_open_id = str(getattr(operator, “open_id”, “”) or “”)
timestamp = datetime.utcnow().isoformat() + “Z”
record = {
“chat_id”: chat_id,
“timestamp”: timestamp,
“ai_output”: ai_output,
“btn_key”: btn_key,
“action_id”: action_id,
“operator_open_id”: operator_open_id,
}
logger.info(
“[Feishu] my_action_1 click: chat_id=%s btn=%s action_id=%s operator=%s”,
chat_id,
btn_key,
action_id,
operator_open_id,
)
try:
log_path = os.environ.get(
_MY_ACTION_1_LOG_FILE_ENV,
str(Path(__file__).parent.parent.parent / “my_action_1_log.jsonl”),
)
await asyncio.to_thread(
self._append_my_action_1_record, log_path, record
)
except Exception as exc:
logger.error(“[Feishu] my_action_1 failed to persist record: %s”, exc)
@staticmethod
def _append_my_action_1_record(log_path: str, record: Dict[str, Any]) -> None:
“””Synchronous helper: append one JSON line to the JSONL log file.”””
with open(log_path, “a”, encoding=”utf-8″) as fh:
fh.write(json.dumps(record, ensure_ascii=False) + “\n”)
async def _resolve_approval(self, approval_id: Any, choice: str, user_name: str) -> None:
“””Pop approval state and unblock the waiting agent thread.”””
state = self._approval_state.pop(approval_id, None)
if not state:
logger.debug(“[Feishu] Approval %s already resolved or unknown”, approval_id)
return
try:
from tools.approval import resolve_gateway_approval
count = resolve_gateway_approval(state[“session_key”], choice)
logger.info(
“Feishu button resolved %d approval(s) for session %s (choice=%s, user=%s)”,
count, state[“session_key”], choice, user_name,
)
except Exception as exc:
logger.error(“Failed to resolve gateway approval from Feishu button: %s”, exc)
…
$ vi gateway\platforms\base.py
…
self._schedule_ephemeral_delete(
chat_id=event.source.chat_id,
message_id=result.message_id,
ttl_seconds=_ephemeral_ttl,
)
# my_action_1: after a successful AI text reply on Feishu,
# send a two-button interactive card so the user can confirm
# or dismiss. Gated on hasattr so all other platform adapters
# are completely unaffected.
if (
result.success
and text_content
and hasattr(self, “send_my_action_1”)
and getattr(event.source, “platform”, None) == Platform.FEISHU
):
try:
await self.send_my_action_1(
chat_id=event.source.chat_id,
ai_output=text_content,
metadata=_thread_metadata,
)
except Exception as _ma1_err:
logger.warning(
“[%s] send_my_action_1 failed (non-fatal): %s”,
self.name, _ma1_err,
)
# Human-like pacing delay between text and media
human_delay = self._get_human_delay()
…
回复卡片时，不再生成另一个文本回复：
$ gateway/platforms/feishu.py
def _build_outbound_payload(self, content: str) -> tuple[str, str]:
# Detect interactive card JSON:
# if the content looks like a Card 2.0 JSON object, send it directly
# as msg_type=interactive so the LLM can return a card as its final
# reply without needing a separate REST call.
stripped = content.strip()
if stripped.startswith(“{“):
try:
parsed = json.loads(stripped)
if isinstance(parsed, dict):
# Card 2.0: {“schema”: “2.0”, “body”: {…}, …}
if parsed.get(“schema”) == “2.0” and “body” in parsed:
return “interactive”, json.dumps(parsed, ensure_ascii=False)
# Wrapped: {“msg_type”: “interactive”, “content”: {…}}
if parsed.get(“msg_type”) == “interactive” and “content” in parsed:
inner = parsed[“content”]
if isinstance(inner, str):
return “interactive”, inner
return “interactive”, json.dumps(inner, ensure_ascii=False)
except (json.JSONDecodeError, ValueError):
pass
# Feishu post-type ‘md’ elements do not render markdown tables;
# sending table content as post causes the message to appear blank
# on the client. Force plain text for anything that looks like a
# markdown table.
if _MARKDOWN_TABLE_RE.search(content):
text_payload = {“text”: content}
return “text”, json.dumps(text_payload, ensure_ascii=False)
if _MARKDOWN_HINT_RE.search(content):
return “post”, _build_markdown_post_payload(content)
text_payload = {“text”: content}
return “text”, json.dumps(text_payload, ensure_ascii=False)
增加plugin生成图片：
$ cd plugins/image_gen
$ mkdir gemini-imagen
$ cd gemini-imagen
$ vi __init__.py
from __future__ import annotations
import os
import sys
import base64
import logging
from agent.image_gen_provider import (
ImageGenProvider, error_response, success_response, save_b64_image
)
logger = logging.getLogger(__name__)
SA_JSON = ‘/opt/data/common_config/data-collect-11111111.json’
GCP_PROJECT = ‘data-collect’
GCP_LOCATION = ‘us-central1’
DEFAULT_MODEL = ‘imagen-3.0-generate-002’
ASPECT_RATIO_MAP = {
‘landscape’: ’16:9′,
‘portrait’: ‘9:16’,
‘square’: ‘1:1’,
}
class GeminiImagenProvider(ImageGenProvider):
@property
def name(self) -> str:
return “gemini-imagen”
@property
def display_name(self) -> str:
return “Google Imagen (Vertex AI / Service Account)”
def is_available(self) -> bool:
if not os.path.exists(SA_JSON):
return False
try:
sys.path.insert(0, ‘/opt/data/python_packages’)
import google.genai
from google.oauth2 import service_account
except ImportError:
return False
return True
def list_models(self):
return [
{“id”: “imagen-3.0-generate-002”, “label”: “Imagen 3.0”, “note”: “recommended”},
{“id”: “imagen-4.0-generate-001”, “label”: “Imagen 4.0”, “note”: “high quality”},
{“id”: “imagen-4.0-fast-generate-001”, “label”: “Imagen 4.0 Fast”, “note”: “fast”},
{“id”: “imagen-4.0-ultra-generate-001”, “label”: “Imagen 4.0 Ultra”, “note”: “best quality”},
]
def generate(self, prompt: str, **kwargs) -> dict:
try:
sys.path.insert(0, ‘/opt/data/python_packages’)
os.environ[‘GOOGLE_GENAI_USE_VERTEXAI’] = ‘true’
from google import genai
from google.genai import types
from google.oauth2 import service_account
credentials = service_account.Credentials.from_service_account_file(
SA_JSON,
scopes=[‘https://www.googleapis.com/auth/cloud-platform‘]
)
client = genai.Client(
vertexai=True,
project=GCP_PROJECT,
location=GCP_LOCATION,
credentials=credentials,
)
model_id = kwargs.get(“model”) or DEFAULT_MODEL
raw_ratio = kwargs.get(“aspect_ratio”, “landscape”)
aspect_ratio = ASPECT_RATIO_MAP.get(raw_ratio, raw_ratio)
response = client.models.generate_images(
model=model_id,
prompt=prompt,
config=types.GenerateImagesConfig(
number_of_images=1,
aspect_ratio=aspect_ratio,
),
)
img_bytes = None
for img in response.generated_images:
img_bytes = img.image.image_bytes
break
client.close()
if img_bytes is None:
return error_response(
error=”Model returned no image.”,
provider=self.name,
prompt=prompt,
)
b64 = base64.b64encode(img_bytes).decode()
path = save_b64_image(b64, prefix=”gemini”, extension=”png”)
return success_response(
image=str(path),
model=model_id,
prompt=prompt,
aspect_ratio=aspect_ratio,
provider=self.name,
)
except Exception as e:
logger.exception(“Gemini image generation failed”)
return error_response(error=str(e), provider=self.name, prompt=prompt)
def register(ctx):
ctx.register_image_gen_provider(GeminiImagenProvider())
$ vi plugin.yaml
name: gemini-imagen
version: 1.0.0
description: “Google Gemini Imagen 3 image generation backend. Saves generated images to $HERMES_HOME/cache/images/.”
author: custom
kind: backend
$ vi $HERMES_HOME/.env
…
GOOGLE_GENAI_USE_VERTEXAI=true # 表示使用企业版