File size: 1,560 Bytes
1c9c13f
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
"""ν…ŒμŠ€νŠΈ 더블 β€” ScriptedLlm: μ •ν•΄μ§„ 응닡을 μˆœμ„œλŒ€λ‘œ λ‚΄λ†“λŠ” BaseLlm.

LLM 호좜 없이 νŒŒμ΄ν”„λΌμΈ(κ°€λ“œβ†’μΊλ¦­ν„°β†’λ””λ ‰ν„°β†’λ„κ΅¬β†’state)을 κ²°μ •μ μœΌλ‘œ κ²€μ¦ν•œλ‹€.
ν”„λ‘œλ•μ…˜ κ²½λ‘œλŠ” LiteLlm(engine/model.py) κ·ΈλŒ€λ‘œμ΄λ©°, 이 더블은 tests/ μ „μš©μ΄λ‹€.
"""

from __future__ import annotations

from typing import AsyncGenerator

from google.adk.models.base_llm import BaseLlm
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.genai import types
from pydantic import ConfigDict


def text_response(text: str) -> LlmResponse:
    return LlmResponse(content=types.Content(role="model", parts=[types.Part(text=text)]))


def tool_call(name: str, **args) -> types.Part:
    return types.Part(function_call=types.FunctionCall(name=name, args=args))


def tool_response(*parts: types.Part) -> LlmResponse:
    return LlmResponse(content=types.Content(role="model", parts=list(parts)))


class ScriptedLlm(BaseLlm):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    model: str = "scripted"
    script: list[LlmResponse] = []
    requests: list[LlmRequest] = []  # κ²€μ¦μš©: μ‹€μ œλ‘œ μ£Όμž…λœ instruction 확인

    async def generate_content_async(
        self, llm_request: LlmRequest, stream: bool = False
    ) -> AsyncGenerator[LlmResponse, None]:
        self.requests.append(llm_request)
        if not self.script:
            yield text_response("(λŒ€λ³Έ μ†Œμ§„)")
            return
        yield self.script.pop(0)