애플리케이션 템플릿 맞춤설정

컬렉션을 사용해 정리하기 내 환경설정을 기준으로 콘텐츠를 저장하고 분류하세요.

from typing import Any, Callable, Iterable, Sequence

class CLASS_NAME:
    def __init__(
            self,
            model: str,
            tools: Sequence[Callable],
            project: str,
            location: str,
        ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.location = location

    def set_up(self):
        """All unpickle-able logic should go here.

        The .set_up() method should not be called for an object that is being
        prepared for deployment.
        """
        import vertexai
        from langchain_google_vertexai import ChatVertexAI
        from langchain.agents import AgentExecutor
        from langchain.agents.format_scratchpad.tools import format_to_tool_messages
        from langchain.agents.output_parsers.tools import ToolsAgentOutputParser
        from langchain.tools.base import StructuredTool
        from langchain_core import prompts

        vertexai.init(project=self.project, location=self.location)

        prompt = {
            "input": lambda x: x["input"],
            "agent_scratchpad": (
                lambda x: format_to_tool_messages(x["intermediate_steps"])
            ),
        } | prompts.ChatPromptTemplate.from_messages([
            ("user", "{input}"),
            prompts.MessagesPlaceholder(variable_name="agent_scratchpad"),
        ])

        llm = ChatVertexAI(model_name=self.model_name)
        if self.tools:
            llm = llm.bind_tools(tools=self.tools)

        self.agent_executor = AgentExecutor(
            agent=prompt | llm | ToolsAgentOutputParser(),
            tools=[StructuredTool.from_function(tool) for tool in self.tools],
        )

    def query(self, input: str):
        """Query the application.

        Args:
            input: The user prompt.

        Returns:
            The output of querying the application with the given input.
        """
        return self.agent_executor.invoke(input={"input": input})

    def stream_query(self, input: str) -> Iterable[Any]:
        """Query the application and stream the output.

        Args:
            input: The user prompt.

        Yields:
            Chunks of the response as they become available.
        """
        for chunk in self.agent_executor.stream(input={"input": input}):
            yield chunk

agent = CLASS_NAME(
    model=model,  # Required.
    tools=[get_exchange_rate],  # Optional.
    project=PROJECT_ID,
    location=LOCATION,
)
agent.set_up()

response = agent.query(
    input="What is the exchange rate from US dollars to Swedish currency?"
)

{"input": "What is the exchange rate from US dollars to Swedish currency?",
 # ...
 "output": "For 1 US dollar you will get 10.7345 Swedish Krona."}

import pprint

for chunk in agent.stream_query(
    input="What is the exchange rate from US dollars to Swedish currency?"
):
    # Use pprint with depth=1 for a more concise, high-level view of the
    # streamed output.
    # To see the full content of the chunk, use:
    # print(chunk)
    pprint.pprint(chunk, depth=1)

{'actions': [...], 'messages': [...]}
{'messages': [...], 'steps': [...]}
{'messages': [...],
 'output': 'The exchange rate from US dollars to Swedish currency is 1 USD to '
           '10.5751 SEK. \n'}

from typing import Dict, List, Any, Iterable

class CLASS_NAME:
    # ... other methods ...

    def custom_method_1(...):
        # ...

    def custom_method_2(...):
        # ...

    def custom_stream_method_1(...) -> Iterable[Any]:
        # ...

    def custom_stream_method_2(...) -> Iterable[Any]:
        # ...

    def register_operations(self) -> Dict[str, List[str]]:
        return {
            "": [
                "custom_method_1", "custom_method_2",
            ],
            "stream": [
                "custom_stream_method_1", "custom_stream_method_2",
            ],
        }

response = agent.custom_method_1(
    input="What is the exchange rate from US dollars to Swedish currency?"
)

for chunk in agent.custom_stream_method_1(
    input="What is the exchange rate from US dollars to Swedish currency?"
):
    # Use pprint with depth=1 for a more concise, high-level view of the
    # streamed output.
    # To see the full content of the chunk, use:
    # print(chunk)
    pprint.pprint(chunk, depth=1)

from typing import Dict, List, Any

class CLASS_NAME:
    # ... other methods ...

    def custom_method_1(...):
        # ...

    def custom_method_2(...):
        # ...

    def custom_stream_method_1(...) -> Iterable[Any]:
        # ...

    def custom_stream_method_2(...) -> Iterable[Any]:
        # ...

    def register_operations(self) -> Dict[str, List[str]]:
        return {
            # The list of synchronous methods to be registered as operations.
            "": [
                "custom_method_1", "custom_method_2",
            ],
        }