|
|
import re |
|
|
from typing import Any, List, Optional, Sequence, Tuple, Union |
|
|
|
|
|
from langchain_core._api import deprecated |
|
|
from langchain_core.agents import AgentAction |
|
|
from langchain_core.callbacks import BaseCallbackManager |
|
|
from langchain_core.language_models import BaseLanguageModel |
|
|
from langchain_core.prompts import BasePromptTemplate |
|
|
from langchain_core.prompts.chat import ( |
|
|
ChatPromptTemplate, |
|
|
HumanMessagePromptTemplate, |
|
|
SystemMessagePromptTemplate, |
|
|
) |
|
|
from langchain_core.runnables import Runnable, RunnablePassthrough |
|
|
from langchain_core.tools import BaseTool |
|
|
from langchain_core.tools.render import ToolsRenderer |
|
|
from pydantic import Field |
|
|
|
|
|
from langchain.agents.agent import Agent, AgentOutputParser |
|
|
from langchain.agents.format_scratchpad import format_log_to_str |
|
|
from langchain.agents.output_parsers import JSONAgentOutputParser |
|
|
from langchain.agents.structured_chat.output_parser import ( |
|
|
StructuredChatOutputParserWithRetries, |
|
|
) |
|
|
from langchain.agents.structured_chat.prompt import FORMAT_INSTRUCTIONS, PREFIX, SUFFIX |
|
|
from langchain.chains.llm import LLMChain |
|
|
from langchain.tools.render import render_text_description_and_args |
|
|
|
|
|
HUMAN_MESSAGE_TEMPLATE = "{input}\n\n{agent_scratchpad}" |
|
|
|
|
|
|
|
|
@deprecated("0.1.0", alternative="create_structured_chat_agent", removal="1.0") |
|
|
class StructuredChatAgent(Agent): |
|
|
"""Structured Chat Agent.""" |
|
|
|
|
|
output_parser: AgentOutputParser = Field( |
|
|
default_factory=StructuredChatOutputParserWithRetries |
|
|
) |
|
|
"""Output parser for the agent.""" |
|
|
|
|
|
@property |
|
|
def observation_prefix(self) -> str: |
|
|
"""Prefix to append the observation with.""" |
|
|
return "Observation: " |
|
|
|
|
|
@property |
|
|
def llm_prefix(self) -> str: |
|
|
"""Prefix to append the llm call with.""" |
|
|
return "Thought:" |
|
|
|
|
|
def _construct_scratchpad( |
|
|
self, intermediate_steps: List[Tuple[AgentAction, str]] |
|
|
) -> str: |
|
|
agent_scratchpad = super()._construct_scratchpad(intermediate_steps) |
|
|
if not isinstance(agent_scratchpad, str): |
|
|
raise ValueError("agent_scratchpad should be of type string.") |
|
|
if agent_scratchpad: |
|
|
return ( |
|
|
f"This was your previous work " |
|
|
f"(but I haven't seen any of it! I only see what " |
|
|
f"you return as final answer):\n{agent_scratchpad}" |
|
|
) |
|
|
else: |
|
|
return agent_scratchpad |
|
|
|
|
|
@classmethod |
|
|
def _validate_tools(cls, tools: Sequence[BaseTool]) -> None: |
|
|
pass |
|
|
|
|
|
@classmethod |
|
|
def _get_default_output_parser( |
|
|
cls, llm: Optional[BaseLanguageModel] = None, **kwargs: Any |
|
|
) -> AgentOutputParser: |
|
|
return StructuredChatOutputParserWithRetries.from_llm(llm=llm) |
|
|
|
|
|
@property |
|
|
def _stop(self) -> List[str]: |
|
|
return ["Observation:"] |
|
|
|
|
|
@classmethod |
|
|
def create_prompt( |
|
|
cls, |
|
|
tools: Sequence[BaseTool], |
|
|
prefix: str = PREFIX, |
|
|
suffix: str = SUFFIX, |
|
|
human_message_template: str = HUMAN_MESSAGE_TEMPLATE, |
|
|
format_instructions: str = FORMAT_INSTRUCTIONS, |
|
|
input_variables: Optional[List[str]] = None, |
|
|
memory_prompts: Optional[List[BasePromptTemplate]] = None, |
|
|
) -> BasePromptTemplate: |
|
|
tool_strings = [] |
|
|
for tool in tools: |
|
|
args_schema = re.sub("}", "}}", re.sub("{", "{{", str(tool.args))) |
|
|
tool_strings.append(f"{tool.name}: {tool.description}, args: {args_schema}") |
|
|
formatted_tools = "\n".join(tool_strings) |
|
|
tool_names = ", ".join([tool.name for tool in tools]) |
|
|
format_instructions = format_instructions.format(tool_names=tool_names) |
|
|
template = "\n\n".join([prefix, formatted_tools, format_instructions, suffix]) |
|
|
if input_variables is None: |
|
|
input_variables = ["input", "agent_scratchpad"] |
|
|
_memory_prompts = memory_prompts or [] |
|
|
messages = [ |
|
|
SystemMessagePromptTemplate.from_template(template), |
|
|
*_memory_prompts, |
|
|
HumanMessagePromptTemplate.from_template(human_message_template), |
|
|
] |
|
|
return ChatPromptTemplate(input_variables=input_variables, messages=messages) |
|
|
|
|
|
@classmethod |
|
|
def from_llm_and_tools( |
|
|
cls, |
|
|
llm: BaseLanguageModel, |
|
|
tools: Sequence[BaseTool], |
|
|
callback_manager: Optional[BaseCallbackManager] = None, |
|
|
output_parser: Optional[AgentOutputParser] = None, |
|
|
prefix: str = PREFIX, |
|
|
suffix: str = SUFFIX, |
|
|
human_message_template: str = HUMAN_MESSAGE_TEMPLATE, |
|
|
format_instructions: str = FORMAT_INSTRUCTIONS, |
|
|
input_variables: Optional[List[str]] = None, |
|
|
memory_prompts: Optional[List[BasePromptTemplate]] = None, |
|
|
**kwargs: Any, |
|
|
) -> Agent: |
|
|
"""Construct an agent from an LLM and tools.""" |
|
|
cls._validate_tools(tools) |
|
|
prompt = cls.create_prompt( |
|
|
tools, |
|
|
prefix=prefix, |
|
|
suffix=suffix, |
|
|
human_message_template=human_message_template, |
|
|
format_instructions=format_instructions, |
|
|
input_variables=input_variables, |
|
|
memory_prompts=memory_prompts, |
|
|
) |
|
|
llm_chain = LLMChain( |
|
|
llm=llm, |
|
|
prompt=prompt, |
|
|
callback_manager=callback_manager, |
|
|
) |
|
|
tool_names = [tool.name for tool in tools] |
|
|
_output_parser = output_parser or cls._get_default_output_parser(llm=llm) |
|
|
return cls( |
|
|
llm_chain=llm_chain, |
|
|
allowed_tools=tool_names, |
|
|
output_parser=_output_parser, |
|
|
**kwargs, |
|
|
) |
|
|
|
|
|
@property |
|
|
def _agent_type(self) -> str: |
|
|
raise ValueError |
|
|
|
|
|
|
|
|
def create_structured_chat_agent( |
|
|
llm: BaseLanguageModel, |
|
|
tools: Sequence[BaseTool], |
|
|
prompt: ChatPromptTemplate, |
|
|
tools_renderer: ToolsRenderer = render_text_description_and_args, |
|
|
*, |
|
|
stop_sequence: Union[bool, List[str]] = True, |
|
|
) -> Runnable: |
|
|
"""Create an agent aimed at supporting tools with multiple inputs. |
|
|
|
|
|
Args: |
|
|
llm: LLM to use as the agent. |
|
|
tools: Tools this agent has access to. |
|
|
prompt: The prompt to use. See Prompt section below for more. |
|
|
stop_sequence: bool or list of str. |
|
|
If True, adds a stop token of "Observation:" to avoid hallucinates. |
|
|
If False, does not add a stop token. |
|
|
If a list of str, uses the provided list as the stop tokens. |
|
|
|
|
|
Default is True. You may to set this to False if the LLM you are using |
|
|
does not support stop sequences. |
|
|
tools_renderer: This controls how the tools are converted into a string and |
|
|
then passed into the LLM. Default is `render_text_description`. |
|
|
|
|
|
Returns: |
|
|
A Runnable sequence representing an agent. It takes as input all the same input |
|
|
variables as the prompt passed in does. It returns as output either an |
|
|
AgentAction or AgentFinish. |
|
|
|
|
|
Examples: |
|
|
|
|
|
.. code-block:: python |
|
|
|
|
|
from langchain import hub |
|
|
from langchain_community.chat_models import ChatOpenAI |
|
|
from langchain.agents import AgentExecutor, create_structured_chat_agent |
|
|
|
|
|
prompt = hub.pull("hwchase17/structured-chat-agent") |
|
|
model = ChatOpenAI() |
|
|
tools = ... |
|
|
|
|
|
agent = create_structured_chat_agent(model, tools, prompt) |
|
|
agent_executor = AgentExecutor(agent=agent, tools=tools) |
|
|
|
|
|
agent_executor.invoke({"input": "hi"}) |
|
|
|
|
|
# Using with chat history |
|
|
from langchain_core.messages import AIMessage, HumanMessage |
|
|
agent_executor.invoke( |
|
|
{ |
|
|
"input": "what's my name?", |
|
|
"chat_history": [ |
|
|
HumanMessage(content="hi! my name is bob"), |
|
|
AIMessage(content="Hello Bob! How can I assist you today?"), |
|
|
], |
|
|
} |
|
|
) |
|
|
|
|
|
Prompt: |
|
|
|
|
|
The prompt must have input keys: |
|
|
* `tools`: contains descriptions and arguments for each tool. |
|
|
* `tool_names`: contains all tool names. |
|
|
* `agent_scratchpad`: contains previous agent actions and tool outputs as a string. |
|
|
|
|
|
Here's an example: |
|
|
|
|
|
.. code-block:: python |
|
|
|
|
|
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder |
|
|
|
|
|
system = '''Respond to the human as helpfully and accurately as possible. You have access to the following tools: |
|
|
|
|
|
{tools} |
|
|
|
|
|
Use a json blob to specify a tool by providing an action key (tool name) and an action_input key (tool input). |
|
|
|
|
|
Valid "action" values: "Final Answer" or {tool_names} |
|
|
|
|
|
Provide only ONE action per $JSON_BLOB, as shown: |
|
|
|
|
|
``` |
|
|
{{ |
|
|
"action": $TOOL_NAME, |
|
|
"action_input": $INPUT |
|
|
}} |
|
|
``` |
|
|
|
|
|
Follow this format: |
|
|
|
|
|
Question: input question to answer |
|
|
Thought: consider previous and subsequent steps |
|
|
Action: |
|
|
``` |
|
|
$JSON_BLOB |
|
|
``` |
|
|
Observation: action result |
|
|
... (repeat Thought/Action/Observation N times) |
|
|
Thought: I know what to respond |
|
|
Action: |
|
|
``` |
|
|
{{ |
|
|
"action": "Final Answer", |
|
|
"action_input": "Final response to human" |
|
|
}} |
|
|
|
|
|
Begin! Reminder to ALWAYS respond with a valid json blob of a single action. Use tools if necessary. Respond directly if appropriate. Format is Action:```$JSON_BLOB```then Observation''' |
|
|
|
|
|
human = '''{input} |
|
|
|
|
|
{agent_scratchpad} |
|
|
|
|
|
(reminder to respond in a JSON blob no matter what)''' |
|
|
|
|
|
prompt = ChatPromptTemplate.from_messages( |
|
|
[ |
|
|
("system", system), |
|
|
MessagesPlaceholder("chat_history", optional=True), |
|
|
("human", human), |
|
|
] |
|
|
) |
|
|
""" |
|
|
missing_vars = {"tools", "tool_names", "agent_scratchpad"}.difference( |
|
|
prompt.input_variables + list(prompt.partial_variables) |
|
|
) |
|
|
if missing_vars: |
|
|
raise ValueError(f"Prompt missing required variables: {missing_vars}") |
|
|
|
|
|
prompt = prompt.partial( |
|
|
tools=tools_renderer(list(tools)), |
|
|
tool_names=", ".join([t.name for t in tools]), |
|
|
) |
|
|
if stop_sequence: |
|
|
stop = ["\nObservation"] if stop_sequence is True else stop_sequence |
|
|
llm_with_stop = llm.bind(stop=stop) |
|
|
else: |
|
|
llm_with_stop = llm |
|
|
|
|
|
agent = ( |
|
|
RunnablePassthrough.assign( |
|
|
agent_scratchpad=lambda x: format_log_to_str(x["intermediate_steps"]), |
|
|
) |
|
|
| prompt |
|
|
| llm_with_stop |
|
|
| JSONAgentOutputParser() |
|
|
) |
|
|
return agent |
|
|
|
