Source code for langchain.agents.react.base
"""实现了来自https://arxiv.org/pdf/2210.03629.pdf 的ReAct论文的链条。"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, List, Optional, Sequence
from langchain_core._api import deprecated
from langchain_core.documents import Document
from langchain_core.language_models import BaseLanguageModel
from langchain_core.prompts import BasePromptTemplate
from langchain_core.pydantic_v1 import Field
from langchain_core.tools import BaseTool, Tool
from langchain.agents.agent import Agent, AgentExecutor, AgentOutputParser
from langchain.agents.agent_types import AgentType
from langchain.agents.react.output_parser import ReActOutputParser
from langchain.agents.react.textworld_prompt import TEXTWORLD_PROMPT
from langchain.agents.react.wiki_prompt import WIKI_PROMPT
from langchain.agents.utils import validate_tools_single_input
if TYPE_CHECKING:
from langchain_community.docstore.base import Docstore
[docs]@deprecated("0.1.0", removal="0.3.0")
class ReActDocstoreAgent(Agent):
"""ReAct链的代理。"""
output_parser: AgentOutputParser = Field(default_factory=ReActOutputParser)
@classmethod
def _get_default_output_parser(cls, **kwargs: Any) -> AgentOutputParser:
return ReActOutputParser()
@property
def _agent_type(self) -> str:
"""返回代理类型的标识符。"""
return AgentType.REACT_DOCSTORE
[docs] @classmethod
def create_prompt(cls, tools: Sequence[BaseTool]) -> BasePromptTemplate:
"""返回默认提示符。"""
return WIKI_PROMPT
@classmethod
def _validate_tools(cls, tools: Sequence[BaseTool]) -> None:
validate_tools_single_input(cls.__name__, tools)
super()._validate_tools(tools)
if len(tools) != 2:
raise ValueError(f"Exactly two tools must be specified, but got {tools}")
tool_names = {tool.name for tool in tools}
if tool_names != {"Lookup", "Search"}:
raise ValueError(
f"Tool names should be Lookup and Search, got {tool_names}"
)
@property
def observation_prefix(self) -> str:
"""要附加到观测值前面的前缀。"""
return "Observation: "
@property
def _stop(self) -> List[str]:
return ["\nObservation:"]
@property
def llm_prefix(self) -> str:
"""LLM调用前要附加的前缀。"""
return "Thought:"
[docs]@deprecated("0.1.0", removal="0.3.0")
class DocstoreExplorer:
"""用于辅助探索文档存储的类。"""
[docs] def __init__(self, docstore: Docstore):
"""使用文档存储初始化,并将初始文档设置为None。"""
self.docstore = docstore
self.document: Optional[Document] = None
self.lookup_str = ""
self.lookup_index = 0
[docs] def search(self, term: str) -> str:
"""在文档存储库中搜索一个术语,如果找到则保存。"""
result = self.docstore.search(term)
if isinstance(result, Document):
self.document = result
return self._summary
else:
self.document = None
return result
[docs] def lookup(self, term: str) -> str:
"""在文档中查找术语(如果已保存)。"""
if self.document is None:
raise ValueError("Cannot lookup without a successful search first")
if term.lower() != self.lookup_str:
self.lookup_str = term.lower()
self.lookup_index = 0
else:
self.lookup_index += 1
lookups = [p for p in self._paragraphs if self.lookup_str in p.lower()]
if len(lookups) == 0:
return "No Results"
elif self.lookup_index >= len(lookups):
return "No More Results"
else:
result_prefix = f"(Result {self.lookup_index + 1}/{len(lookups)})"
return f"{result_prefix} {lookups[self.lookup_index]}"
@property
def _summary(self) -> str:
return self._paragraphs[0]
@property
def _paragraphs(self) -> List[str]:
if self.document is None:
raise ValueError("Cannot get paragraphs without a document")
return self.document.page_content.split("\n\n")
[docs]@deprecated("0.1.0", removal="0.3.0")
class ReActTextWorldAgent(ReActDocstoreAgent):
"""ReAct TextWorld链的代理。"""
[docs] @classmethod
def create_prompt(cls, tools: Sequence[BaseTool]) -> BasePromptTemplate:
"""返回默认提示符。"""
return TEXTWORLD_PROMPT
@classmethod
def _validate_tools(cls, tools: Sequence[BaseTool]) -> None:
validate_tools_single_input(cls.__name__, tools)
super()._validate_tools(tools)
if len(tools) != 1:
raise ValueError(f"Exactly one tool must be specified, but got {tools}")
tool_names = {tool.name for tool in tools}
if tool_names != {"Play"}:
raise ValueError(f"Tool name should be Play, got {tool_names}")
[docs]@deprecated("0.1.0", removal="0.3.0")
class ReActChain(AgentExecutor):
"""[已弃用] 实现ReAct论文的链条。"""
def __init__(self, llm: BaseLanguageModel, docstore: Docstore, **kwargs: Any):
"""使用LLM和文档存储初始化。"""
docstore_explorer = DocstoreExplorer(docstore)
tools = [
Tool(
name="Search",
func=docstore_explorer.search,
description="Search for a term in the docstore.",
),
Tool(
name="Lookup",
func=docstore_explorer.lookup,
description="Lookup a term in the docstore.",
),
]
agent = ReActDocstoreAgent.from_llm_and_tools(llm, tools)
super().__init__(agent=agent, tools=tools, **kwargs)