当前位置：首页 > news >正文

Workflow

news 来源：原创 2025/5/29 10:15:10

一、为什么我们需要工作流？

❓ 什么样的职场打工人是合格的打工人？

反应快，理解能力强？
有相关经验，学过相关学科？
有行动力，不纸上谈兵，还要能下地干活？

❓ 哪些因素会影响大模型应用的效果？

模型能力（智力）
- 通识理解和泛化能力
- 输入信息理解、推理、规划、执行能力
- 输入信息补充知识学习能力
- 文字生成创作的风格
相关信息（知识）
- 与任务相关的信息
- 与互动背景相关的信息
模型输出控制（行动方法）
- 单次请求控制
  - Prompt表达优化
  - 以CoT为代表的思维链控制方法
  - 输出格式控制（文本格式语法、工程结构化数据输出…）
- 多次请求控制
  - 以ReAct（Action-Observation-Reflection）为代表的多轮自我反思优化
  - 复杂任务的执行过程编排管理

单次请求的局限性

上下文窗口长度限制、输出长度限制（早期的LangChain长文本Summarize）
直接进行CoT控制（尤其是用自然语言表达CoT）会输出思考过程，但我们不希望用户看到这个过程
随着工作进展出现的新信息，对任务时序、编排有依赖的信息，不一定能在单次请求中一次性完成输入

工作流的优势

将工作任务拆分成多个工作节点
能够将模型单次请求调用视作一个工作节点
能够灵活将其他代码逻辑也写入工作节点
能够对工作节点进行任务编排
能够在工作节点之间进行数据传递

[试一试]

直接请求模型的效果：

from ENV import deep_seek_url, deep_seek_api_key, deep_seek_default_model
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. **检查退货政策**：首先，查看您购买鞋子时商家提供的退货政策。通常，商家会在其网站上或随商品提供的文件中明确退货条件，包括退货期限、是否需要保持商品原包装和未使用状态等。2. **联系客服**：如果符合退货条件，您应该联系商家的客服部门。这可以通过电话、电子邮件或在线聊天完成。告知他们您希望退货，并询问需要提供哪些信息和文件。3. **准备退货物品**：按照商家的要求准备退货物品。通常，这包括将鞋子放回原包装中，并附上所有原始标签和收据。4. **填写退货表格**：有些商家可能要求您填写退货表格或提供退货授权号码（RMA）。5. **邮寄退货**：将退货物品邮寄回商家指定的地址。确保使用可追踪的邮寄服务，并保留邮寄凭证，以防需要跟踪退货或证明邮寄日期。6. **等待退款**：一旦商家收到退货物品并确认其符合退货条件，他们将处理退款。退款金额和方式（例如原支付方式退回或商店信用）将根据商家的政策而定。如果您是在实体店购买的鞋子，可以直接携带鞋子和收据到店内进行退货。店员将指导您完成退货流程。请记住，不同国家和地区的消费者保护法律可能有所不同，因此在退货过程中，您可能需要了解并遵守当地的法律规定。如果您遇到任何问题，可以考虑联系当地的消费者保护机构寻求帮助。

使用工作流：

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("END").else_condition().connect_to("generate_other_topic_reply").connect_to("@reply").connect_to("END")
)workflow.start()
pass

[请输入您的要求]:  我想要退货，这个鞋子不合脚
[回复]:  非常抱歉给您带来不便，我明白您想要退货因为鞋子不合脚。请稍等，我会立即查看如何为您处理退货事宜。
[回复]:  亲爱的客户，感谢您的耐心等待。
我理解您希望退货，是因为鞋子不合脚，您的心情一定非常失望。
请不要担心，我们会尽力帮助您解决这个问题。我们会尽快和相关人员沟通，并尽量进行满足。请留下您的联系方式以方便我们尽快处理后与您联系。

在这里插入图片描述

二、让我们从吴恩达博士的开源翻译工作流项目说起

项目地址：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
关键代码片段：

def one_chunk_initial_translation(source_lang: str, target_lang: str, source_text: str
) -> str:"""Translate the entire text as one chunk using an LLM.Args:source_lang (str): The source language of the text.target_lang (str): The target language for translation.source_text (str): The text to be translated.Returns:str: The translated text."""system_message = f"You are an expert linguist, specializing in translation from {source_lang} to {target_lang}."translation_prompt = f"""This is an {source_lang} to {target_lang} translation, please provide the {target_lang} translation for this text. \
Do not provide any explanations or text apart from the translation.
{source_lang}: {source_text}{target_lang}:"""prompt = translation_prompt.format(source_text=source_text)translation = get_completion(prompt, system_message=system_message)return translationdef one_chunk_reflect_on_translation(source_lang: str,target_lang: str,source_text: str,translation_1: str,country: str = "",
) -> str:"""Use an LLM to reflect on 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 of the translation.source_text (str): The original text in the source language.translation_1 (str): The initial translation of the source text.country (str): Country specified for target language.Returns:str: The LLM's reflection on the translation, providing constructive criticism and suggestions for improvement."""system_message = f"You are an expert linguist specializing in translation from {source_lang} to {target_lang}. \
You will be provided with a source text and its translation and your goal is to improve the translation."if country != "":reflection_prompt = f"""Your task is to carefully read a source text and a translation from {source_lang} to {target_lang}, and then give constructive criticism and helpful suggestions to improve the translation. \
The final style and tone of the translation should match the style of {target_lang} colloquially spoken in {country}.The source text and initial translation, delimited by XML tags <SOURCE_TEXT></SOURCE_TEXT> and <TRANSLATION></TRANSLATION>, are as follows:<SOURCE_TEXT>
{source_text}
</SOURCE_TEXT><TRANSLATION>
{translation_1}
</TRANSLATION>When writing suggestions, pay attention to whether there are ways to improve the translation's \n\
(i) accuracy (by correcting errors of addition, mistranslation, omission, or untranslated text),\n\
(ii) fluency (by applying {target_lang} grammar, spelling and punctuation rules, and ensuring there are no unnecessary repetitions),\n\
(iii) style (by ensuring the translations reflect the style of the source text and takes into account any cultural context),\n\
(iv) terminology (by ensuring terminology use is consistent and reflects the source text domain; and by only ensuring you use equivalent idioms {target_lang}).\n\Write a list of specific, helpful and constructive suggestions for improving the translation.
Each suggestion should address one specific part of the translation.
Output only the suggestions and nothing else."""else:reflection_prompt = f"""Your task is to carefully read a source text and a translation from {source_lang} to {target_lang}, and then give constructive criticism and helpful suggestions to improve the translation. \The source text and initial translation, delimited by XML tags <SOURCE_TEXT></SOURCE_TEXT> and <TRANSLATION></TRANSLATION>, are as follows:<SOURCE_TEXT>
{source_text}
</SOURCE_TEXT><TRANSLATION>
{translation_1}
</TRANSLATION>When writing suggestions, pay attention to whether there are ways to improve the translation's \n\
(i) accuracy (by correcting errors of addition, mistranslation, omission, or untranslated text),\n\
(ii) fluency (by applying {target_lang} grammar, spelling and punctuation rules, and ensuring there are no unnecessary repetitions),\n\
(iii) style (by ensuring the translations reflect the style of the source text and takes into account any cultural context),\n\
(iv) terminology (by ensuring terminology use is consistent and reflects the source text domain; and by only ensuring you use equivalent idioms {target_lang}).\n\Write a list of specific, helpful and constructive suggestions for improving the 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 reflectiondef 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