| import re |
| from json import JSONDecodeError |
| from typing import List, Union |
|
|
| from langchain_core.agents import AgentAction, AgentActionMessageLog, AgentFinish |
| from langchain_core.exceptions import OutputParserException |
| from langchain_core.messages import ( |
| AIMessage, |
| BaseMessage, |
| ) |
| from langchain_core.outputs import ChatGeneration, Generation |
|
|
| from langchain.agents.agent import AgentOutputParser |
| from agents.output_parsers.utils import parse_tool_call, check_tool_call |
| import ast |
|
|
| class FunctionsAgentOutputParser(AgentOutputParser): |
| """Parses a message into agent action/finish. |
| |
| Is meant to be used with a model with Nous Hermes 2 Pro as the base, as it relies on the specific |
| function_call parameter from Nous Research to convey what tools to use. |
| |
| If a function_call parameter is passed, then that is used to get |
| the tool and tool input. |
| |
| If one is not passed, then the AIMessage is assumed to be the final output. |
| It was add a |
| """ |
|
|
| @property |
| def _type(self) -> str: |
| return "functions-agent" |
|
|
| @staticmethod |
| def _parse_ai_message(message: BaseMessage): |
| """Parse an AI message.""" |
| if not isinstance(message, AIMessage): |
| raise TypeError(f"Expected an AI message got {type(message)}") |
|
|
| actions = [] |
|
|
| pattern = re.compile(r"<tool_call>(.*?)</tool_call>", re.DOTALL) |
| try: |
| tool_calls = [parse_tool_call(t.strip()) for t in pattern.findall(message.content)] |
| except: |
| raise OutputParserException( |
| f"Could not parse tool calls from message content: {message.content}. Please ensure that the tool calls are valid JSON." |
| ) |
|
|
| if not tool_calls: |
| return AgentFinish( |
| return_values={"output": message.content}, log=str(message.content) |
| ) |
|
|
| for tool_call in tool_calls: |
| tool_name, tool_input = check_tool_call(tool_call) |
| content_msg = f"\n{message.content}\n" if message.content else "\n" |
| log = f"\nInvoking: `{tool_name}` with `{tool_input}`\n{content_msg}\n" |
| actions.append(AgentActionMessageLog( |
| tool=tool_name, |
| tool_input=tool_input, |
| log=log, |
| message_log=[message], |
| )) |
| |
| return actions |
|
|
| def parse_result( |
| self, result: List[Generation], *, partial: bool = False |
| ) -> Union[AgentAction, AgentFinish]: |
| if not isinstance(result[0], ChatGeneration): |
| raise ValueError("This output parser only works on ChatGeneration output") |
| message = result[0].message |
| return self._parse_ai_message(message) |
|
|
| def parse(self, text: str) -> Union[AgentAction, AgentFinish]: |
| raise ValueError("Can only parse messages") |
