Skip to content

feat(hermes): Phase 2d-i — system-prompt.md injection on all 3 dispatch paths #276

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
HongmingWang-Rabbit merged 1 commit into from
Apr 15, 2026
Merged

feat(hermes): Phase 2d-i — system-prompt.md injection on all 3 dispatch paths #276

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion workspace-template/adapters/hermes/adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,10 @@ async def create_executor(self, config: AdapterConfig): # pragma: no cover
# Resolve API key: prefer workspace secrets (runtime_config), then env vars
hermes_api_key = config.runtime_config.get("hermes_api_key") or None

executor = create_executor(hermes_api_key=hermes_api_key)
executor = create_executor(
hermes_api_key=hermes_api_key,
config_path=config.config_path, # Phase 2d-i: system-prompt.md injection
)

# Override model from config if provided
model = config.model
Expand Down
116 changes: 73 additions & 43 deletions workspace-template/adapters/hermes/executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ def create_executor(
hermes_api_key: Optional[str] = None,
provider: Optional[str] = None,
model: Optional[str] = None,
config_path: Optional[str] = None,
):
"""Create and return a LangGraph-compatible executor for the Hermes adapter.

Expand All @@ -65,12 +66,17 @@ def create_executor(
model:
Override the provider's default model. Passed straight through to
``chat.completions.create``.
config_path:
Path to the workspace's ``/configs`` directory. Phase 2d-i reads
``system-prompt.md`` from here on every ``execute()`` call and
passes the content as a system instruction to the native SDK.
Optional — omit to skip system-prompt injection (tests do this).

Returns
-------
HermesA2AExecutor
A ready-to-use executor wired with the resolved api_key + base_url
+ model.
+ model + config_path.

Raises
------
Expand All @@ -86,6 +92,7 @@ def create_executor(
provider_cfg=cfg,
api_key=hermes_api_key,
model=model or cfg.default_model,
config_path=config_path,
)

# Path 2/3: registry resolution (either explicit provider name or auto-detect).
Expand All @@ -101,6 +108,7 @@ def create_executor(
provider_cfg=cfg,
api_key=api_key,
model=model or cfg.default_model,
config_path=config_path,
)


Expand All @@ -123,12 +131,18 @@ def __init__(
api_key: str,
model: str,
heartbeat=None,
config_path: Optional[str] = None,
):
self.provider_cfg = provider_cfg
self.api_key = api_key
self.base_url = provider_cfg.base_url
self.model = model
self._heartbeat = heartbeat
# Phase 2d-i: config_path lets execute() read /configs/system-prompt.md
# on each turn and pass it to the native SDK's `system=` /
# `system_instruction=` / prepended message. Optional because older
# callers + tests construct executors directly.
self._config_path = config_path

# ------------------------------------------------------------------
# History → provider-specific message list converters
Expand Down Expand Up @@ -203,17 +217,17 @@ async def _do_openai_compat(
self,
user_message: str,
history: "list[tuple[str, str]] | None" = None,
system_prompt: Optional[str] = None,
) -> str:
"""OpenAI-compat inference — used by every provider with auth_scheme='openai'.

13 of the 15 registered providers route here. Uses ``openai.AsyncOpenAI``
pointed at the provider's base_url; every provider's API is wire-
compatible with the OpenAI Chat Completions shape.

Phase 2c: accepts multi-turn history. The old single-``task_text`` call
shape (pre-2c) is preserved — pass the flattened text as ``user_message``
with no history and the call degrades gracefully to the original
behavior. See ``_history_to_openai_messages`` for the conversion.
Phase 2c: accepts multi-turn history.
Phase 2d-i: accepts optional system_prompt, prepended as a
``{"role":"system"}`` message per the OpenAI Chat Completions convention.
"""
import openai

Expand All @@ -222,6 +236,8 @@ async def _do_openai_compat(
base_url=self.base_url,
)
messages = self._history_to_openai_messages(user_message, history or [])
if system_prompt:
messages = [{"role": "system", "content": system_prompt}, *messages]
response = await client.chat.completions.create(
model=self.model,
messages=messages,
Expand All @@ -232,20 +248,20 @@ async def _do_anthropic_native(
self,
user_message: str,
history: "list[tuple[str, str]] | None" = None,
system_prompt: Optional[str] = None,
) -> str:
"""Native Anthropic Messages API inference.

Uses the official ``anthropic`` Python SDK for correct tool-calling,
vision, and extended-thinking semantics that don't translate cleanly
through the OpenAI-compat shim.

If the ``anthropic`` package is not installed in the workspace image,
we raise a clear error rather than silently falling back to the
OpenAI-compat path — silent fallback would mask the fidelity loss
(tool_use blocks become plain text, vision gets stripped, etc.).

Phase 2a: single-turn text in, text out. Phase 2c: multi-turn history.
Tools + vision remain Phase 2d.
Phase 2a: single-turn text.
Phase 2c: multi-turn history.
Phase 2d-i: optional system_prompt passed via Anthropic's native
top-level ``system=`` parameter — NOT as a message in the messages
list (Anthropic's Messages API requires system prompts to be at the
top level, not inline like OpenAI).
"""
try:
import anthropic
Expand All @@ -259,11 +275,14 @@ async def _do_anthropic_native(

client = anthropic.AsyncAnthropic(api_key=self.api_key)
messages = self._history_to_anthropic_messages(user_message, history or [])
response = await client.messages.create(
model=self.model,
max_tokens=4096,
messages=messages,
)
create_kwargs: dict = {
"model": self.model,
"max_tokens": 4096,
"messages": messages,
}
if system_prompt:
create_kwargs["system"] = system_prompt
response = await client.messages.create(**create_kwargs)
# response.content is a list of ContentBlock; for text-only the first
# block is a TextBlock with a .text attribute.
if response.content and hasattr(response.content[0], "text"):
Expand All @@ -274,6 +293,7 @@ async def _do_gemini_native(
self,
user_message: str,
history: "list[tuple[str, str]] | None" = None,
system_prompt: Optional[str] = None,
) -> str:
"""Native Google Gemini ``generateContent`` inference.

Expand All @@ -282,17 +302,15 @@ async def _do_gemini_native(
thinking config. These all get stripped or mis-translated through
the OpenAI-compat ``/v1beta/openai`` shim.

If the ``google-genai`` package is not installed in the workspace
image, raise a clear error with install instructions rather than
silently falling back to the OpenAI-compat shim (same fail-loud
semantics as the anthropic path).

Phase 2b: single-turn text in, text out. Phase 2c: multi-turn history
via Gemini's ``contents=[{role,parts}]`` shape (note: role is
``"user"`` / ``"model"``, NOT ``"assistant"``).
Phase 2b: single-turn text.
Phase 2c: multi-turn history via Gemini's ``contents=[{role,parts}]``
shape (note: role is ``"user"`` / ``"model"``, NOT ``"assistant"``).
Phase 2d-i: system_prompt passed via native
``config.system_instruction`` — Gemini's top-level system field.
"""
try:
from google import genai # type: ignore[import-not-found]
from google.genai import types as genai_types # type: ignore[import-not-found]
except ImportError as exc: # pragma: no cover — exercised by test_missing_sdk
raise RuntimeError(
"Hermes gemini native path requires the `google-genai` package. "
Expand All @@ -303,10 +321,15 @@ async def _do_gemini_native(

client = genai.Client(api_key=self.api_key)
contents = self._history_to_gemini_contents(user_message, history or [])
response = await client.aio.models.generate_content(
model=self.model,
contents=contents,
)
generate_kwargs: dict = {
"model": self.model,
"contents": contents,
}
if system_prompt:
generate_kwargs["config"] = genai_types.GenerateContentConfig(
system_instruction=system_prompt,
)
response = await client.aio.models.generate_content(**generate_kwargs)
# response.text is the flattened text across all parts of the first
# candidate. For text-only that's the whole reply.
return response.text or ""
Expand All @@ -315,27 +338,29 @@ async def _do_inference(
self,
user_message: str,
history: "list[tuple[str, str]] | None" = None,
system_prompt: Optional[str] = None,
) -> str:
"""Dispatch to the right inference path based on provider auth_scheme.

Phase 2c: takes ``user_message`` + optional ``history`` list-of-tuples,
passes through to the chosen path. Each path has its own history →
provider-message conversion via the static helpers above.
Phase 2c: multi-turn history.
Phase 2d-i: optional system_prompt is passed through to the native
system field of whichever path wins dispatch (OpenAI ``{role:system}``
/ Anthropic ``system=`` / Gemini ``system_instruction=``).
"""
scheme = self.provider_cfg.auth_scheme
if scheme == "anthropic":
return await self._do_anthropic_native(user_message, history)
return await self._do_anthropic_native(user_message, history, system_prompt)
if scheme == "gemini":
return await self._do_gemini_native(user_message, history)
return await self._do_gemini_native(user_message, history, system_prompt)
if scheme == "openai":
return await self._do_openai_compat(user_message, history)
return await self._do_openai_compat(user_message, history, system_prompt)
# Unknown scheme — treat as openai-compat for forward-compat with any
# future provider the registry adds without yet having a native path.
logger.warning(
"Hermes: unknown auth_scheme=%r for provider=%s — falling back to openai-compat",
scheme, self.provider_cfg.name,
)
return await self._do_openai_compat(user_message, history)
return await self._do_openai_compat(user_message, history, system_prompt)

# ------------------------------------------------------------------
# AgentExecutor interface
Expand All @@ -344,12 +369,13 @@ async def _do_inference(
async def execute(self, context, event_queue): # pragma: no cover
"""Execute a Hermes inference request and push the reply to event_queue.

Phase 2c: passes the conversation history to the dispatch layer as a
structured list of (role, text) turns instead of flattening via
``build_task_text``. Each provider path converts the list into its
native multi-turn message shape (OpenAI messages, Anthropic messages,
or Gemini contents). This gives the model its native multi-turn
awareness for instruction following.
Phase 2c: multi-turn history.
Phase 2d-i: reads ``/configs/system-prompt.md`` via
``executor_helpers.get_system_prompt`` each turn (supports hot-reload)
and passes the text to the dispatch layer. Each provider path uses
its native system field — Anthropic's top-level ``system=``, Gemini's
``system_instruction=`` via ``GenerateContentConfig``, or OpenAI's
``{"role":"system"}`` message at the head of the messages list.
"""
from a2a.utils import new_agent_text_message
from adapters.shared_runtime import (
Expand All @@ -358,6 +384,7 @@ async def execute(self, context, event_queue): # pragma: no cover
extract_message_text,
set_current_task,
)
from executor_helpers import get_system_prompt

user_message = extract_message_text(context)
if not user_message:
Expand All @@ -368,7 +395,10 @@ async def execute(self, context, event_queue): # pragma: no cover

try:
history = extract_history(context)
reply = await self._do_inference(user_message, history)
system_prompt = (
get_system_prompt(self._config_path) if self._config_path else None
)
reply = await self._do_inference(user_message, history, system_prompt)
except Exception as exc:
logger.exception("Hermes executor error: %s", exc)
reply = f"Hermes error: {exc}"
Expand Down
Loading
Loading