Langchain ConversationChain自定义Output Parser验证错误求助

问题描述 #

使用Langchain的ConversationChain开发带会话记忆的聊天机器人时，AI响应末尾会出现换行和乱码。自定义解析器移除乱码后，触发ValidationError，提示信息如下：

ValidationError: 1 validation error for ConversationChain
output_parser
value is not a valid dict (type=type_error.dict)

相关代码：

from langchain.chains.conversation.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

class MyOutputParser:
    def __init__(self):
        pass

    def parse(self, output):
        cut_off = output.find("\n", 3)
        # 删除换行后的所有内容
        return output[:cut_off]

template = """You will answer the following questions the best you can, being as informative and factual as possible.
If you don't know, say you don't know. 

Current conversation:
{history}
Human: {input}
AI Assistant:"""

the_output_parser=MyOutputParser()
print(type(the_output_parser))

PROMPT = PromptTemplate(input_variables=["history", "input"], template=template)
conversation = ConversationChain(
    prompt=PROMPT,
    llm=local_llm,
    memory=ConversationBufferWindowMemory(k=4),
    return_final_only=True,
    verbose=False,
    output_parser=the_output_parser,
)

问题原因 #

Langchain的ConversationChain通过Pydantic校验参数合法性，自定义的输出解析器必须继承Langchain提供的BaseOutputParser基类，而非自定义普通类。普通类无法通过Pydantic的序列化校验，因此触发"不是有效字典"的错误。

解决方案 #

方案1：继承BaseOutputParser实现自定义解析器 #

修改自定义解析器，继承BaseOutputParser并实现必要的方法：

from langchain.chains.conversation.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory
from langchain.schema import BaseOutputParser  # 导入基类

class MyOutputParser(BaseOutputParser):  # 继承BaseOutputParser
    def parse(self, output: str) -> str:
        cut_off = output.find("\n", 3)
        # 处理未找到换行的情况，避免返回空字符串
        return output[:cut_off] if cut_off != -1 else output
    
    def _get_format_instructions(self) -> str:
        # 必须实现的方法，返回空字符串即可（无需给LLM格式提示）
        return ""

template = """You will answer the following questions the best you can, being as informative and factual as possible.
If you don't know, say you don't know. 

Current conversation:
{history}
Human: {input}
AI Assistant:"""

the_output_parser = MyOutputParser()

PROMPT = PromptTemplate(input_variables=["history", "input"], template=template)
conversation = ConversationChain(
    prompt=PROMPT,
    llm=local_llm,
    memory=ConversationBufferWindowMemory(k=4),
    return_final_only=True,
    verbose=False,
    output_parser=the_output_parser,
)

方案2：直接处理预测结果（更简便） #

如果不需要将解析逻辑绑定到Chain中，可以在获取响应后直接处理：

# 调用聊天机器人获取响应
response = conversation.predict(input="你的问题内容")
# 清理换行后的乱码
cut_off = response.find("\n", 3)
cleaned_response = response[:cut_off] if cut_off != -1 else response

内容的提问来源于stack exchange，提问作者Z S