Source code for langchain.agents.self_ask_with_search.base

"""链式结构，执行自我询问并进行搜索。"""
from __future__ import annotations

from typing import TYPE_CHECKING, Any, Sequence, Union

from langchain_core._api import deprecated
from langchain_core.language_models import BaseLanguageModel
from langchain_core.prompts import BasePromptTemplate
from langchain_core.pydantic_v1 import Field
from langchain_core.runnables import Runnable, RunnablePassthrough
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.format_scratchpad import format_log_to_str
from langchain.agents.self_ask_with_search.output_parser import SelfAskOutputParser
from langchain.agents.self_ask_with_search.prompt import PROMPT
from langchain.agents.utils import validate_tools_single_input

if TYPE_CHECKING:
    from langchain_community.utilities.google_serper import GoogleSerperAPIWrapper
    from langchain_community.utilities.searchapi import SearchApiAPIWrapper
    from langchain_community.utilities.serpapi import SerpAPIWrapper


@deprecated("0.1.0", alternative="create_self_ask_with_search", removal="0.3.0")
class SelfAskWithSearchAgent(Agent):
    """自问自答搜索论文的代理。"""

    output_parser: AgentOutputParser = Field(default_factory=SelfAskOutputParser)

    @classmethod
    def _get_default_output_parser(cls, **kwargs: Any) -> AgentOutputParser:
        return SelfAskOutputParser()

    @property
    def _agent_type(self) -> str:
        """返回代理类型的标识符。"""
        return AgentType.SELF_ASK_WITH_SEARCH

    @classmethod
    def create_prompt(cls, tools: Sequence[BaseTool]) -> BasePromptTemplate:
        """提示不依赖于工具。"""
        return 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 != {"Intermediate Answer"}:
            raise ValueError(
                f"Tool name should be Intermediate Answer, got {tool_names}"
            )

    @property
    def observation_prefix(self) -> str:
        """要附加到观测值前面的前缀。"""
        return "Intermediate answer: "

    @property
    def llm_prefix(self) -> str:
        """LLM调用前要添加的前缀。"""
        return ""


@deprecated("0.1.0", removal="0.3.0")
class SelfAskWithSearchChain(AgentExecutor):
    """[已弃用] 执行自我询问并进行搜索的链。"""

    def __init__(
        self,
        llm: BaseLanguageModel,
        search_chain: Union[
            GoogleSerperAPIWrapper, SearchApiAPIWrapper, SerpAPIWrapper
        ],
        **kwargs: Any,
    ):
        """仅使用LLM和搜索链进行初始化。"""
        search_tool = Tool(
            name="Intermediate Answer",
            func=search_chain.run,
            coroutine=search_chain.arun,
            description="Search",
        )
        agent = SelfAskWithSearchAgent.from_llm_and_tools(llm, [search_tool])
        super().__init__(agent=agent, tools=[search_tool], **kwargs)


def create_self_ask_with_search_agent(
    llm: BaseLanguageModel, tools: Sequence[BaseTool], prompt: BasePromptTemplate
) -> Runnable:
    """创建一个使用自我询问和搜索提示的代理。

参数：
    llm：用作代理的LLM。
    tools：工具列表。应该只有一个工具，该工具的名称为`Intermediate Answer`。
    prompt：要使用的提示，必须具有输入键`agent_scratchpad`，其中将包含代理操作和工具输出。

返回：
    代表代理的可运行序列。它接受与传入的提示相同的所有输入变量。它返回一个AgentAction或AgentFinish。

示例：

    .. code-block:: python

        from langchain import hub
        from langchain_community.chat_models import ChatAnthropic
        from langchain.agents import (
            AgentExecutor, create_self_ask_with_search_agent
        )

        prompt = hub.pull("hwchase17/self-ask-with-search")
        model = ChatAnthropic(model="claude-3-haiku-20240307")
        tools = [...]  # 应该只有一个名称为`Intermediate Answer`的工具

        agent = create_self_ask_with_search_agent(model, tools, prompt)
        agent_executor = AgentExecutor(agent=agent, tools=tools)

        agent_executor.invoke({"input": "hi"})

提示：

    提示必须具有输入键`agent_scratchpad`，其中将包含代理操作和工具输出的字符串。

    这是一个示例：

    .. code-block:: python

        from langchain_core.prompts import PromptTemplate

        template = '''问题：谁活得更久，穆罕默德·阿里还是艾伦·图灵？
        这里是否需要后续问题：是的。
        后续问题：穆罕默德·阿里去世时多大年纪？
        中间答案：穆罕默德·阿里去世时74岁。
        后续问题：艾伦·图灵去世时多大年纪？
        中间答案：艾伦·图灵去世时41岁。
        因此最终答案是：穆罕默德·阿里

        问题：craigslist的创始人是什么时候出生的？
        这里是否需要后续问题：是的。
        后续问题：craigslist的创始人是谁？
        中间答案：craigslist的创始人是Craig Newmark。
        后续问题：Craig Newmark是什么时候出生的？
        中间答案：Craig Newmark于1952年12月6日出生。
        因此最终答案是：1952年12月6日

        问题：乔治·华盛顿的外祖父是谁？
        这里是否需要后续问题：是的。
        后续问题：乔治·华盛顿的母亲是谁？
        中间答案：乔治·华盛顿的母亲是玛丽·波尔·华盛顿。
        后续问题：玛丽·波尔·华盛顿的父亲是谁？
        中间答案：玛丽·波尔·华盛顿的父亲是约瑟夫·波尔。
        因此最终答案是：约瑟夫·波尔

        问题：《大白鲨》和《皇家赌场》的导演都来自同一个国家吗？
        这里是否需要后续问题：是的。
        后续问题：《大白鲨》的导演是谁？
        中间答案：《大白鲨》的导演是史蒂文·斯皮尔伯格。
        后续问题：史蒂文·斯皮尔伯格来自哪里？
        中间答案：美国。
        后续问题：《皇家赌场》的导演是谁？
        中间答案：《皇家赌场》的导演是马丁·坎贝尔。
        后续问题：马丁·坎贝尔来自哪里？
        中间答案：新西兰。
        因此最终答案是：不是

        问题：{input}
        这里是否需要后续问题：{agent_scratchpad}'''

        prompt = PromptTemplate.from_template(template)
"""  # noqa: E501
    missing_vars = {"agent_scratchpad"}.difference(
        prompt.input_variables + list(prompt.partial_variables)
    )
    if missing_vars:
        raise ValueError(f"Prompt missing required variables: {missing_vars}")

    if len(tools) != 1:
        raise ValueError("This agent expects exactly one tool")
    tool = list(tools)[0]
    if tool.name != "Intermediate Answer":
        raise ValueError(
            "This agent expects the tool to be named `Intermediate Answer`"
        )

    llm_with_stop = llm.bind(stop=["\nIntermediate answer:"])
    agent = (
        RunnablePassthrough.assign(
            agent_scratchpad=lambda x: format_log_to_str(
                x["intermediate_steps"],
                observation_prefix="\nIntermediate answer: ",
                llm_prefix="",
            ),
            # Give it a default
            chat_history=lambda x: x.get("chat_history", ""),
        )
        | prompt
        | llm_with_stop
        | SelfAskOutputParser()
    )
    return agent