Source code for langchain.agents.chat.output_parser

import json
import re
from typing import Pattern, Union

from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.exceptions import OutputParserException

from langchain.agents.agent import AgentOutputParser
from langchain.agents.chat.prompt import FORMAT_INSTRUCTIONS

FINAL_ANSWER_ACTION = "Final Answer:"


[docs]class ChatOutputParser(AgentOutputParser): """Output parser for the chat agent.""" format_instructions: str = FORMAT_INSTRUCTIONS """Default formatting instructions""" pattern: Pattern = re.compile(r"^.*?`{3}(?:json)?\n(.*?)`{3}.*?$", re.DOTALL) """Regex pattern to parse the output."""
[docs] def get_format_instructions(self) -> str: """Returns formatting instructions for the given output parser.""" return self.format_instructions
[docs] def parse(self, text: str) -> Union[AgentAction, AgentFinish]: """Parse the output from the agent into an AgentAction or AgentFinish object. Args: text: The text to parse. Returns: An AgentAction or AgentFinish object. Raises: OutputParserException: If the output could not be parsed. ValueError: If the action could not be found. """ includes_answer = FINAL_ANSWER_ACTION in text try: found = self.pattern.search(text) if not found: # Fast fail to parse Final Answer. raise ValueError("action not found") action = found.group(1) response = json.loads(action.strip()) includes_action = "action" in response if includes_answer and includes_action: raise OutputParserException( "Parsing LLM output produced a final answer " f"and a parse-able action: {text}" ) return AgentAction( response["action"], response.get("action_input", {}), text ) except Exception as exc: if not includes_answer: raise OutputParserException( f"Could not parse LLM output: {text}" ) from exc output = text.split(FINAL_ANSWER_ACTION)[-1].strip() return AgentFinish({"output": output}, text)
@property def _type(self) -> str: return "chat"