mirror of
https://github.com/hwchase17/langchain
synced 2024-11-11 19:11:02 +00:00
af875cff57
Co-authored-by: Micky Liu <wayliu@microsoft.com> Co-authored-by: wayliums <wayliums@users.noreply.github.com> Co-authored-by: Erick Friis <erick@langchain.dev>
44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
from typing import Optional, Tuple, Union
|
|
|
|
from langchain.agents import AgentOutputParser
|
|
from langchain_core.agents import AgentAction, AgentFinish
|
|
|
|
|
|
def extract_action_details(text: str) -> Tuple[Optional[str], Optional[str]]:
|
|
# Split the text into lines and strip whitespace
|
|
lines = [line.strip() for line in text.strip().split("\n")]
|
|
|
|
# Initialize variables to hold the extracted values
|
|
action = None
|
|
action_input = None
|
|
|
|
# Iterate through the lines to find and extract the desired information
|
|
for line in lines:
|
|
if line.startswith("Action:"):
|
|
action = line.split(":", 1)[1].strip()
|
|
elif line.startswith("Action Input:"):
|
|
action_input = line.split(":", 1)[1].strip()
|
|
|
|
return action, action_input
|
|
|
|
|
|
class FakeOutputParser(AgentOutputParser):
|
|
def parse(self, text: str) -> Union[AgentAction, AgentFinish]:
|
|
print("FakeOutputParser", text)
|
|
action, input = extract_action_details(text)
|
|
|
|
if action:
|
|
log = f"\nInvoking: `{action}` with `{input}"
|
|
|
|
return AgentAction(tool=action, tool_input=(input or ""), log=log)
|
|
elif "Final Answer" in text:
|
|
return AgentFinish({"output": text}, text)
|
|
|
|
return AgentAction(
|
|
"Intermediate Answer", "after_colon", "Final Answer: This should end"
|
|
)
|
|
|
|
@property
|
|
def _type(self) -> str:
|
|
return "self_ask"
|