# 清理环境信息，与上课内容无关
import os
os.environ["LANGCHAIN_PROJECT"] = ""
os.environ["LANGCHAIN_API_KEY"] = ""
os.environ["LANGCHAIN_ENDPOINT"] = ""
os.environ["LANGCHAIN_TRACING_V2"] = ""# 安装所需要使用的包
!pip install openai langgraph Agently==3.3.4.5 mermaid-python nest_asyncio# 因为本课使用的langgraph可能需要依赖langchain 0.2.10版本，但其他课件依赖langchain 0.1.20版本
# 请学习完本课之后对langchain进行降级，以免在其他课程出现运行错误
#!pip install langchain==0.1.20
#!pip install langchain-openai==0.1.6
#!pip install langchain-community==0.0.38# 使用nest_asyncio确保异步稳定性
import nest_asyncio
nest_asyncio.apply()1. 了解工作流对大模型进行高质量工作的辅助意义
2. 学会复现吴恩达博士的翻译工作流开源项目
3. 了解构成大模型工作流系统的关键元素
4. 学会搭建一个更复杂的业务场景工作流工作流其实主要是,针对大模型的局限性来做一些补充的.
大模型在处理复杂工作,以及需要思维步骤工作的时候,是做的不好的,虽然可以做,但是结果不好,
所以这个时候,我们可以把要做的事情,进行拆解,拆解成一步步,并且把步骤进行编排.
让大模型以这种,工作流的形式去思考,去处理问题.这样会把复杂问题,更好的去解决.
10.213.5.175:9009代码能力要求：**中**，AI/数学基础要求：**低**
1. 有编程基础的同学 - 能够自己动手实现一套复杂的大模型工作流
2. 没有编程基础的同学 - 可以关注和理解工作流对于大模型应用的意义、关键元素和构建思路 - 不需要复杂编程知识，可以尝试复现简单的翻译工作流## 一、为什么我们需要工作流？
### ❓ 什么样的职场打工人是合格的打工人？
- 反应快，理解能力强？
- 有相关经验，学过相关学科？
- 有行动力，不纸上谈兵，还要能下地干活？
### ❓ 哪些因素会影响大模型应用的效果？1.以下几个点代表了模型的能力:
- 模型能力（智力）- 通识理解和泛化能力- 输入信息理解、推理、规划能力- 输入信息补充知识学习能力- 文字生成创作的风格2.对模型知识的补充:
- 相关信息（知识）- 与任务相关的信息- 与互动背景相关的信息3.让模型以更好处理问题的方法
- 模型输出控制（行动方法）- 单次请求控制- Prompt提示词表达优化- 以CoT为代表的思维链控制方法- 输出格式控制（文本格式语法、工程结构化数据输出…）- 多次请求控制- 以ReAct（Action-Observation-Reflection）为代表的多轮自我反思优化- 复杂任务的执行过程编排管理“Chain of Thought”（思维链），这是一种用于解决复杂问题或进行决策的推理方法。COT思维链是一种结构化的思考过程，它涉及将问题分解成一系列的逻辑步骤或环节，以便更清晰地理解问题并找到解决方案。以下是COT思维链的一些关键组成部分：问题定义：明确需要解决的问题或做出的决策是什么。
信息收集：搜集与问题相关的所有必要信息和数据。
假设建立：基于现有信息，提出可能的假设或解释。
逻辑推理：使用逻辑推理来分析每个假设，评估其合理性和可能的结果。
关键问题识别：确定哪些问题是解决整个问题的关键。
解决方案生成：基于推理，生成可能的解决方案或决策选项。
风险评估：评估每个解决方案或决策可能带来的风险和后果。
决策制定：选择最佳的解决方案或决策，并制定实施计划。
执行：实施决策，并监控结果。
反馈和调整：根据实施结果提供反馈，并根据需要调整解决方案或决策。
COT思维链在许多领域都有应用，包括军事规划、商业策略、项目管理、科学研究等。它帮助个体或团队以一种有条理和系统的方式思考，从而提高问题解决的效率和效果。通过COT思维链，可以确保在决策过程中考虑到所有相关因素，并遵循逻辑路径来达到最终目标。"多轮自我反思优化ReAct"实际上是一个涉及人工智能和机器学习领域的技术概念。它结合了ReAct和Reflexion两个框架，用于提升AI系统的决策能力和自我优化。
ReAct框架：ReAct是一种基于“思考-行动-观察”（TAO）循环的方法，源自于《ReAct: Synergizing Reasoning and Acting in Language Models》这篇论文。它通过结合语言模型中的推理和行动来解决多样化的语言推理和决策任务。ReAct的核心是TAO循环，包括以下几个步骤：
思考（Thought）：AI系统对问题进行深入思考，确定关键信息和推理步骤。
行动（Action）：AI系统根据思考结果采取行动，执行特定任务。
观察（Observation）：AI系统观察行动结果，检验其有效性，并据此调整策略。
Reflexion框架：Reflexion是一个通过语言反馈来强化基于语言的智能体的框架。它将环境的反馈（自由形式的语言或标量）转换为语言反馈，为智能体提供上下文，帮助其从之前的错误中学习，提升任务性能。Reflexion由三个不同的模型组成：
评估者（Evaluator）：对智能体的输出进行评价，输出奖励分数。
自我反思（Self-Reflection）：生成语言强化线索，帮助智能体自我完善。
记忆组件：存储智能体的经验和反馈，用于快速改进决策。
将这两个框架结合使用，可以实现多轮自我反思优化，即AI系统通过不断的“思考-行动-观察”循环，并结合自我反思和评估，从每次迭代中学习并优化其行为，以更好地解决决策、编程和推理等任务。然后再来看一下:
### 单次请求的局限性- 上下文窗口长度限制、输出长度限制（早期的LangChain长文本Summarize）:大模型的输出长度往往有限制.- 直接进行CoT控制（尤其是用自然语言表达CoT）会输出思考过程，但我们不希望用户看到这个过程大模型使用思维链思考的时候,会把思考过程打印出来.- 随着工作进展出现的新信息，对任务时序、编排有依赖的信息，不一定能在单次请求中一次性完成输入如果使用思维链,思考过程比较长,那么如果不能一次输出,可能会导致以前的信息遗忘.
### 工作流的优势- 将工作任务拆分成多个工作节点
- 能够将模型单次请求调用视作一个工作节点
- 能够灵活将其他代码逻辑也写入工作节点
- 能够对工作节点进行任务编排
- 能够在工作节点之间进行数据传递可以看到工作流很重要就是因为他创造了一种工作方式,来弥补,大模型无法无限输出内容的短板.
让大模型可以从整体,细节上把握如何来处理问题.的关键.
首先让我们体验一下,使用工作流和不使用工作流来处理问题的区别:
[试一试]
直接请求模型的效果：
首先我们有个ENV.py这个文件
import os
deep_seek_url = os.environ["DEEPSEEK_BASE_URL"]
deep_seek_api_key = os.environ["DEEPSEEK_API_KEY"]
deep_seek_default_model = 'deepseek-chat'加载了对应的参数.#然后使用ENV,导入了这个参数
from ENV import deep_seek_url, deep_seek_api_key, deep_seek_default_model
#上一节使用了智能体Agently下一个博文会补充.,去创建一个智能体.
import Agently
agent = (Agently.create_agent().set_settings("current_model", "OAIClient").set_settings("model.OAIClient.url", deep_seek_url).set_settings("model.OAIClient.auth", { "api_key": deep_seek_api_key }).set_settings("model.OAIClient.options", { "model": deep_seek_default_model })
)result = agent.input(input("[请输入您的要求]: ")).start()
print("[回复]: ", result)[请输入您的要求]: 我想要退货，这个鞋子不合脚
[回复]: 如果您购买的鞋子不合脚，您可以按照以下步骤尝试退货：1. **检查退货政策**：首先，查看您购买时的退货政策。大多数零售商都有一定的退货期限，通常是购买后的30天内。确保您的退货请求在期限内。2. **保留收据和包装**：保留您的购买收据和鞋子的原始包装。这些通常是退货时必需的。3. **联系客服**：通过电话、电子邮件或在线聊天联系零售商的客服部门。说明您的情况，并询问退货流程。4. **准备退货物品**：按照零售商的要求准备退货物品。可能需要您将鞋子放回原始包装中，并附上收据。5. **退货方式**：根据零售商的指示，您可能需要亲自到店铺退货，或者通过邮寄的方式退货。如果是邮寄，确保使用可追踪的邮寄服务，并保留邮寄凭证。6. **等待处理**：一旦零售商收到您的退货物品，他们将进行检查并处理退款。退款通常会退回到您原支付方式中。7. **跟进**：如果在合理的时间内没有收到退款，及时跟进并联系客服询问退款状态。可以看到上面就是我们没有使用大模型工作流的情况下,大模型的一个回答.
然后下面是使用大模型工作流的情况下,大模型的一个回答:
请注意，不同的零售商可能有不同的退货政策和流程，所以最好是直接咨询您购买鞋子的
零售商以获取最准确的信息。
如果是在线购买，通常可以在网站的“帮助”或“客户服务”部分找到退货指南。
使用工作流：下面是一个实现代码,可以后面再具体分析:
#这里也是用了Agently这个框架.可以看到
workflow = Agently.Workflow()#第一个chunk是输入用户需求,存储用户的需求
@workflow.chunk()
def user_input(inputs, storage):storage.set("user_input", input("[请输入您的要求]: "))return#第二步,来判断用户的意图,并且把意图判断,得到结果以后,存储了结果.
#存储了quick_reply结果,然后返回了user_intent判断出来的意图
@workflow.chunk()
def judge_intent_and_quick_reply(inputs, storage):result = (agent.input(storage.get("user_input")).output({"user_intent": ("闲聊 | 售后问题 | 其他", "判断用户提交的{input}内容属于给定选项中的哪一种"),"quick_reply": ("str",
"""如果{user_intent}=='闲聊'，那么直接给出你的回应；
如果{user_intent}=='售后问题'，那么请用合适的方式告诉用户你已经明白用户的诉求，安抚客户情绪并请稍等你去看看应该如何处理；
如果{user_intent}=='其他'，此项输出null""")}).start())storage.set("reply", result["quick_reply"])return result["user_intent"]#然后第三个步骤:
#根据第二步给出的quick_reply这种回答问题的方式,来进行进一步回答.
@workflow.chunk()
def generate_after_sales_reply(inputs, storage):storage.set("reply", (agent.input(storage.get("user_input")).instruct(
"""请根据{input}的要求，以一个专业客户服务人员的角色给出回复，遵循如下模板进行回复：
亲爱的客户，感谢您的耐心等待。
我理解您希望{{复述客户的要求}}，是因为{{复述客户要求提出要求的理由}}，您的心情一定非常{{阐述你对客户心情/感受的理解}}。
{{给出对客户当前心情的抚慰性话语}}。
我们会尽快和相关人员沟通，并尽量进行满足。请留下您的联系方式以方便我们尽快处理后与您联系。
"""
).start()))return#这个方法也是一个步骤,用来判断如果用户说的是其他问题,提示用户应该说售后问题.
@workflow.chunk()
def generate_other_topic_reply(inputs, storage):storage.set("reply", "我们好像不应该聊这个，还是回到您的问题或诉求上来吧。")return#定义reply方法,chunk_class,然后输入inputs,和storage,然后
#获取replay的内容,任何返回,根据replay做具体的返回.
@workflow.chunk_class()
def reply(inputs, storage):print("[回复]: ", storage.get("reply"))return(workflow.connect_to("user_input").connect_to("judge_intent_and_quick_reply")#首先进行了输入用户需求,然后判断用户意图,然后根据意图,进行判断#根据不同意图调用不同方法,进行回答..if_condition(lambda return_value, storage: return_value=="闲聊").connect_to("@reply").connect_to("end").elif_condition(lambda return_value, storage: return_value=="售后问题")#第一个reply回复用户第一个reply非常抱歉给您带来不便，#我明白您想要退货因为鞋子不合脚。请稍等，我会立即查看如何为您处理退货事宜。.connect_to("@reply").connect_to("generate_after_sales_reply")#第二个reply是执行generate_after_sales_reply以后生成的reply.connect_to("@reply")#回复完了,然后再连接用户的输入,让用户继续输入,对话继续进行..connect_to("user_input").else_condition().connect_to("generate_other_topic_reply").connect_to("@reply").connect_to("END")
)workflow.start()
pass然后我们看一下结果:
[请输入您的要求]: 我想要退货，这个鞋子不合脚
[回复]: 非常抱歉给您带来不便，我明白您想要退货因为鞋子不合脚。请稍等，我会立即查看如何为您处理退货事宜。
[回复]: 亲爱的客户，感谢您的耐心等待。
我理解您希望退货，是因为鞋子不合脚，您的心情一定非常失望。
请不要担心，我们会尽力帮助您解决这个问题。我们会尽快和相关人员沟通，并尽量进行满足。请留下您的联系方式以方便我们尽快处理后与您联系。然后我们再来看一下完整的没有注释的代码:
workflow = Agently.Workflow()@workflow.chunk()
def user_input(inputs, storage):storage.set("user_input", input("[请输入您的要求]: "))return@workflow.chunk()
def judge_intent_and_quick_reply(inputs, storage):result = (agent.input(storage.get("user_input")).output({"user_intent": ("闲聊 | 售后问题 | 其他", "判断用户提交的{input}内容属于给定选项中的哪一种"),"quick_reply": ("str",
"""如果{user_intent}=='闲聊'，那么直接给出你的回应；
如果{user_intent}=='售后问题'，那么请用合适的方式告诉用户你已经明白用户的诉求，安抚客户情绪并请稍等你去看看应该如何处理；
如果{user_intent}=='其他'，此项输出null""")}).start())storage.set("reply", result["quick_reply"])return result["user_intent"]@workflow.chunk()
def generate_after_sales_reply(inputs, storage):storage.set("reply", (agent.input(storage.get("user_input")).instruct(
"""请根据{input}的要求，以一个专业客户服务人员的角色给出回复，遵循如下模板进行回复：
亲爱的客户，感谢您的耐心等待。
我理解您希望{{复述客户的要求}}，是因为{{复述客户要求提出要求的理由}}，您的心情一定非常{{阐述你对客户心情/感受的理解}}。
{{给出对客户当前心情的抚慰性话语}}。
我们会尽快和相关人员沟通，并尽量进行满足。请留下您的联系方式以方便我们尽快处理后与您联系。
"""
).start()))return@workflow.chunk()
def generate_other_topic_reply(inputs, storage):storage.set("reply", "我们好像不应该聊这个，还是回到您的问题或诉求上来吧。")return@workflow.chunk_class()
def reply(inputs, storage):print("[回复]: ", storage.get("reply"))return(workflow.connect_to("user_input").connect_to("judge_intent_and_quick_reply").if_condition(lambda return_value, storage: return_value=="闲聊").connect_to("@reply").connect_to("end").elif_condition(lambda return_value, storage: return_value=="售后问题").connect_to("@reply").connect_to("generate_after_sales_reply").connect_to("@reply").connect_to("user_input").else_condition().connect_to("generate_other_topic_reply").connect_to("@reply").connect_to("END")
)workflow.start()
pass然后我们来看一下这个过程是怎么做的?
其实就是用下面的思维工作流实现的.
其实就是下面的实现方式.
具体说其实就是下面这个图中的过程.
然后我们再来看一个大模型工作流的开源项目:
- 项目地址：[https://github.com/andrewyng/translation-agent](https://github.com/andrewyng/translation-agent)- 项目基本思路：- 让模型在完成首轮翻译之后，通过自我反思后修正的工作流优化翻译结果，以提升最终文本翻译的质量- 关键步骤：1. 第一步：- 输入信息：**原始文本语言(source_lang)** 、**翻译目标语言(target_lang)** 和 **原始文本(source_text)**- 角色设定：以翻译文本为任务目标的语言学家- 输出结果：基于所有输入信息，对 **原始文本(source_text)** 进行 **第一轮翻译的结果(translation_1)**；2. 第二步：- 输入信息：**原始文本语言(source_lang)** 、**翻译目标语言(target_lang)** 、 **原始文本(source_text)** 和 **第一轮翻译结果(translation_1)**- 角色设定：以阅读原始文本和翻译文本，并给出翻译改进意见为任务目标的语言学家- 输出结果：基于所有输入信息，对 **第一轮翻译结果(translation_1)** 提出的 **改进意见反思(reflection)**3. 第三步：- 输入信息：**原始文本语言(source_lang)** 、**翻译目标语言(target_lang)** 、 **原始文本(source_text)** 、 **第一轮翻译结果(translation_1)** 和 **改进意见反思(reflection)**- 角色设定：以翻译文本为任务目标的语言学家（和第一步相同）- 输出结果：基于所有输入信息，给出的**第二轮优化后翻译结果(translation_2)**- 关键代码文件：[https://github.com/andrewyng/translation-agent/blob/main/src/translation_agent/utils.py](https://github.com/andrewyng/translation-agent/blob/main/src/translation_agent/utils.py)- 关键代码片段：这个项目是吴恩达博士的开源翻译工作流项目
- 项目地址：[https://github.com/andrewyng/translation-agent](https://github.com/andrewyng/translation-agent)
项目地址在上面.
具体的可以去查看,项目中的代码,实现逻辑就是上面讲的.
分成三步:
1.给出,带翻译的文本语言,以及要翻译成的目标语言,然后给出带翻译文本,先让大模型进行翻译.得到第一次翻译的文本translation_1
2.给出,带翻译的文本语言,以及要翻译成的目标语言,然后给出带翻译文本,然后给出大模型第一个翻译的文本,并让大模型进行提出翻译的优化建议.
3.将,带翻译的文本语言,以及要翻译成的目标语言,然后给出带翻译文本,然后给出大模型第一个翻译的文本,并给出大模型进行提出翻译的优化建议,让大模型进行优化,再把优化后的结果给出最后结果.
#1.可以看到第一个方法.给出了source_lang带翻译文本的语言,target_lang翻译成目标文本的语言
#source_text以及这个带翻译文本.并且将结果定义为translation_1
#并且返回了结果
def one_chunk_initial_translation(
Each suggestion should address one specific part of the translation.
Output only the suggestions and nothing else."""prompt = reflection_prompt.format(source_lang=source_lang,target_lang=target_lang,source_text=source_text,translation_1=translation_1,)reflection = get_completion(prompt, system_message=system_message)return reflection#第一次翻译后的结果会提供给,第二次给大模型提出建议的提示词中.
def one_chunk_improve_translation(source_lang: str,target_lang: str,source_text: str,translation_1: str,reflection: str,
) -> str:"""Use the reflection to improve the translation, treating the entire text as one chunk.Args:source_lang (str): The source language of the text.target_lang (str): The target language for the translation.source_text (str): The original text in the source language.translation_1 (str): The initial translation of the source text.reflection (str): Expert suggestions and constructive criticism for improving the translation.Returns:str: The improved translation based on the expert suggestions."""system_message = f"You are an expert linguist, specializing in translation editing from {source_lang} to {target_lang}."prompt = f"""Your task is to carefully read, then edit, a translation from {source_lang} to {target_lang}, taking into
account a list of expert suggestions and constructive criticisms.The source text, the initial translation, and the expert linguist suggestions are delimited by XML tags <SOURCE_TEXT></SOURCE_TEXT>, <TRANSLATION></TRANSLATION> and <EXPERT_SUGGESTIONS></EXPERT_SUGGESTIONS> \
as follows:<SOURCE_TEXT>
{source_text}
</SOURCE_TEXT><TRANSLATION>
{translation_1}
</TRANSLATION><EXPERT_SUGGESTIONS>
{reflection}
</EXPERT_SUGGESTIONS>Please take into account the expert suggestions when editing the translation. Edit the translation by ensuring:(i) accuracy (by correcting errors of addition, mistranslation, omission, or untranslated text),
(ii) fluency (by applying {target_lang} grammar, spelling and punctuation rules and ensuring there are no unnecessary repetitions), \
(iii) style (by ensuring the translations reflect the style of the source text)
(iv) terminology (inappropriate for context, inconsistent use), or
(v) other errors.Output only the new translation and nothing else."""translation_2 = get_completion(prompt, system_message)return translation_2def one_chunk_translate_text(source_lang: str, target_lang: str, source_text: str, country: str = ""
) -> str:"""Translate a single chunk of text from the source language to the target language.This function performs a two-step translation process:1. Get an initial translation of the source text.2. Reflect on the initial translation and generate an improved translation.Args:source_lang (str): The source language of the text.target_lang (str): The target language for the translation.source_text (str): The text to be translated.country (str): Country specified for target language.Returns:str: The improved translation of the source text."""translation_1 = one_chunk_initial_translation(source_lang, target_lang, source_text)reflection = one_chunk_reflect_on_translation(source_lang, target_lang, source_text, translation_1, country)translation_2 = one_chunk_improve_translation(source_lang, target_lang, source_text, translation_1, reflection)return translation_2下面是一套添加了注释的,比较清楚一些的,便于理解的代码版本:
def one_chunk_initial_translation(source_lang: str,target_lang: str,source_text: str,translation_1: str,
) -> str:"""生成对文本块初步翻译的反思。参数:source_lang (str): 文本被翻译的源语言。target_lang (str): 文本被翻译的目标语言。source_text (str): 源语言中的原始文本。translation_1 (str): 将源文本翻译成目标语言的第一次尝试。返回:str: 对初步翻译的反思，其中包括改进的专家建议。"""# 使用给定的参数格式化反射提示prompt = reflection_prompt.format(source_lang=source_lang,target_lang=target_lang,source_text=source_text,translation_1=translation_1,)# 使用提示和系统消息调用外部函数以获取反射reflection = get_completion(prompt, system_message=system_message)# 返回反射，用于改进翻译return reflection
def one_chunk_improve_translation(source_lang: str,target_lang: str,source_text: str,translation_1: str,reflection: str,
) -> str:"""根据提供的反思中的专家建议改进翻译。参数:source_lang (str): 文本的源语言。target_lang (str): 翻译的目标语言。source_text (str): 源语言中的原始文本。translation_1 (str): 源文本的初步翻译。reflection (str): 用于改进翻译的专家建议和建设性批评。返回:str: 根据专家建议改进后的翻译。"""# 设置系统消息，指明是一个精通从源语言到目标语言翻译编辑的专家system_message = f"你是一个精通从{source_lang}到{target_lang}翻译编辑的专家。"# 构建提示，包含源文本、初步翻译和专家建议prompt = f"""你的任务是仔细阅读，然后编辑从{source_lang}到{target_lang}的翻译，
考虑一系列专家建议和建设性批评。
源文本、初步翻译和专家建议分别用XML标签<SOURCE_TEXT></SOURCE_TEXT>、
<TRANSLATION></TRANSLATION>和<EXPERT_SUGGESTIONS></EXPERT_SUGGESTIONS>分隔，如下所示：
<SOURCE_TEXT>
{source_text}
</SOURCE_TEXT>
<TRANSLATION>
{translation_1}
</TRANSLATION>
<EXPERT_SUGGESTIONS>
{reflection}
</EXPERT_SUGGESTIONS>
请在编辑翻译时考虑专家建议。编辑翻译时要确保：
(i) 准确性（通过纠正添加、误译、遗漏或未翻译文本的错误），
(ii) 流畅性（应用{target_lang}的语法、拼写和标点规则，并确保没有不必要的重复），
(iii) 风格（确保翻译反映源文本的风格），
(iv) 术语（不适当的上下文，使用不一致），
(v) 其他错误。
只输出新的翻译，不要输出其他内容。"""# 获取根据提示和系统消息改进后的翻译translation_2 = get_completion(prompt, system_message)# 返回改进后的翻译return translation_2具体的过程还可以需实际的看代码,但是代码不是很复杂,可以看明白.
## 三、使用LangGraph和Agently Workflow分别复现这个工作流
另外还出现了一些用于工作流开发的框架.这些大模型工作流开发框架有助于,大模型工作流的开发.
