拯救多代理系统！从根本上加速高层次设计创建？

使用多智能体系统生成高层设计 — 通过智能工作流加速 HLD 创建

第 5 部分，共 9 部分 — 构建人工智能驱动的软件开发生命周期 — 实用指南

合著者 Rishi Arora

来源：使用 DALL-E 生成

要设计一个稳健的软件系统，全面的高层设计是必不可少的。可以将其视为一个蓝图，概述了各种组件如何交互、数据如何在它们之间流动以及实现它们的技术。然而，传统的高层设计创建方法耗时、容易出错，并且常常导致利益相关者之间的沟通不畅。

本文探讨了如何利用多智能体系统，借助如 LangGraph 这样的框架，在我们之前的文章中建立的人工智能驱动的软件开发生命周期框架内自动化高层设计的创建，从而显著提高效率和协作。我们将演示这些系统如何促进自动化，弥合在第 4 部分中定义的需求与软件之间的差距。

如往常一样，如果您想跳过理论，可以直接跳到实现部分.

为什么考虑自动化高层设计的创建？ — 挑战

创建高层设计（HLD）的传统过程涉及几个关键任务：

1. 系统上下文：定义系统的边界及其与外部系统或利益相关者的交互。
2. 应用概述：概述应用程序的核心功能和特点，提供其目的的一般描述。
3. 组件规格：详细说明每个组件的职责、输入/输出和约束条件。
4. 数据架构：描述数据的存储、访问和管理方式，包括数据库选择。
5. 数据流：可视化数据在组件和外部系统之间的流动方式。
6. 技术栈：选择适当的技术、平台和工具进行实施。
7. 集成点：定义系统与第三方服务或现有基础设施的集成。
8. 部署架构：指定物理或云基础设施的设置。
9. 安全考虑：概述访问控制、数据加密和身份验证措施。
10. 可扩展性与性能：解决系统的可扩展性和满足性能要求的能力。

这些任务通常是：

• 耗时且容易出错。
• 缺乏灵活性以应对变更。
• 受制于手动修订，导致不一致性。

自动化与LangGraph如何提升高层设计创建

自动化，借助像LangGraph这样的工具，改善了高层设计创建过程，通过简化工作流程、减少错误和优化协作来实现。以下是这种变革性方法的工作原理：

主要好处

1. 简化工作流程

   • 自动化减少手动任务。
   • 促进更快的项目交付。

2. 减少错误

   • 最小化文档中的人为错误。
   • 确保高层设计的一致性。

3. 优化协作

   • 增强团队沟通。
   • 提供一个集中反馈的平台。

工作原理

def create_hld(project_name, requirements):
    hld = {}
    hld['project_name'] = project_name
    hld['requirements'] = requirements
    return hld

结论

自动化和像LangGraph这样的工具显著提升了高层设计创建过程，使其更加高效和可靠。

自动化的关键优势

• 加速设计过程: AI agents 迅速生成草稿和修订，促进更快的迭代，并缩短项目时间线。
• 确保一致性: 结构化的方法确保组件和交互的统一定义，最小化设计中的错误和差异。
• 实现快速修订: 人工智能驱动的反馈循环允许高效地纳入变更，使设计与不断变化的需求保持一致。
• 最小化人工干预: 通过自动化重复任务，人工智能使用户和利益相关者能够专注于关键决策和最终审查，减少手动工作。

高层设计中可以自动化的任务

• 架构推荐: AI agents 分析过去的项目或预定义模板，以建议最佳系统架构。
• 组件分解: 根据需求自动生成每个组件功能、输入和输出的详细描述。
• 数据流映射: Agents 创建与定义的系统组件一致的准确数据流图。
• 技术推荐: 提供针对每个组件的最合适技术的量身定制建议，考虑约束和架构。
• 集成规范: 自动记录与第三方系统或现有基础设施的集成点。

LangGraph 如何帮助这个过程

LangGraph 充当高层设计自动化的协调者，通过促进人工智能代理之间的无缝协作。这些代理高效地管理任务，如收集需求、与用户互动和起草设计，从而实现并行执行和更快的输出。例如：

1. 需求分析：一个代理分析软件需求规格说明书 (SRS) 并与利益相关者互动以获取澄清。
2. 高层设计草稿：另一个代理通过参考预定义的模板/格式生成高层设计，并创建数据流图或部署架构。
3. 审查与迭代：一个审查代理根据其他代理和利益相关者的反馈来完善草稿。

自动化高层设计 — 设计团队的角色

HLD生成的代理工作流

代理工作流用于自动化HLD生成

设计思维

• 代理分析软件需求规格说明书 (SRS)，寻求澄清，并与用户和利益相关者验证他们的理解。

生成 HLD 的初始版本

• 定义高层设计结构 (例如，系统上下文、组件详细信息、数据架构) 基于组织要求。
• 生成草稿使用人工智能驱动的软件开发生命周期代理，结合 SRS 和预定义模板的见解。

迭代修订

• 初始草稿经过审阅代理的迭代评审，以优化设计并使其与用户期望对齐。

最终审查与批准

• 用户和利益相关者提供反馈并以最小的手动努力批准设计。人工智能代理通过提供详细的见解和建议来促进这一过程。

自动化在高层设计创建中的好处

提高速度

在高层设计（HLD）的创建中，自动化显著加快了整个设计过程。它带来了更快的反馈循环，使团队能够快速迭代设计，从而实现更快的决策和更迅速的利益相关者批准。

一致性

在高层设计（HLD）创建中，主要挑战之一是确保不同团队和设计迭代之间的一致性。自动化确保所有文档遵循标准化的模板、格式和设计结构。这种一致性有助于避免由于风格、格式或术语的变化而导致的错误。

增强协作

通过将自动化工具集成到设计过程中，团队成员之间的协作变得无缝。代理或自动化系统可以实现实时反馈，确保任何更新或修订能够即时传达。团队可以实时跟踪设计的演变，使得讨论变更、提出新想法和迅速解决问题变得更加容易。这导致了一个更加透明和动态的工作流程，在这个流程中，贡献得到了认可，并在设计的背景下进行了调整。

提高生产力

自动化在优化高层设计（HLD）创建过程的各个阶段中发挥着关键作用，从而显著提升整体生产力。通过自动化诸如设计思维（通过人工智能辅助的头脑风暴）、文档（通过文本的自动生成）和审查（通过自动验证检查或同行评审工作流）等耗时任务，团队可以将更多时间用于高价值活动，例如战略制定、创意设计调整或客户会议。

风险与缓解措施

虽然自动化在高层设计创建中提供了诸多好处，但特定风险包括：

• 不准确的规格：自动化工具可能误解需求，导致不正确的高层设计。
• 缺乏上下文：自动化可能忽视特定的业务上下文或细微差别，这对于有效设计至关重要。
• 对工具的依赖：过度依赖自动化工具可能导致对基础架构缺乏理解。
• 集成挑战：自动化设计可能无法与现有系统或流程无缝集成。

缓解策略

为了解决这些风险，请考虑以下缓解策略：

1. 定期审查：定期审查自动化输出，以确保准确性和相关性。
2. 人工监督：邀请经验丰富的架构师监督自动化流程并提供上下文。
3. 培训：确保团队成员在工具和架构方面接受培训，以保持自动化与人类专业知识之间的平衡。
4. 迭代反馈：实施迭代反馈循环，根据实际使用情况持续改进自动化过程。

结论

在高层设计创建中，自动化可以显著提高效率，但必须保持警惕，关注潜在风险，并实施有效的缓解策略，以确保成功的结果。

不准确的系统上下文或组件规格

自动化工具可能会生成不准确的系统上下文图或组件规格，如果它们误解了模糊的需求。例如，自动化工具可能会忽视对设计至关重要的特定约束或环境因素。

为防止这种情况，系统应纳入来自用户和利益相关者的反馈循环，以在最终确定设计之前完善这些方面。确保人工智能模型在多样化、高质量的数据集上进行训练，也将最大限度地减少这些关键领域出错的可能性。

复杂设计的过度简化

虽然自动化可以加速设计生成，但它可能会通过未能考虑所有边缘情况或组件之间复杂的依赖关系而过度简化复杂系统。为了减轻这一问题，人类审核者应验证设计，确保在自动化过程中互动的复杂性，例如数据流或安全考虑，没有丢失。融合自动化建议与专家见解的混合方法可以帮助保持设计的深度。

代理错误或偏见

代理可能容易出现错误或偏见，因此监控它们的性能并在代理产生不正确或偏见的输出时进行干预是很重要的。

现在让我们深入实现并动手实践。

使用 LangGraph 自动化 HLD 生成的实践

安装库，确保您已安装与 LangGraph 和 LangChain 相关的必要工具：

pip install --upgrade pip
pip install -U langgraph langchain_openai

配置 API 密钥 设置您的 OpenAI API 密钥以访问 LLM 服务：

import os
os.environ["OPENAI_API_KEY"] = "your_openai_key"

配置文件系统 用于读取/写入输入/输出文件：

from typing import List

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, AIMessage, HumanMessage, ToolMessage

from pydantic import BaseModel, Field
from tempfile import TemporaryDirectory
from langchain_community.agent_toolkits import FileManagementToolkit
file_stores = FileManagementToolkit(
    selected_tools=["read_file", "write_file", "list_directory"]
).get_tools()

初始化 ChatOpenAI 模型：

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="<choose the model>")

识别数据点 用于架构设计并将其绑定到 LLM：

from pydantic import BaseModel

class HighLevelDesign(BaseModel):
    """Instructions on how to prompt the LLM."""
    sector: str
    drivers: str
    functionalities: str
    non_functional_requirement: str
    channels: str
    application_type: str
    cloud: str
    architecture_style: str
    technology_choice: str
    architecture_decisions: str
    others: str

llm = ChatOpenAI(model=model_name, temperature=0, max_retries=1)
llm_with_tool = llm.bind_tools([HighLevelDesign])

定义代理以从用户收集澄清和确认：

LAPrompt = """ Define prompt to gather identified data points"""

def get_messages_info(state):
    srs = "SRS:\n" + state['srs'][-1].content
    last_message = "User Query:\n" + state['messages'][-1].content

    return [SystemMessage(content=LAPrompt), 
            HumanMessage(content=srs)] + state['messages']

def information_gathering(state):
    messages = get_messages_info(state["messages"])
    hld = state.get('hld')

    if hld:
        pass
    else:
        response = llm_with_tool.invoke(messages)
        return {"messages": [response]}

代理检查是否已收到识别的数据点：

def is_clarified(state):
    messages = state["messages"]
    hld = state.get('hld')
    if isinstance(messages[-1], AIMessage) and messages[-1].tool_calls:
        return "yes" 
    else:
        return "no"

代理结束对话，如果它收到了所有数据点：

def conclude_conversation(state):
    return {
        "messages": [
            ToolMessage(
                content="Clarified and proceeding further",
                tool_call_id=state["messages"][-1].tool_calls[0]["id"],
            )
        ]
    }

定义代理以生成 HLD：

生成高层设计文档，基于收集到的数据：

SAPrompt = '''
Prompt to generate HLD
'''

SAReviewPrompt = '''
Prompt to revise HLD based on review comments or feedback
'''

def get_prompt_messages(messages: list, state):
    tool_call = None
    other_msgs = []

    for m in messages:
        if isinstance(m, AIMessage) and m.tool_calls:
            tool_call = m.tool_calls[0]["args"]
        elif isinstance(m, ToolMessage):
            continue
        elif tool_call is not None:
            other_msgs.append(m)

    iteration = state['iteration']
    format = state['hld_format'][-1].content
    last_message = state['messages'][-1].content
    srs = state['srs'][-1].content

    print("***** Revision Number *****", iteration)

    hld = state.get('hld')
    if hld:
        hld = state['hld'][-1].content

        return [
            SystemMessage(content=SAReviewPrompt.format(review=last_message, format=format)),
            HumanMessage(content=hld + srs)
        ]
    else:
        return [
            SystemMessage(content=SAPrompt.format(reqs=tool_call, format=format)),
            HumanMessage(content=srs)
        ] + other_msgs

def generate_hld(state):
    messages = get_prompt_messages(state["messages"], state)

    response = llm.invoke(messages)
    iteration = str(state['iteration'])
    file_name = "output/hld v" + str(iteration) + ".md"

    write_file.invoke({"file_path": file_name, "text": response.content})

    return {"messages": [response], 
            "hld": [response]}

定义代理以审查 HLD：

审查者代理决定 HLD 是否需要改进，或者是否可以进行下一步：

critique = """
Prompt to review the HLD and provide feedback for revision
"""

def get_feedback_info(hld, srs):
    return [SystemMessage(content=critique),
           HumanMessage(content="SRS:\n" + srs[-1].content + "Current HLD:\n" + hld[-1].content)]
           
llm = ChatOpenAI(model=model_name, temperature=0, max_retries=1)

def reviewer(state):
    hld = state['hld']
    srs = state['srs']
    messages = get_feedback_info(hld, srs)

    response = llm.invoke(messages)

    iteration = str(state['iteration'])
    file_name = "output/hld_feedback v" + str(iteration) + ".md"

    write_file.invoke({"file_path": file_name, "text": response.content})

    max_iteration = state['max_iteration']
    iteration = state['iteration'] + 1

    return {
        "messages": [response],
        "iteration": iteration
    }

def is_reviewed(state):
    max_iteration = state['max_iteration']
    iteration = state['iteration']
    last_message = state['messages'][-1].content

    if "Satisfied" in last_message.lower():
        return 'reviewed'
    elif iteration > max_iteration:
        return 'reviewed'
    else:
        return 'enhance'

现在，通过组装我们迄今为止定义的节点和边来定义工作流：

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import END

class State(TypedDict):
    messages: Annotated[list, add_messages]
    srs: Annotated[list, add_messages]
    hld: Annotated[list, add_messages]
    hld_format: Annotated[list, add_messages]
    max_iteration: int
    iteration: int

memory = MemorySaver()
workflow = StateGraph(State)

workflow.add_edge(START, "information_gathering")

workflow.add_node("information_gathering", information_gathering)
workflow.add_node("generate_hld", generate_hld)
workflow.add_node("conclude_conversation", conclude_conversation)
workflow.add_node("reviewer", reviewer)

workflow.add_conditional_edges(
    "information_gathering", 
    is_clarified, 
    {"yes": "conclude_conversation",  "no": END}
)

workflow.add_conditional_edges(
    "reviewer", 
    is_reviewed, 
    {"reviewed": END,  "enhance": "generate_hld"}
)

workflow.add_edge("conclude_conversation", "generate_hld")
workflow.add_edge("generate_hld", "reviewer")

graph = workflow.compile(checkpointer=memory)

编译工作流：

graph = workflow.compile(checkpointer=memory)

显示已编译的工作流：

from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))

步骤涉及 LangGraph 工作流生成 HLD：

执行工作流以检查结果：

thread = {"configurable": {"thread_id": 4}}
SRS = read_file.invoke({"file_path": "input/srs.md"})
FORMAT = read_file.invoke({"file_path": "input/hld_format.md"})

count = 1

while True:
    user = input("User (q/Q to quit): ")
    if user.lower() in ["quit", "q", "Q"]:
        print("AI: Byebye")
        break
    output = None

    user_msg = user

    for output in graph.stream(
        {
            "messages": [HumanMessage(content=user_msg)],
            "srs": [HumanMessage(content=SRS)],
            "hld_format": [HumanMessage(content=FORMAT)],
            "iteration": 1,
            "max_iteration":

## 结论

使用多智能体框架，如 LangGraph 和 LangChain，自动化高层设计的生成显著提升了软件开发过程。通过最小化手动任务，这种方法促进了效率和准确性，使团队能够专注于战略决策。

## 仅供说明目的的生产力分析

请注意，下面提到的分析，包括手动工作和生产力提升，仅供说明目的，应该在特定用例的背景下进行验证。

## 免责声明

> 虽然自动化在像 SRS 和高层设计 (HLD) 创建等流程中提供了可观的好处，但最终批准和验证时必须涉及人工监督。这确保了所有需求都被准确捕获，与业务目标对齐，并符合组织标准。本讨论中包含的代码片段旨在作为自动化过程的示例。这些示例可能需要增强以满足特定项目需求。请注意，提到的财务收益或投资回报率计算是粗略估算，而非实际数字。

## 关键要点：

> 在高层设计创建过程中采用自动化可以简化工作流程，但必须与人类的洞察力相平衡。

> 迭代审查和反馈循环对于实现最佳设计结果至关重要。

## 接下来是什么？

在我们系列的下一部分中，我们将深入探讨软件开发生命周期的编码阶段。我们将探讨如何利用Agentic Workflow和最佳实践将软件需求规格（生成SRS）和高层设计（生成HLD）转化为代码，以确保质量和可维护性。

敬请关注，了解如何利用人工智能和自动化来增强编码，并使开发周期更加高效！

如果您错过了上一部分，请查看下面的内容：