| | 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:" |
| |
|
| |
|
| | 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.""" |
| |
|
| | def get_format_instructions(self) -> str: |
| | """Returns formatting instructions for the given output parser.""" |
| | return self.format_instructions |
| |
|
| | 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: |
| | |
| | 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" |
| |
|
