from langchain.chains import SimpleChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.callbacks import CallbackManager
from langchain.callbacks.stdout import StdOutCallbackHandler
from langchain.memory import SimpleMemory
# 定义预处理函数,用于清理输入文本
def preprocess_text(text):
return text.strip().lower()
# 定义后处理函数,用于处理模型输出
def postprocess_text(text):
return text.capitalize()
# 定义一个简单的错误处理函数
def error_handler(e):
print(f"Error encountered: {e}")
return "Sorry, an error occurred."
# 定义一个提示模板
prompt_template = PromptTemplate(
input_variables=["input_text"],
template="将以下文本翻译成中文: {input_text}"
)
# 创建一个OpenAI模型实例(假设你已经配置了API密钥)
llm = OpenAI(model="text-davinci-003")
# 使用CallbackManager来管理回调
callback_manager = CallbackManager([StdOutCallbackHandler()])
# 使用SimpleMemory来管理会话记忆
memory = SimpleMemory()
# 创建一个完善的链式处理流程
class EnhancedChain(SimpleChain):
def __init__(self, prompt_template, llm, preprocess=None, postprocess=None, on_error=None):
super().__init__(prompt_template=prompt_template, llm=llm)
self.preprocess = preprocess
self.postprocess = postprocess
self.on_error = on_error
def run(self, inputs):
try:
# 预处理
if self.preprocess:
inputs['input_text'] = self.preprocess(inputs['input_text'])
# 执行链式处理
output = super().run(inputs)
# 后处理
if self.postprocess:
output = self.postprocess(output)
return output
except Exception as e:
if self.on_error:
return self.on_error(e)
raise e
# 实例化增强的链式处理流程
enhanced_chain = EnhancedChain(
prompt_template=prompt_template,
llm=llm,
preprocess=preprocess_text,
postprocess=postprocess_text,
on_error=error_handler
)
# 测试链,传入示例文本
input_text = " Hello, how are you? "
result = enhanced_chain.run({"input_text": input_text})
# 输出结果
print(result)