| | 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" |
| |
|