mirror of
https://github.com/hwchase17/langchain
synced 2024-10-29 17:07:25 +00:00
c0d67420e5
<!-- Thank you for contributing to LangChain! Replace this entire comment with: - Description: a description of the change, - Issue: the issue # it fixes (if applicable), - Dependencies: any dependencies required for this change, - Tag maintainer: for a quicker response, tag the relevant maintainer (see below), - Twitter handle: we announce bigger features on Twitter. If your PR gets announced and you'd like a mention, we'll gladly shout you out! Please make sure your PR is passing linting and testing before submitting. Run `make format`, `make lint` and `make test` to check this locally. See contribution guidelines for more information on how to write/run tests, lint, etc: https://github.com/hwchase17/langchain/blob/master/.github/CONTRIBUTING.md If you're adding a new integration, please include: 1. a test for the integration, preferably unit tests that do not rely on network access, 2. an example notebook showing its use. These live is docs/extras directory. If no one reviews your PR within a few days, please @-mention one of @baskaryan, @eyurtsev, @hwchase17, @rlancemartin. -->
48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
from abc import abstractmethod
|
|
from typing import Any, List, Optional
|
|
|
|
from langchain.callbacks.manager import Callbacks
|
|
from langchain.chains.llm import LLMChain
|
|
|
|
from langchain_experimental.plan_and_execute.schema import Plan, PlanOutputParser
|
|
from langchain_experimental.pydantic_v1 import BaseModel
|
|
|
|
|
|
class BasePlanner(BaseModel):
|
|
"""Base planner."""
|
|
|
|
@abstractmethod
|
|
def plan(self, inputs: dict, callbacks: Callbacks = None, **kwargs: Any) -> Plan:
|
|
"""Given input, decide what to do."""
|
|
|
|
@abstractmethod
|
|
async def aplan(
|
|
self, inputs: dict, callbacks: Callbacks = None, **kwargs: Any
|
|
) -> Plan:
|
|
"""Given input, asynchronously decide what to do."""
|
|
|
|
|
|
class LLMPlanner(BasePlanner):
|
|
"""LLM planner."""
|
|
|
|
llm_chain: LLMChain
|
|
"""The LLM chain to use."""
|
|
output_parser: PlanOutputParser
|
|
"""The output parser to use."""
|
|
stop: Optional[List] = None
|
|
"""The stop list to use."""
|
|
|
|
def plan(self, inputs: dict, callbacks: Callbacks = None, **kwargs: Any) -> Plan:
|
|
"""Given input, decide what to do."""
|
|
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:
|
|
"""Given input, asynchronously decide what to do."""
|
|
llm_response = await self.llm_chain.arun(
|
|
**inputs, stop=self.stop, callbacks=callbacks
|
|
)
|
|
return self.output_parser.parse(llm_response)
|