Source code for langchain_experimental.plan_and_execute.planners.base

from abc import abstractmethod
from typing import Any, List, Optional

from langchain.chains.llm import LLMChain
from langchain_core.callbacks.manager import Callbacks

from langchain_experimental.plan_and_execute.schema import Plan, PlanOutputParser
from langchain_experimental.pydantic_v1 import BaseModel


class BasePlanner(BaseModel):
    """基础规划器。"""

    @abstractmethod
    def plan(self, inputs: dict, callbacks: Callbacks = None, **kwargs: Any) -> Plan:
        """给定输入，决定要做什么。"""

    @abstractmethod
    async def aplan(
        self, inputs: dict, callbacks: Callbacks = None, **kwargs: Any
    ) -> Plan:
        """给定输入，异步决定要做什么。"""


class LLMPlanner(BasePlanner):
    """LLM规划器。"""

    llm_chain: LLMChain
    """要使用的LLM链。"""
    output_parser: PlanOutputParser
    """要使用的输出解析器。"""
    stop: Optional[List] = None
    """要使用的停用词列表。"""

    def plan(self, inputs: dict, callbacks: Callbacks = None, **kwargs: Any) -> Plan:
        """给定输入，决定要做什么。"""
        llm_response = self.llm_chain.run(**inputs, stop=self.stop, callbacks=callbacks)
        return self.output_parser.parse(llm_response)

    async def aplan(
        self, inputs: dict, callbacks: Callbacks = None, **kwargs: Any
    ) -> Plan:
        """给定输入，异步决定要做什么。"""
        llm_response = await self.llm_chain.arun(
            **inputs, stop=self.stop, callbacks=callbacks
        )
        return self.output_parser.parse(llm_response)