编程的未来 — 使用 LangGraph 的多智能体LLM框架

人工智能（AI）正在迅速改变我们的生活和工作方式，软件工程领域也正在经历深刻变革。软件开发在塑造我们未来方面发挥着关键作用，尤其是随着人工智能和自动化的出现。如今，软件开发站在这场技术革命的前沿。

基于LLM的智能体越来越多地被用于自动化软件开发中的许多任务，从代码编写到测试和调试。利用多智能体框架使这些智能体能够自主运行，减少了对广泛人类干预的必要性。本文深入探讨了使用LangGraph开发多智能体编码器的过程，探讨了其功能和智能体与基于链的LLM解决方案的集成。

最近的论文和工具，如 AgentCoder， AlphaCodium，以及 GPTEngineer**，强调了这种方法的重要性。**首先，让我们深入探讨LangGraph为创建这些智能体和部署基于链的LLM解决方案提供的能力。

LangGraph简介

LangGraph通过在循环图结构中利用大型语言模型的能力，增强了LangChain生态系统，从而实现了复杂智能体运行时的开发。

LangGraph的关键组件

1. 状态图： LangGraph中的主要图类型是StatefulGraph，其由传递给每个节点的状态对象参数化。图中的节点返回操作以更新此状态，可以通过设置特定属性或添加到现有属性来完成此操作。
2. 节点： LangGraph中的节点代表应用程序中负责特定任务的智能体组件。每个节点与状态对象交互，并根据其在智能体架构中的功能返回操作以更新它。
3. 边： 边连接图中的节点，定义了信息和操作在应用程序的不同组件之间的流动。它们促进节点之间的通信和协调，以实现所需的功能。

使用LangGraph实现

在这里，我们首先定义架构流程和不同智能体。我们将这些智能体专门化为特定任务并分配角色。

高级流程及每个智能体的角色

智能体节点“程序员”

任务： 根据需求专门编写程序

class Code(BaseModel):
    """未来要遵循的计划"""

    code: str = Field(
        description="根据提供的需求详细优化无错误的Python代码"
    )

from langchain.chains.openai_functions import create_structured_output_runnable
from langchain_core.prompts import ChatPromptTemplate

code_gen_prompt = ChatPromptTemplate.from_template(
    '''**角色**：您是一位专业的软件Python程序员。您需要开发Python代码
**任务**：作为程序员，您需要完成函数。使用一种思维链的方法来分解问题，创建伪代码，然后用Python语言编写代码。确保您的代码是高效的、可读的和有良好注释的。

**说明**：
1. **理解和澄清**：确保您理解任务。
2. **算法/方法选择**：决定最有效的方法。
3. **生成伪代码**：将您将遵循的步骤写成伪代码。
4. **代码生成**：将您的伪代码转换为可执行的Python代码

*需求*
{requirement}'''
)
coder = create_structured_output_runnable(
    Code, llm, code_gen_prompt
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

智能体节点“测试员”

任务： 生成输入测试用例和预期输出

class Tester(BaseModel):
    """未来要遵循的计划"""

    Input: List[List] = Field(
        description="用于评估提供的代码的测试用例输入"
    )
    Output: List[List] = Field(
        description="用于评估提供的代码的测试用例预期输出"
    )

from langchain.chains.openai_functions import create_structured_output_runnable
from langchain_core.prompts import ChatPromptTemplate

test_gen_prompt = ChatPromptTemplate.from_template(
    '''**角色**：作为测试员，您的任务是根据提供的需求和Python代码创建基本和简单的测试用例。
这些测试用例应涵盖基本、边缘情况，以确保代码的健壮性、可靠性和可扩展性。
**1. 基本测试用例**：
- **目标**：验证基本功能的基本和小规模测试用例
**2. 边缘测试用例**：
- **目标**：评估函数在极端或不寻常条件下的行为。
**说明**：
- 根据需求实施一套全面的测试用例。
- 特别关注边缘情况，因为它们经常会暴露隐藏的错误。
- 仅生成小规模的基本和边缘情况测试用例。只关注小规模、基本测试用例
*需求*
{requirement}
**代码**
{code}
'''
)
tester_agent = create_structured_output_runnable(
    Test, llm, test_gen_prompt
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

智能体节点“执行器”

任务： 在提供的测试用例中在Python环境中执行代码

class ExecutableCode(BaseModel):
    """未来要遵循的计划"""

    code: str = Field(
        description="带有测试用例断言的详细优化无错误的Python代码"
    )

python_execution_gen = ChatPromptTemplate.from_template(
    """您必须在*Python代码*中添加测试层，以帮助执行代码。您只需传递提供的输入作为参数，并验证给定的预期输出是否匹配。
*指导*：
- 确保在断言失败时返回错误
- 生成可执行的代码
要执行的Python代码：
*Python代码*：{code}
代码的输入和输出：
*输入*：{input}
*预期输出*：{output}"""
)
execution = create_structured_output_runnable(
    ExecutableCode, llm, python_execution_gen
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

智能体节点“调试器”

任务： 使用LLM知识调试代码，并在出现错误时将其发送回“执行器”智能体。

class RefineCode(BaseModel):

    code: str = Field(
        description="优化和改进的Python代码以解决错误"
    )

python_refine_gen = ChatPromptTemplate.from_template(
    """您是Python调试专家。您需要分析给定的代码和错误，并生成可以处理错误的代码
    *指导*：
    - 确保生成无错误的代码
    - 生成的代码能够处理错误
    
    *代码*：{code}
    *错误*：{error}
    """
)
refine_code = create_structured_output_runnable(
    RefineCode, llm, python_refine_gen
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

决策边缘“决定_结束”

任务： 这是一个条件边缘，基于“执行器”。如果遇到错误，则会决定是结束执行还是将其发送到“调试器”以解决错误。

图状态和定义图节点

在LangGraph中，状态对象有助于跟踪图状态。

class AgentCoder(TypedDict):
    requirement: str #跟踪需求
    code: str #最新代码
    tests: Dict[str, any] #生成的测试用例
    errors: Optional[str] #如果遇到错误，跟踪错误
1
2
3
4
5

为每个智能体流程定义功能和执行

def programmer(state):
    print(f'进入程序员')
    requirement = state['requirement']
    code_ = coder.invoke({'requirement':requirement})
    return {'code':code_.code}

def debugger(state):
    print(f'进入调试器')
    errors = state['errors']
    code = state['code']
    refine_code_ = refine_code.invoke({'code':code,'error':errors})
    return {'code':refine_code_.code,'errors':None}

def executer(state):
    print(f'进入执行器')
    tests = state['tests']
    input_ = tests['input']
    output_ = tests['output']
    code = state['code']
    executable_code = execution.invoke({"code":code,"input":input_,'output':output_})
    #print(f"可执行代码 - {executable_code.code}")
    error = None
    try:
        exec(executable_code.code)
        print("代码执行成功")
    except Exception as e:
        print('运行时发现错误')
        error = f"执行错误：{e}"
    return {'code':executable_code.code,'errors':error}

def tester(state):
    print(f'进入测试员')
    requirement = state['requirement']
    code = state['code']
    tests = tester_agent.invoke({'requirement':requirement,'code':code})
    #tester.invoke({'requirement':'Generate fibbinaco series','code':code_.code})
    return {'tests':{'input':tests.Input,'output':tests.Output}}

def decide_to_end(state):
    print(f'进入决定结束')
    if state['errors']:
        return 'debugger'
    else:
        return 'end'
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

通过边缘将不同节点集成在一起，并添加条件边缘“决定_结束”

from langgraph.graph import END, StateGraph

workflow = StateGraph(AgentCoder)

# 定义节点
workflow.add_node("programmer", programmer)  
workflow.add_node("debugger", debugger) 
workflow.add_node("executer", executer) 
workflow.add_node("tester", tester) 
#workflow.add_node('decide_to_end',decide_to_end)

# 构建图
workflow.set_entry_point("programmer")
workflow.add_edge("programmer", "tester")
workflow.add_edge("debugger", "executer")
workflow.add_edge("tester", "executer")
#workflow.add_edge("executer", "decide_to_end")

workflow.add_conditional_edges(
    "executer",
    decide_to_end,
    {
        "end": END,
        "debugger": "debugger",
    },
)

# 编译
app = workflow.compile()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

config = {"recursion_limit": 50}
inputs = {"requirement": requirement}
running_dict = {}
async for event in app.astream(inputs, config=config):
    for k, v in event.items():
        running_dict[k] = v
        if k != "__end__":
            print(v)
1
2
3
4
5
6
7
8

完整代码链接- [https://github.com/anurag899/openAI-project/blob/main/Multi-Agent%20Coding%20Framework%20using%20LangGraph/LangGraph%20-%20Code%20Development%20using%20Multi-Agent%20Flow.ipynb](https://github.com/anurag899/openAI-project/blob/main/Multi-Agent Coding Framework using LangGraph/LangGraph - Code Development using Multi-Agent Flow.ipynb)

结论

基于LLM的代理是一个不断发展的领域，仍然需要进行大量检查才能使其达到生产就绪状态。这是使用基于多代理的流程的原型。我未来计划在代理节点中集成工具（比如允许“调试器”代理引用基于RAG的库文档和Web工具，以提供最新和精炼的代码，使用内部工具通过“执行器”代理执行代码）