| import re |
| from typing import Union |
|
|
| from langchain_core.agents import AgentAction, AgentFinish |
| from langchain_core.exceptions import OutputParserException |
|
|
| from langchain.agents.agent import AgentOutputParser |
|
|
|
|
| class ReActOutputParser(AgentOutputParser): |
| """Output parser for the ReAct agent.""" |
|
|
| def parse(self, text: str) -> Union[AgentAction, AgentFinish]: |
| action_prefix = "Action: " |
| if not text.strip().split("\n")[-1].startswith(action_prefix): |
| raise OutputParserException(f"Could not parse LLM Output: {text}") |
| action_block = text.strip().split("\n")[-1] |
|
|
| action_str = action_block[len(action_prefix) :] |
| |
| re_matches = re.search(r"(.*?)\[(.*?)\]", action_str) |
| if re_matches is None: |
| raise OutputParserException( |
| f"Could not parse action directive: {action_str}" |
| ) |
| action, action_input = re_matches.group(1), re_matches.group(2) |
| if action == "Finish": |
| return AgentFinish({"output": action_input}, text) |
| else: |
| return AgentAction(action, action_input, text) |
|
|
| @property |
| def _type(self) -> str: |
| return "react" |
|
|