介绍

尽管科技在不断进步，但当今世界仍然存在语言障碍。无论是在工作中还是在户外，语言的差异总会造成尴尬的局面。对于团队分布在不同地区、使用不同语言的大型企业来说，尤其如此。该项目旨在通过为工作场所开发多语言代理聊天应用程序，解决语言障碍和其他与工作场所相关的低效问题。

高级框架

鉴于 LangChain 及其基于图形的对应物 LangGraph 的流行，我不希望这是另一个解释这些包及其方法的基础知识的教程。相反，我想更多地关注通过这些包实现我们的解决方案时所面临的设计选择和挑战，因为我觉得从长远来看这会更有用。

LangChain 与 LangGraph

我们面临的第一个设计选择是在 LangChain 和 LangGraph 之间做出选择。

在一个简单的场景中（如下图所示），用户提供的每条信息都会发送给所有其他用户，并翻译成他们喜欢的语言，那么选择 LangChain 就足够了。这是一个单向流，从用户发送信息开始，到用户接收信息结束：

然而，在我们的方案中，最主要的限制因素是要加入一个人工智能助手，我们将其称为 Aya（以 “远征 ”命名）。根据计划，Aya 将成为这款聊天应用的重要组成部分，并为我们的系统增加了一层新的复杂性。有了 Aya，发送用户的信息需要进行分析，根据信息的性质（如果是给 Aya 的命令），系统需要发送回一条信息，然后再发送给接收用户。

定义运行： 与此相关的另一个设计选择是对信息传递周期的一次 “运行 ”或一次 “迭代 ”的定义。

在我们选择的定义中，我们认为每次运行都是由任何用户发送信息开始的，当所有与初始信息相关的信息都到达接收用户时，运行就结束了。

因此，如果一条信息并不针对 Aya，而只是直接发给其他用户的信息，那么当所有用户都收到初始翻译信息时，运行就被认为终止了。而如果信息是针对 Aya 的，那么当所有用户都收到初始信息和 Aya 的回复时，运行才算结束。

因此，根据运行的这种设计选择/定义，我们需要一种流程，在这种流程中，我们等待来自 Aya 的响应生成并推送给用户，然后再终止运行。为了实现这样的流程，我们使用了 LangGraph，因为它就是专门为解决这种情况而设计的。

构建代理

该应用程序的支柱是代理及其交互。总的来说，我们有两种不同类型的代理：

1. 用户代理： 附属于每个用户的代理。主要任务是将接收到的信息翻译成用户喜欢的语言
2. Aya 代理： 与 Aya 相关的各种代理，每个代理都有自己的特定角色/工作

用户代理

UserAgent 类用于定义与聊天室每个用户相关联的代理。UserAgent 类实现的部分功能如下：

1. 将收到的信息翻译成用户喜欢的语言

2. 当用户发送信息时激活/调用图形

3. 保留聊天历史记录，帮助提供翻译任务的上下文，实现 “上下文感知 ”翻译

class UserAgent(object):
    def __init__(self, llm, userid, user_language):
        self.llm = llm
        self.userid = userid
        self.user_language = user_language
        self.chat_history = []
        prompt = ChatPromptTemplate.from_template(USER_SYSTEM_PROMPT2)
        self.chain = prompt | llm

    def set_graph(self, graph):
        self.graph = graph
    def send_text(self,text:str, debug = False):
        message = ChatMessage(message = HumanMessage(content=text), sender = self.userid)
        inputs = {"messages": [message]}
        output = self.graph.invoke(inputs, debug = debug)
        return output
    def display_chat_history(self, content_only = False):
        for i in self.chat_history:
            if content_only == True:
                print(f"{i.sender} : {i.content}")
            else:
                print(i)
    
    def invoke(self, message:BaseMessage) -> AIMessage:
        
        output = self.chain.invoke({'message':message.content, 'user_language':self.user_language})
        return output

在大多数情况下，UserAgent 的实现是非常标准的 LangChain/LangGraph 代码：

• 定义一个 LangChain 链（一个提示模板 + LLM），负责进行实际翻译。
  
• 定义一个 send_text 函数，用于在用户发送新信息时调用图形。

在大多数情况下，该代理的性能取决于 LLM 的翻译质量，因为翻译是该代理的主要目标。而 LLM 的翻译性能会有很大差异，尤其取决于所涉及的语言。某些低资源语言在某些模型的训练数据中没有很好的代表性，这确实会影响这些语言的翻译质量。

Aya 代理

对于 Aya，我们实际上有一个由不同代理组成的系统，这些代理都为整个助手做出了贡献。具体来说，我们有

1. AyaSupervisor ：控制代理，负责监督其他 Aya 代理的运行。
2. AyaQuery：运行基于 RAG 的问题解答的代理
3. AyaSummarizer：用于生成聊天摘要和进行任务识别的代理
4. AyaTranslator： 将信息翻译成英语的代理

class AyaTranslator(object):
    def __init__(self, llm) -> None:
        self.llm = llm 
        prompt = ChatPromptTemplate.from_template(AYA_TRANSLATE_PROMPT)
        self.chain = prompt | llm 
        
    def invoke (self, message: str) -> AIMessage:
        output = self.chain.invoke({'message':message})
        return output
class AyaQuery(object):
    def __init__(self, llm, store, retriever) -> None:
        self.llm = llm
        self.retriever = retriever
        self.store = store
        qa_prompt = ChatPromptTemplate.from_template(AYA_AGENT_PROMPT)
        self.chain = qa_prompt | llm
    def invoke(self, question : str) -> AIMessage:
        context = format_docs(self.retriever.invoke(question))
        rag_output = self.chain.invoke({'question':question, 'context':context})
        return rag_output
class AyaSupervisor(object):
    def __init__(self, llm):
        
        prompt = ChatPromptTemplate.from_template(AYA_SUPERVISOR_PROMPT)
        self.chain = prompt | llm
    def invoke(self, message : str) -> str:
        output = self.chain.invoke(message)
        return output.content
class AyaSummarizer(object):
    def __init__(self, llm):
        
        message_length_prompt = ChatPromptTemplate.from_template(AYA_SUMMARIZE_LENGTH_PROMPT)
        self.length_chain = message_length_prompt | llm 
        
        prompt = ChatPromptTemplate.from_template(AYA_SUMMARIZER_PROMPT)
        self.chain = prompt | llm


    def invoke(self, message : str, agent : UserAgent) -> str:
        length = self.length_chain.invoke(message)
        try:
            length = int(length.content.strip())
        except:
            length = 0
        chat_history = agent.chat_history
        if length == 0:
            messages_to_summarize = [chat_history[i].content for i in range(len(chat_history))]
        else:
            messages_to_summarize = [chat_history[i].content for i in range(min(len(chat_history), length))]
        
        print(length)
        print(messages_to_summarize)
        messages_to_summarize = "\n ".join(messages_to_summarize)
        
        output = self.chain.invoke(messages_to_summarize)
        output_content = output.content 
        print(output_content)
        return output_content

这些代理大多具有类似的结构，主要包括一个由自定义提示符和 LLM 组成的 LangChain 链。但 AyaQuery 代理和 AyaSummarizer 代理除外，前者有一个额外的矢量数据库检索器来实现 RAG，后者有多个 LLM 功能。

设计考虑因素

AyaSupervisor 代理的作用： 在图的设计中，我们有一条固定的边从监督员节点通向用户节点。这意味着所有到达监督员节点的信息都会被推送到用户节点本身。因此，在 Aya 被处理的情况下，我们必须确保从 Aya 向用户推送的只有一个最终输出。我们不希望中间信息（如果有的话）到达用户。因此，我们设立了 AyaSupervisor 代理，作为 Aya 代理的单一联络点。该代理主要负责解释传入信息的意图，将信息发送给相应的特定任务代理，然后输出最终信息与用户共享。

AyaSummarizer 的设计： 与其他 Aya 代理相比，AyaSummarizer 代理稍显复杂，因为它需要执行两个步骤。第一步，代理首先确定需要汇总的信息数量，这是一个 LLM 调用，有自己的提示。在第二步中，一旦我们知道了需要汇总的信息数量，我们就会整理所需的信息，并将其传递给 LLM 以生成实际的摘要。除了摘要，在这一步中，LLM 还会识别信息中的任何行动项目，并将其单独列出。

因此，大致上有三项任务：确定要汇总的信息长度、汇总信息、识别行动项目。不过，鉴于第一项任务在没有任何明确示例的情况下对 LLM 来说有点困难，我选择将其作为一个单独的 LLM 调用，然后将后两项任务合并为各自的 LLM 调用。

也许可以取消额外的 LLM 调用，将所有三个任务合并为一个调用。可能的方案包括：

1. 提供非常详细的示例，一步涵盖所有三项任务
2. 生成大量示例，对 LLM 进行实际微调，使其能够很好地完成这项任务

AyaTranslator 的作用： Aya 的目标之一是让它成为一个多语言的人工智能助手，能够用用户喜欢的语言进行交流。然而，要在 Aya 代理内部处理不同的语言是很困难的。具体来说，如果 Aya 代理的提示语言是英语，而用户信息使用的是另一种语言，就有可能产生问题。因此，为了避免这种情况，作为过滤步骤，我们将所有传入 Aya 的用户信息翻译成英文。因此，Aya 代理群的所有内部工作都是用英语完成的，包括输出。我们不必将 Aya 的输出翻译回原始语言，因为当信息到达用户时，用户代理会将信息翻译成各自指定的语言。

提示设计

在提示设计方面，大部分工作都集中在如何让 LLM 以特定格式输出一致的回复。在大多数情况下，我可以通过提供明确的说明来实现这一目标。在某些情况下，仅有说明是不够的，我还必须提供示例，使代理的行为保持一致。

在大多数情况下，提示模板的结构如下： 

[High level task definition] You are an AI assistant that answers user's questions... 
[List of specific constraints related to the response]
Obey the following rules : 
1. ....
[Providing context/user input]
Message : 

举个具体的例子，我们来看看用户代理使用的提示符：

You are a {user_language} translator, translating a conversation between work colleagues. Translate the message provided by the user into {user_language}. 
Obey the following rules : 
1. Only translate the text thats written after 'Message:' and nothing else
2. If the text is already in {user_language} then return the message as it is.
3. Return only the translated text
4. Ensure that your translation uses formal language
Message:
{message}

关于这个代理，一个重要的约束条件是确保模型只输出翻译文本，而不输出类似 “这里是翻译文本 ”或 “当然，以下是所提供文本的翻译 ”这样的辅助文本。在这种情况下，添加一条需要遵守的特定规则（第 3 条规则）就足以确保模型只输出翻译文本，而不输出其他内容。

需要在提示中举例说明的一个例子是与摘要代理相关的提示。特别是负责确定要总结的信息数量的代理。我发现很难让代理始终如一地提取所列出的信息数量（如果有的话），并以特定格式输出。因此，有必要提供示例，以便更好地解释我所期望的代理响应。

其他实现细节

聊天信息

熟悉 LangChain 的人应该已经知道 AIMessage 和 HumanMessage 类，它们用于保存人工智能和人类的消息。对于我们的用例，我们需要能够存储发送者的 ID，以便下游任务使用。因此，为了解决这个问题，我们创建了一个名为 ChatMessage 的新派生类，用于存储消息和发件人 ID

class ChatMessage(object):
    def __init__(self, message : BaseMessage, sender : str = None):
        self.message = message
        self.sender = sender
        self.content = message.content
    def __repr__(self) -> str:
        return f"{self.sender} | {self.content}"

图形状态

在 LangGraph 中，图形的关键元素之一是图形状态。状态变量/对象对于代理之间的正常通信以及跟踪图工作流的进度至关重要。

def reducer(a : list, b : list | str ) -> list:
    if type(b) == list: 
        return a + b
    else:
        return a
    
class AgentState(TypedDict):
    messages: Annotated[Sequence[ChatMessage], reducer]

在大多数 LangGraph 示例中，状态变量都是一个字符串列表，在通过每个代理后会被不断追加。在我们的用例中，我希望排除某些节点的输出对图状态的影响，尽管工作流已经通过了该节点。为了适应这种情况，我区分了两种类型的状态变化，一种是列表，另一种是字符串。如果状态更新是以列表的形式出现，它就会被附加到整个状态对象中。如果状态更新是字符串，我们会忽略该更新并传播现有状态。这可以通过上面定义的自定义减速器函数来实现。

结论

到此为止，我们已经介绍了代理工作流的关键组件之一：代理的设计选择。