LangGraph范式-langgraph_code_assistant

langgraph_code_assistant 旨在构建一个代码助手代理，能够协助用户生成、调试、优化或解释代码。它通过多轮交互、工具调用和状态管理，处理复杂的代码相关任务，例如编写 Python 脚本、调试错误、生成测试用例或解释代码逻辑。

• 动态调整交互流程，基于用户输入或代码执行结果决定下一步。
• 集成外部工具（如代码解释器、静态分析器）以增强功能。
• 保持对话和代码上下文，确保连贯性和准确性。

Code Assistant 的核心流程包括以下步骤：

1. 用户请求处理（Handle User Request）：解析用户输入的编程任务，生成初始代码或计划。
2. 代码生成（Generate Code）：使用 LLM 生成代码片段，基于用户请求和上下文。
3. 代码执行与验证（Execute and Validate Code）：通过代码解释器（如 Python REPL）运行代码，验证其正确性。
4. 错误修正（Fix Errors）：如果执行失败，分析错误并生成修正后的代码。
5. 迭代优化（Iterate and Optimize）：根据用户反馈或验证结果，优化代码或生成新版本。
6. 最终输出（Final Output）：返回验证通过的代码或解决方案。

Code Assistant 通过多轮交互和工具验证，确保代码的高质量和功能性，适合从简单脚本到复杂算法的编程任务。

数据结构定义

Code Assistant 范式需要维护一个状态对象，用于在节点间传递数据。代码定义了 CodeAssistantState 作为状态类型：

from typing import TypedDict, Optional, List, Dict
class CodeAssistantState(TypedDict):
    user_input: str 用户输入的编程任务描述
    code: Optional[str] LLM 生成的代码片段
    execution_result: Optional[str] 代码执行的结果（成功输出或错误信息）
    errors: Optional[str] 执行代码时捕获的错误信息
    iteration: int 当前迭代次数
    max_iterations: int最大迭代次数（如 5)
    final_output: Optional[str] 最终验证通过的代码或解决方案

2.3 核心组件与工作流

Code Assistant 的实现分为多个节点，每个节点负责工作流的一个阶段

2.3.1 代码生成（Generate Code）

第一步是基于用户输入生成代码：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

generate_prompt = ChatPromptTemplate.from_template(
    """You are a code assistant. Based on the user's request, generate Python code.
    User request: {user_input}
    If this is a retry, consider the previous code and errors:
    Previous code: {code}
    Errors: {errors}
    Provide the code in the following format:
    ```python
    {code}
"""
)
generate_chain = generate_prompt | llm | StrOutputParser()
def generate_code(state: CodeAssistantState):
user_input = state["user_input"]
code = state.get("code", "")
errors = state.get("errors", "")
result = generate_chain.invoke({
"user_input": user_input,
"code": code,
"errors": errors
})

markdown中释放代码

code_block = result.split("python\n")[1].split("")[0].strip()
return {
"code": code_block,
"iteration": state["iteration"] + 1
}

2.3.2 代码执行（Execute Code）

第二步是执行生成的代码并捕获结果：

from langchain_experimental.utilities import PythonREPL
repl = PythonREPL()
def execute_code(state: CodeAssistantState):
    code = state["code"]
    try:
        result = repl.run(code)
        return {"execution_result": result, "errors": None}
    except Exception as e:
        return {"execution_result": None, "errors": str(e)}

• 工具：使用 PythonREPL 执行代码，模拟 Python 解释器。
• 逻辑：
  • 尝试运行代码，捕获成功输出或异常。
  • 如果成功，设置 execution_result 为输出，errors 为 None。
  • 如果失败，设置 errors 为异常信息，execution_result 为 None。
• 节点：execute_code 更新状态中的 execution_result 和 errors。
• 特点：REPL 工具提供真实的代码执行环境，适合验证代码功能。

2.3.3 验证与修正决策（Validate and Decide）

第三步是验证代码执行结果并决定是否需要修正：

def validate_and_decide(state: CodeAssistantState):
    errors = state["errors"]
    execution_result = state["execution_result"]
    iteration = state["iteration"]
    max_iterations = state["max_iterations"]
    if errors is None and execution_result is not None:
        return {"final_output": state["code"]}
    if iteration >= max_iterations:
        return {"final_output": f"Failed after {max_iterations} attempts: {errors}"}
    return state

• 逻辑：
  • 如果无错误且有执行结果，设置 final_output 为当前代码。
  • 如果达到最大迭代次数，返回失败信息。
  • 否则，继续迭代，保持当前状态。
• 节点：validate_and_decide 更新 final_output 或保持状态不变。
• 特点：验证逻辑简单但有效，控制迭代终止条件。

2.3.4 路由逻辑（Router）

代码定义了一个 router 函数，用于动态路由到下一个节点：

def router(state: CodeAssistantState):
    if state["final_output"]:
        return "end"
    if state["errors"]:
        return "generate_code"
    return "execute_code"

• 逻辑：
  • 如果 final_output 已生成，路由到 END。
  • 如果有错误，路由到 generate_code 重试。
  • 否则，路由到 execute_code 执行代码。
• 作用：实现迭代循环，确保错误触发重试。