Source code for langchain.agents.structured_chat.output_parser
from __future__ import annotations
import json
import logging
import re
from typing import Optional, Union
from langchain.agents.agent import AgentOutputParser
from langchain.agents.structured_chat.prompt import FORMAT_INSTRUCTIONS
from langchain.output_parsers import OutputFixingParser
from langchain.pydantic_v1 import Field
from langchain.schema import AgentAction, AgentFinish, OutputParserException
from langchain.schema.language_model import BaseLanguageModel
logger = logging.getLogger(__name__)
[docs]class StructuredChatOutputParser(AgentOutputParser):
    """Output parser for the structured chat agent."""
    pattern = re.compile(r"```(?:json)?\n(.*?)```", re.DOTALL)
[docs]    def get_format_instructions(self) -> str:
        return FORMAT_INSTRUCTIONS
[docs]    def parse(self, text: str) -> Union[AgentAction, AgentFinish]:
        try:
            action_match = self.pattern.search(text)
            if action_match is not None:
                response = json.loads(action_match.group(1).strip(), strict=False)
                if isinstance(response, list):
                    # gpt turbo frequently ignores the directive to emit a single action
                    logger.warning("Got multiple action responses: %s", response)
                    response = response[0]
                if response["action"] == "Final Answer":
                    return AgentFinish({"output": response["action_input"]}, text)
                else:
                    return AgentAction(
                        response["action"], response.get("action_input", {}), text
                    )
            else:
                return AgentFinish({"output": text}, text)
        except Exception as e:
            raise OutputParserException(f"Could not parse LLM output: {text}") from e
    @property