104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
# Copyright 2026 Google LLC
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
#
|
|
# Portions of this file are derived from mobile-use (https://github.com/minitap-ai/mobile-use)
|
|
# Copyright 2025-2026 Minitap, Inc. Licensed under the Apache License 2.0.
|
|
|
|
from pathlib import Path
|
|
|
|
from jinja2 import Template
|
|
from langchain_core.messages import HumanMessage, SystemMessage
|
|
from artemis.context import ArtemisContext
|
|
from artemis.data_engine.trace import trace
|
|
from artemis.services.llm import get_llm, invoke_llm_with_timeout_message, with_fallback
|
|
from artemis.utils.logger import get_logger
|
|
from pydantic import BaseModel, Field, model_validator
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class HopperError(RuntimeError):
|
|
"""Extraction failed; this says nothing about whether the requested data exists."""
|
|
|
|
|
|
class _CyFunctionDetectorMeta(type):
|
|
def __instancecheck__(self, instance):
|
|
name = type(instance).__name__
|
|
return (
|
|
name
|
|
in (
|
|
"cyfunction",
|
|
"cython_function_or_method",
|
|
"builtin_function_or_method",
|
|
)
|
|
or "cyfunction" in name.lower()
|
|
)
|
|
|
|
|
|
class CyFunctionDetector(metaclass=_CyFunctionDetectorMeta):
|
|
pass
|
|
|
|
|
|
class HopperOutput(BaseModel):
|
|
model_config = {"ignored_types": (CyFunctionDetector,)}
|
|
found: bool = Field(description="True if the requested data was found, False otherwise.")
|
|
output: str | None = Field(description="The extracted data if found, null otherwise.")
|
|
reason: str = Field(
|
|
description="A short explanation of what you looked for"
|
|
+ " and how you decided what to extract."
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_found_output(self):
|
|
if self.found and (not self.output or not self.output.strip()):
|
|
raise ValueError("A successful extraction must contain a nonempty output")
|
|
return self
|
|
|
|
|
|
@trace(type="agent", name="hopper")
|
|
async def hopper(
|
|
ctx: ArtemisContext,
|
|
request: str,
|
|
data: str,
|
|
use_fallback: bool = True,
|
|
) -> HopperOutput:
|
|
logger.info(f"Starting Hopper Agent (use_fallback={use_fallback})")
|
|
system_message = Template(
|
|
Path(__file__).parent.joinpath("hopper.md").read_text(encoding="utf-8")
|
|
).render()
|
|
messages = [
|
|
SystemMessage(content=system_message),
|
|
HumanMessage(content=f"{request}\nHere is the data you must dig:\n{data}"),
|
|
]
|
|
|
|
async def extract(model) -> HopperOutput:
|
|
response = await invoke_llm_with_timeout_message(model.ainvoke(messages))
|
|
return HopperOutput.model_validate(response)
|
|
|
|
try:
|
|
llm = get_llm(ctx=ctx, name="hopper", is_utils=True).with_structured_output(HopperOutput)
|
|
if use_fallback:
|
|
llm_fallback = get_llm(
|
|
ctx=ctx, name="hopper", is_utils=True, use_fallback=True
|
|
).with_structured_output(HopperOutput)
|
|
return await with_fallback(
|
|
main_call=lambda: extract(llm),
|
|
fallback_call=lambda: extract(llm_fallback),
|
|
)
|
|
else:
|
|
return await extract(llm)
|
|
except Exception as e:
|
|
logger.error(f"Hopper LLM invocation failed: {e}")
|
|
raise HopperError("模型未能完成数据提取，请查看模型请求诊断。") from e
|