Hermes 与 Agent 工程实战：从模型选型到生产级智能体的完整代码实现

<tool_call>
{"name": "get_weather", "arguments": {"city": "上海"}}
</tool_call>

<tool_response>
{"temperature": 24, "condition": "多云"}
</tool_response>

<json>
{"intent": "query_weather", "slots": {"city": "上海"}}
</json>

<|im_start|>system
你是一个企业助手。<|im_end|>
<|im_start|>user
帮我查一下上海天气。<|im_end|>
<|im_start|>assistant
<tool_call>
{"name": "get_weather", "arguments": {"city": "上海"}}
</tool_call><|im_end|>
<|im_start|>tool
<tool_response>
{"temperature": 24}
</tool_response><|im_end|>
<|im_start|>assistant
上海当前 24 度，多云。<|im_end|>

你可以使用以下工具：

<tools>
[
  {
    "name": "get_weather",
    "description": "查询城市天气",
    "parameters": {
      "type": "object",
      "properties": {
        "city": {"type": "string", "description": "城市名"}
      },
      "required": ["city"]
    }
  }
]
</tools>

调用工具时，输出：
<tool_call>
{"name": "工具名", "arguments": {...}}
</tool_call>

用户目标
  -> 意图理解
  -> 上下文检索（记忆 + RAG）
  -> 规划
  -> 选择工具
  -> 执行工具
  -> 观察结果
  -> 反思
  -> 继续或输出

hermes-agent/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── db.py
│   ├── models.py
│   ├── hermes_client.py
│   ├── prompt.py
│   ├── tools.py
│   ├── parser.py
│   ├── agent.py
│   ├── memory.py
│   ├── security.py
│   ├── audit.py
│   ├── evaluator.py
│   └── config.py
├── tests/
│   └── test_hermes_agent.py
├── requirements.txt
└── Dockerfile

pip install fastapi uvicorn pydantic sqlmodel httpx pytest

fastapi
uvicorn[standard]
pydantic
sqlmodel
httpx
pytest

import os

HERMES_BASE_URL = os.getenv("HERMES_BASE_URL", "http://localhost:8000/v1")
HERMES_MODEL = os.getenv("HERMES_MODEL", "NousResearch/Hermes-3-Llama-3.1-8B")
HERMES_API_KEY = os.getenv("HERMES_API_KEY", "not-needed")
HERMES_TIMEOUT = float(os.getenv("HERMES_TIMEOUT", "60"))
HERMES_MAX_TOKENS = int(os.getenv("HERMES_MAX_TOKENS", "1024"))
HERMES_TEMPERATURE = float(os.getenv("HERMES_TEMPERATURE", "0.2"))

import httpx
from typing import List, Dict, Any
from app.config import (
    HERMES_BASE_URL,
    HERMES_MODEL,
    HERMES_API_KEY,
    HERMES_TIMEOUT,
    HERMES_MAX_TOKENS,
    HERMES_TEMPERATURE,
)


class HermesClient:
    """
    兼容 OpenAI Chat Completions 协议。
    可对接 vLLM / Ollama / llama.cpp server / TGI。
    """

    def __init__(
        self,
        base_url: str = HERMES_BASE_URL,
        model: str = HERMES_MODEL,
        api_key: str = HERMES_API_KEY,
    ):
        self.base_url = base_url.rstrip("/")
        self.model = model
        self.api_key = api_key
        self.client = httpx.Client(timeout=HERMES_TIMEOUT)

    def chat(
        self,
        messages: List[Dict[str, str]],
        tools: List[Dict[str, Any]] | None = None,
        temperature: float = HERMES_TEMPERATURE,
        max_tokens: int = HERMES_MAX_TOKENS,
        stop: List[str] | None = None,
    ) -> Dict[str, Any]:
        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
        }
        if tools:
            payload["tools"] = tools
        if stop:
            payload["stop"] = stop

        resp = self.client.post(
            f"{self.base_url}/chat/completions",
            json=payload,
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
        )
        resp.raise_for_status()
        return resp.json()

    def complete(
        self,
        prompt: str,
        temperature: float = HERMES_TEMPERATURE,
        max_tokens: int = HERMES_MAX_TOKENS,
    ) -> str:
        payload = {
            "model": self.model,
            "prompt": prompt,
            "temperature": temperature,
            "max_tokens": max_tokens,
        }
        resp = self.client.post(
            f"{self.base_url}/completions",
            json=payload,
            headers={"Authorization": f"Bearer {self.api_key}"},
        )
        resp.raise_for_status()
        data = resp.json()
        return data["choices"][0]["text"]

from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List


@dataclass
class ToolSpec:
    name: str
    description: str
    parameters: Dict[str, Any]
    handler: Callable[..., Any]
    requires_roles: List[str] = field(default_factory=lambda: ["member"])
    risk_level: str = "low"
    requires_confirmation: bool = False

    def to_openai_schema(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            },
        }


class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, ToolSpec] = {}

    def register(self, spec: ToolSpec):
        self._tools[spec.name] = spec

    def get(self, name: str) -> ToolSpec | None:
        return self._tools.get(name)

    def schemas_for_role(self, role: str) -> List[dict]:
        return [
            t.to_openai_schema()
            for t in self._tools.values()
            if role in t.requires_roles
        ]

    def execute(self, name: str, arguments: dict, context: dict) -> dict:
        spec = self.get(name)
        if not spec:
            return {"ok": False, "error": f"未知工具：{name}"}
        try:
            result = spec.handler(**arguments)
            return {"ok": True, "tool": name, "result": result}
        except TypeError as e:
            return {"ok": False, "tool": name, "error": f"参数错误：{e}"}
        except Exception as e:
            return {"ok": False, "tool": name, "error": f"执行失败：{e}"}


# ---------- 示例工具 ----------

def tool_get_weather(city: str) -> dict:
    return {"city": city, "temperature": 24, "condition": "多云"}


def tool_calculator(expression: str) -> dict:
    allowed = set("0123456789+-*/(). ")
    if not set(expression) <= allowed:
        raise ValueError("表达式含非法字符")
    value = eval(expression, {"__builtins__": {}}, {})
    return {"expression": expression, "value": value}


def tool_search_docs(query: str, top_k: int = 3) -> dict:
    return {
        "query": query,
        "results": [
            {"title": f"文档 {i+1}", "snippet": f"关于 {query} 的说明 {i+1}"}
            for i in range(top_k)
        ],
    }


def tool_create_ticket(title: str, priority: str = "medium") -> dict:
    return {
        "ticket_id": f"TK{abs(hash(title)) % 100000}",
        "title": title,
        "priority": priority,
        "status": "open",
    }


def tool_send_email(to: str, subject: str, body: str) -> dict:
    return {"to": to, "subject": subject, "status": "queued"}


def build_default_registry() -> ToolRegistry:
    reg = ToolRegistry()

    reg.register(ToolSpec(
        name="get_weather",
        description="查询指定城市的当前天气",
        parameters={
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名，例如 上海"},
            },
            "required": ["city"],
        },
        handler=tool_get_weather,
        requires_roles=["member", "manager", "admin"],
        risk_level="low",
    ))

    reg.register(ToolSpec(
        name="calculator",
        description="计算数学表达式",
        parameters={
            "type": "object",
            "properties": {
                "expression": {"type": "string", "description": "如 2+3*4"},
            },
            "required": ["expression"],
        },
        handler=tool_calculator,
        requires_roles=["member", "manager", "admin"],
        risk_level="low",
    ))

    reg.register(ToolSpec(
        name="search_docs",
        description="在企业知识库中检索文档",
        parameters={
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "top_k": {"type": "integer", "default": 3},
            },
            "required": ["query"],
        },
        handler=tool_search_docs,
        requires_roles=["member", "manager", "admin"],
        risk_level="low",
    ))

    reg.register(ToolSpec(
        name="create_ticket",
        description="创建一个工单",
        parameters={
            "type": "object",
            "properties": {
                "title": {"type": "string"},
                "priority": {
                    "type": "string",
                    "enum": ["low", "medium", "high", "urgent"],
                },
            },
            "required": ["title"],
        },
        handler=tool_create_ticket,
        requires_roles=["member", "manager", "admin"],
        risk_level="medium",
    ))

    reg.register(ToolSpec(
        name="send_email",
        description="发送邮件（高风险，需要确认）",
        parameters={
            "type": "object",
            "properties": {
                "to": {"type": "string"},
                "subject": {"type": "string"},
                "body": {"type": "string"},
            },
            "required": ["to", "subject", "body"],
        },
        handler=tool_send_email,
        requires_roles=["manager", "admin"],
        risk_level="high",
        requires_confirmation=True,
    ))

    return reg

import json
from typing import List


SYSTEM_TEMPLATE = """你是 Hermes Agent，一个可靠的企业智能体。

## 你的职责
- 理解用户目标；
- 在需要时调用工具；
- 基于工具结果给出准确回答；
- 不确定时明确说明，不要编造。

## 工具调用格式
调用工具时，必须输出：
<tool_call>
{{"name": "工具名", "arguments": {{...}}}}
</tool_call>

## 可用工具
<tools>
{tools_json}
</tools>

## 行为准则
- 每次只调用一个工具；
- 参数必须符合 Schema；
- 工具失败时分析原因，可重试一次；
- 高风险操作需等待人工确认；
- 不要输出密钥、密码等敏感信息。
"""


def build_system_prompt(tool_schemas: List[dict], extra: str = "") -> str:
    tools_json = json.dumps(tool_schemas, ensure_ascii=False, indent=2)
    prompt = SYSTEM_TEMPLATE.format(tools_json=tools_json)
    if extra:
        prompt += f"\n## 补充说明\n{extra}\n"
    return prompt

import json
import re
from dataclasses import dataclass
from typing import Optional


TOOL_CALL_RE = re.compile(
    r"<tool_call>\s*(\{.*?\})\s*</tool_call>",
    re.DOTALL,
)


@dataclass
class ToolCall:
    name: str
    arguments: dict


@dataclass
class ParsedOutput:
    kind: str                       # "tool_call" / "final"
    content: str = ""
    tool_call: Optional[ToolCall] = None


def parse_hermes_output(text: str) -> ParsedOutput:
    """
    解析 Hermes 输出，支持：
    - <tool_call>{"name":..., "arguments":{...}}</tool_call>
    - 纯文本最终回答
    """
    if not text:
        return ParsedOutput(kind="final", content="")

    match = TOOL_CALL_RE.search(text)
    if match:
        raw = match.group(1)
        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            return ParsedOutput(
                kind="final",
                content=f"工具调用解析失败：{raw}",
            )

        name = data.get("name") or data.get("tool")
        arguments = data.get("arguments") or data.get("parameters") or {}
        if not name:
            return ParsedOutput(
                kind="final",
                content=f"工具调用缺少 name：{raw}",
            )

        return ParsedOutput(
            kind="tool_call",
            tool_call=ToolCall(name=name, arguments=arguments),
        )

    # 兼容 OpenAI tool_calls 字段（若 vLLM 已解析）
    return ParsedOutput(kind="final", content=text.strip())


def strip_tool_tags(text: str) -> str:
    return TOOL_CALL_RE.sub("", text).strip()

import json
from typing import List, Dict, Any

from app.hermes_client import HermesClient
from app.prompt import build_system_prompt
from app.parser import parse_hermes_output
from app.tools import ToolRegistry


MAX_STEPS = 8
MAX_REPEATS = 2


class HermesAgent:
    def __init__(
        self,
        client: HermesClient,
        tools: ToolRegistry,
        role: str = "member",
        extra_system: str = "",
    ):
        self.client = client
        self.tools = tools
        self.role = role
        self.extra_system = extra_system

    def run(self, user_message: str) -> dict:
        tool_schemas = self.tools.schemas_for_role(self.role)
        system_prompt = build_system_prompt(tool_schemas, self.extra_system)

        messages: List[Dict[str, str]] = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_message},
        ]

        steps = []
        action_history = []

        for step in range(MAX_STEPS):
            resp = self.client.chat(messages=messages)
            content = resp["choices"][0]["message"]["content"] or ""

            parsed = parse_hermes_output(content)

            if parsed.kind == "final":
                return {
                    "reply": parsed.content,
                    "steps": steps,
                    "raw_model_output": content,
                }

            tc = parsed.tool_call
            key = f"{tc.name}:{json.dumps(tc.arguments, sort_keys=True)}"
            action_history.append(key)

            if action_history.count(key) > MAX_REPEATS:
                return {
                    "reply": f"检测到重复调用 {tc.name}，已停止。",
                    "steps": steps,
                    "error": "repeat_action",
                }

            steps.append({"tool": tc.name, "arguments": tc.arguments})

            result = self.tools.execute(tc.name, tc.arguments, {"role": self.role})

            messages.append({"role": "assistant", "content": content})
            messages.append({
                "role": "tool",
                "content": json.dumps(result, ensure_ascii=False),
            })

        return {
            "reply": "达到最大步数，任务未完成。",
            "steps": steps,
            "error": "max_steps",
        }

import concurrent.futures


def run_parallel_tools(agent: HermesAgent, calls: list) -> list:
    """
    并行执行多个工具调用。
    """
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
        futures = {
            pool.submit(
                agent.tools.execute, c["name"], c["arguments"], {"role": agent.role}
            ): c
            for c in calls
        }
        for fut in concurrent.futures.as_completed(futures):
            results.append(fut.result())
    return results

class MultiAgentOrchestrator:
    """
    简单编排：研究员 + 写手 + 审校。
    """

    def __init__(self, client: HermesClient, tools: ToolRegistry):
        self.researcher = HermesAgent(client, tools, role="member",
                                      extra_system="你负责检索资料，只输出事实。")
        self.writer = HermesAgent(client, tools, role="member",
                                  extra_system="你负责把资料整理成简洁报告。")
        self.reviewer = HermesAgent(client, tools, role="manager",
                                    extra_system="你负责审校事实与逻辑，指出问题。")

    def run(self, topic: str) -> dict:
        research = self.researcher.run(f"检索并整理主题：{topic}")
        draft = self.writer.run(f"基于以下资料写报告：\n{research['reply']}")
        review = self.reviewer.run(f"审校以下报告：\n{draft['reply']}")
        return {
            "research": research,
            "draft": draft,
            "review": review,
        }

from datetime import datetime
from typing import Optional
from sqlmodel import SQLModel, Field


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(index=True, unique=True)
    role: str = "member"
    enabled: bool = True
    created_at: datetime = Field(default_factory=datetime.utcnow)


class Conversation(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(index=True)
    title: str = ""
    created_at: datetime = Field(default_factory=datetime.utcnow)


class MessageRecord(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    conversation_id: int = Field(index=True)
    role: str
    content: str
    created_at: datetime = Field(default_factory=datetime.utcnow)


class Memory(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(index=True)
    key: str
    value: str
    created_at: datetime = Field(default_factory=datetime.utcnow)


class ToolAudit(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int
    conversation_id: int
    tool_name: str
    arguments: str
    result: str
    status: str = "ok"
    created_at: datetime = Field(default_factory=datetime.utcnow)

from sqlmodel import Session, select
from app.models import Memory


def remember(db: Session, user_id: int, key: str, value: str):
    existing = db.exec(
        select(Memory).where(Memory.user_id == user_id, Memory.key == key)
    ).first()
    if existing:
        existing.value = value
        db.add(existing)
    else:
        db.add(Memory(user_id=user_id, key=key, value=value))
    db.commit()


def recall(db: Session, user_id: int) -> dict:
    rows = db.exec(select(Memory).where(Memory.user_id == user_id)).all()
    return {r.key: r.value for r in rows}


def build_memory_prompt(db: Session, user_id: int) -> str:
    mem = recall(db, user_id)
    if not mem:
        return ""
    lines = ["已知用户信息："]
    for k, v in mem.items():
        lines.append(f"- {k}: {v}")
    return "\n".join(lines)

from fastapi import Header, HTTPException, Depends
from sqlmodel import Session, select
from app.db import get_session
from app.models import User

API_KEYS = {
    "admin-key": "admin",
    "manager-key": "manager",
    "member-key": "member",
}


def get_current_user(
    x_api_key: str = Header(..., alias="X-API-Key"),
    session: Session = Depends(get_session),
) -> User:
    role = API_KEYS.get(x_api_key)
    if not role:
        raise HTTPException(status_code=401, detail="Invalid API Key")

    user = session.exec(select(User).where(User.username == role)).first()
    if not user or not user.enabled:
        raise HTTPException(status_code=401, detail="User not found")
    return user


def require_role(*roles):
    def checker(user: User = Depends(get_current_user)):
        if user.role not in roles:
            raise HTTPException(status_code=403, detail="Forbidden")
        return user
    return checker

import json
from sqlmodel import Session, select
from app.models import ToolAudit


def log_tool_call(
    db: Session,
    user_id: int,
    conversation_id: int,
    tool_name: str,
    arguments: dict,
    result: dict,
    status: str = "ok",
):
    db.add(ToolAudit(
        user_id=user_id,
        conversation_id=conversation_id,
        tool_name=tool_name,
        arguments=json.dumps(arguments, ensure_ascii=False),
        result=json.dumps(result, ensure_ascii=False, default=str),
        status=status,
    ))
    db.commit()


def list_audits(db: Session, user_id: int | None = None, limit: int = 200):
    stmt = select(ToolAudit)
    if user_id:
        stmt = stmt.where(ToolAudit.user_id == user_id)
    return db.exec(stmt.order_by(ToolAudit.id.desc()).limit(limit)).all()

import json
from typing import List, Dict, Any
from app.parser import parse_hermes_output


def evaluate_tool_selection(
    cases: List[Dict[str, Any]],
    agent,
) -> dict:
    """
    cases: [{"input": "...", "expected_tool": "get_weather"}]
    """
    total = len(cases)
    correct = 0
    details = []

    for c in cases:
        result = agent.run(c["input"])
        steps = result.get("steps", [])
        selected = steps[0]["tool"] if steps else None
        ok = selected == c["expected_tool"]
        correct += int(ok)
        details.append({
            "input": c["input"],
            "expected": c["expected_tool"],
            "selected": selected,
            "ok": ok,
        })

    return {
        "total": total,
        "correct": correct,
        "accuracy": round(correct / total, 4) if total else 0.0,
        "details": details,
    }


def evaluate_parser(cases: List[Dict[str, str]]) -> dict:
    """
    cases: [{"raw": "<tool_call>...</tool_call>", "expect_kind": "tool_call"}]
    """
    correct = 0
    details = []

    for c in cases:
        parsed = parse_hermes_output(c["raw"])
        ok = parsed.kind == c["expect_kind"]
        correct += int(ok)
        details.append({
            "raw": c["raw"][:60],
            "expected": c["expect_kind"],
            "got": parsed.kind,
            "ok": ok,
        })

    total = len(cases)
    return {
        "total": total,
        "correct": correct,
        "accuracy": round(correct / total, 4) if total else 0.0,
        "details": details,
    }

TOOL_SELECTION_CASES = [
    {"input": "上海天气怎么样", "expected_tool": "get_weather"},
    {"input": "帮我算一下 12*8", "expected_tool": "calculator"},
    {"input": "查一下报销制度", "expected_tool": "search_docs"},
    {"input": "给 IT 提个工单，打印机坏了", "expected_tool": "create_ticket"},
]

PARSER_CASES = [
    {
        "raw": '<tool_call>{"name": "get_weather", "arguments": {"city": "上海"}}</tool_call>',
        "expect_kind": "tool_call",
    },
    {
        "raw": "上海今天 24 度，多云。",
        "expect_kind": "final",
    },
]

import json
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from sqlmodel import Session

from app.db import init_db, get_session
from app.models import User, Conversation, MessageRecord
from app.security import get_current_user, require_role
from app.hermes_client import HermesClient
from app.tools import build_default_registry
from app.agent import HermesAgent
from app.audit import list_audits, log_tool_call
from app.memory import build_memory_prompt

app = FastAPI(title="Hermes Agent Engineering")

client = HermesClient()
registry = build_default_registry()


@app.on_event("startup")
def on_startup():
    init_db()


class ChatRequest(BaseModel):
    message: str
    conversation_id: int | None = None


@app.get("/health")
def health():
    return {"status": "ok", "model": client.model, "base_url": client.base_url}


@app.get("/tools")
def list_tools(user: User = Depends(get_current_user)):
    return [
        {
            "name": t.name,
            "description": t.description,
            "risk_level": t.risk_level,
            "requires_confirmation": t.requires_confirmation,
        }
        for t in registry._tools.values()
        if user.role in t.requires_roles
    ]


@app.post("/chat")
def chat(
    payload: ChatRequest,
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    if payload.conversation_id:
        conv = session.get(Conversation, payload.conversation_id)
        if not conv or conv.user_id != user.id:
            raise HTTPException(status_code=404, detail="Conversation not found")
    else:
        conv = Conversation(user_id=user.id, title=payload.message[:30])
        session.add(conv)
        session.commit()
        session.refresh(conv)

    session.add(MessageRecord(
        conversation_id=conv.id,
        role="user",
        content=payload.message,
    ))
    session.commit()

    memory_prompt = build_memory_prompt(session, user.id)

    agent = HermesAgent(
        client=client,
        tools=registry,
        role=user.role,
        extra_system=memory_prompt,
    )

    result = agent.run(payload.message)

    session.add(MessageRecord(
        conversation_id=conv.id,
        role="assistant",
        content=result.get("reply", ""),
    ))
    session.commit()

    for step in result.get("steps", []):
        log_tool_call(
            session,
            user_id=user.id,
            conversation_id=conv.id,
            tool_name=step["tool"],
            arguments=step["arguments"],
            result={"step": "executed"},
        )

    return {
        "conversation_id": conv.id,
        "reply": result.get("reply"),
        "steps": result.get("steps", []),
        "error": result.get("error"),
    }


@app.get("/audit/tool-calls")
def audit_tool_calls(
    user: User = Depends(require_role("admin", "manager")),
    session: Session = Depends(get_session),
):
    return list_audits(session)

pip install vllm

python -m vllm.entrypoints.openai.api_server \
  --model NousResearch/Hermes-3-Llama-3.1-8B \
  --served-model-name hermes-3-8b \
  --host 0.0.0.0 \
  --port 8000 \
  --max-model-len 8192 \
  --gpu-memory-utilization 0.9

export HERMES_BASE_URL=http://localhost:8000/v1
export HERMES_MODEL=hermes-3-8b

ollama pull hermes3
ollama serve

export HERMES_BASE_URL=http://localhost:11434/v1
export HERMES_MODEL=hermes3
export HERMES_API_KEY=ollama

./llama-server \
  -m hermes-3-8b.Q4_K_M.gguf \
  --host 0.0.0.0 \
  --port 8080 \
  -c 8192

export HERMES_BASE_URL=http://localhost:8080/v1
export HERMES_MODEL=hermes-3-8b

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app ./app

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

docker build -t hermes-agent .
docker run -p 8000:8000 \
  -e HERMES_BASE_URL=http://host.docker.internal:8000/v1 \
  -e HERMES_MODEL=hermes-3-8b \
  hermes-agent

import os
import pytest
from fastapi.testclient import TestClient

os.environ["DATABASE_URL"] = "sqlite:///./test_hermes.db"

from app.main import app  # noqa: E402
from app.parser import parse_hermes_output  # noqa: E402
from app.tools import build_default_registry  # noqa: E402

client = TestClient(app)


def test_health():
    r = client.get("/health")
    assert r.status_code == 200


def test_parser_tool_call():
    raw = '<tool_call>{"name": "get_weather", "arguments": {"city": "上海"}}</tool_call>'
    parsed = parse_hermes_output(raw)
    assert parsed.kind == "tool_call"
    assert parsed.tool_call.name == "get_weather"
    assert parsed.tool_call.arguments == {"city": "上海"}


def test_parser_final():
    parsed = parse_hermes_output("上海今天 24 度。")
    assert parsed.kind == "final"
    assert "24" in parsed.content


def test_parser_malformed():
    parsed = parse_hermes_output("<tool_call>{bad json}</tool_call>")
    assert parsed.kind == "final"
    assert "解析失败" in parsed.content


def test_registry_schema():
    reg = build_default_registry()
    schemas = reg.schemas_for_role("member")
    names = [s["function"]["name"] for s in schemas]
    assert "get_weather" in names
    assert "calculator" in names


def test_registry_permission():
    reg = build_default_registry()
    schemas = reg.schemas_for_role("member")
    names = [s["function"]["name"] for s in schemas]
    assert "send_email" not in names   # 高风险工具只对 manager/admin 开放

    schemas_mgr = reg.schemas_for_role("manager")
    names_mgr = [s["function"]["name"] for s in schemas_mgr]
    assert "send_email" in names_mgr


def test_tool_execute():
    reg = build_default_registry()
    result = reg.execute("calculator", {"expression": "2+3*4"}, {})
    assert result["ok"] is True
    assert result["result"]["value"] == 14


def test_tool_execute_error():
    reg = build_default_registry()
    result = reg.execute("calculator", {"expression": "__import__('os')"}, {})
    assert result["ok"] is False


def test_tools_endpoint():
    r = client.get("/tools", headers={"X-API-Key": "member-key"})
    assert r.status_code == 200
    names = [t["name"] for t in r.json()]
    assert "get_weather" in names
    assert "send_email" not in names


def test_audit_requires_role():
    r = client.get("/audit/tool-calls", headers={"X-API-Key": "member-key"})
    assert r.status_code == 403

    r = client.get("/audit/tool-calls", headers={"X-API-Key": "admin-key"})
    assert r.status_code == 200

pytest -v

Hermes = 能稳定调用工具的开放基座
Agent  = 把模型能力转化为可执行任务的系统

Agent 系统 = Hermes 模型 + 工具系统 + Agent Loop + 状态记忆 + 权限审计 + 评估

hermes-agent/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── db.py
│   ├── models.py
│   ├── config.py
│   ├── hermes_client.py
│   ├── prompt.py
│   ├── tools.py
│   ├── parser.py
│   ├── agent.py
│   ├── memory.py
│   ├── security.py
│   ├── audit.py
│   └── evaluator.py
├── tests/
│   └── test_hermes_agent.py
├── requirements.txt
└── Dockerfile

pip install -r requirements.txt
# 启动 Hermes 服务（vLLM 或 Ollama）
pytest -v
uvicorn app.main:app --reload