网站建设销售一个月营业额,安顺网站设计,遵义网站建设公司,制作网页的网站1. 案例目标
本案例旨在构建一个基于LangGraph的AI驱动Python代码调试系统#xff0c;通过自动化流程执行代码、分析错误、建议修复并验证修正。
系统主要实现以下目标#xff1a;
自动执行Python代码并捕获错误使用AI分析错误并识别根本原因生成修复后的代码和单元测试验…1. 案例目标本案例旨在构建一个基于LangGraph的AI驱动Python代码调试系统通过自动化流程执行代码、分析错误、建议修复并验证修正。系统主要实现以下目标自动执行Python代码并捕获错误使用AI分析错误并识别根本原因生成修复后的代码和单元测试验证修正后的代码是否正常工作展示LangGraph框架在构建复杂工作流中的应用2. 技术栈与核心依赖LangChainLangGraphOpenAI GPTPythonJupyter NotebookPydanticSubprocess3. 环境配置3.1 安装依赖# 安装所需包 %pip install langchain-opentutorial from langchain_opentutorial import package package.install( [ langsmith, langchain, langchain_core, langchain_community, langchain_openai, langgraph, ], verboseFalse, upgradeFalse, )3.2 环境变量设置# 设置环境变量 from langchain_opentutorial import set_env set_env( { OPENAI_API_KEY: , LANGCHAIN_API_KEY: , LANGCHAIN_TRACING_V2: true, LANGCHAIN_ENDPOINT: https://api.smith.langchain.com, LANGCHAIN_PROJECT: CodeDebuggingSystem, } )4. 案例实现4.1 系统架构设计4.2 基础组件4.2.1 代码执行器import os import sys import subprocess import io import contextlib from typing_extensions import TypedDict from typing import Optional class ExecutionResult(TypedDict): success: bool output: Optional[str] error: Optional[str] code: Optional[str] def execute_code(source: str) - ExecutionResult: 统一代码执行函数。 如果source是有效的文件路径则读取并执行该文件。 否则将source视为Python代码片段并在进程内执行。 Args: source (str): Python文件路径或代码片段。 Returns: ExecutionResult: 包含执行结果的字典: - success (bool) - output (str): 捕获的标准输出 - error (Optional[str]): 错误信息如果有 - code (str): 已执行的源代码 # 如果源是文件路径执行该文件 if os.path.isfile(source): file_path source with open(file_path, r, encodingutf-8) as f: code f.read() # 编译文件内容以预先检查语法错误 try: compile(code, file_path, exec) except SyntaxError as se: return ExecutionResult( successFalse, output, errorfSyntaxError: {se}, codecode ) # 通过subprocess执行文件捕获运行时错误 result subprocess.run( [sys.executable, file_path], capture_outputTrue, textTrue ) success result.returncode 0 error_msg result.stderr if not success else None return ExecutionResult( successsuccess, outputresult.stdout, errorerror_msg, codecode ) else: # 否则假设源是代码字符串并在内部执行 code source try: compiled_code compile(code, , exec) except SyntaxError as se: return ExecutionResult( successFalse, output, errorfSyntaxError: {se}, codecode ) # 捕获exec()执行期间的输出 output_capture io.StringIO() try: with contextlib.redirect_stdout(output_capture): exec(compiled_code, {}) except Exception as e: return ExecutionResult( successFalse, outputoutput_capture.getvalue(), errorstr(e), codecode ) return ExecutionResult( successTrue, outputoutput_capture.getvalue(), errorNone, codecode )4.2.2 命令执行器from typing import List, Union from abc import ABC, abstractmethod class BaseCommandExecutor(ABC): 命令执行基类 def __init__(self): self.python_path sys.executable self.venv_path os.path.dirname(os.path.dirname(self.python_path)) abstractmethod def install_package(self, package: str) - bool: 包安装方法 pass def run_command(self, command: Union[List[str], str], shell: bool False) - bool: 执行通用命令 try: if isinstance(command, str): shell True result subprocess.run( command, shellshell, textTrue, capture_outputTrue, checkTrue ) if result.stdout: print(Output:, result.stdout) if result.stderr: print(Stderr:, result.stderr) return True except subprocess.CalledProcessError as e: print(fCommand failed: {e}) print(Error output:, e.stderr) return False except Exception as e: print(fUnexpected error: {e}) return False class UVCommandExecutor(BaseCommandExecutor): UV环境执行器 def install_package(self, package: str) - bool: cmd [uv, pip, install, package] print(fInstalling with UV: { .join(cmd)}) return self.run_command(cmd) class VenvCommandExecutor(BaseCommandExecutor): 标准venv环境执行器 def install_package(self, package: str) - bool: cmd [self.python_path, -m, pip, install, package] print(fInstalling with pip: { .join(cmd)}) return self.run_command(cmd) class PoetryCommandExecutor(BaseCommandExecutor): Poetry环境执行器 def install_package(self, package: str) - bool: cmd [poetry, add, package] print(fInstalling with Poetry: { .join(cmd)}) return self.run_command(cmd) class CondaCommandExecutor(BaseCommandExecutor): Conda环境执行器 def install_package(self, package: str) - bool: cmd [conda, install, -y, package] print(fInstalling with Conda: { .join(cmd)}) return self.run_command(cmd) def get_appropriate_executor() - BaseCommandExecutor: 返回适合当前环境的执行器 def check_notebook_env() - bool: 检查当前环境是否为Jupyter Notebook。 try: from IPython import get_ipython ip get_ipython() # 如果存在IPKernelApp表示它在Jupyter Notebook或qtconsole中运行。 if ip is not None and IPKernelApp in ip.config: print(Jupyter Notebook environment detected.) return True except Exception: return False def check_uv_available() - bool: try: result subprocess.run( [uv, --version], capture_outputTrue, textTrue ) return result.returncode 0 except Exception: return False def check_poetry_available() - bool: if os.path.exists(pyproject.toml): try: import toml except ImportError: toml None if toml: try: config toml.load(pyproject.toml) if tool in config and poetry in config[tool]: print(Poetry configuration detected in pyproject.toml.) return True except Exception: pass else: # 如果toml模块未安装基于文件存在性假设环境 print(pyproject.toml found (toml module not installed), assuming Poetry environment.) return True return False def check_conda_available() - bool: try: result subprocess.run([conda, --version], capture_outputTrue, textTrue) return result.returncode 0 except Exception: return False # 检查环境类型 if check_poetry_available(): print(Poetry environment detected, using PoetryCommandExecutor) return PoetryCommandExecutor() elif check_uv_available(): print(UV environment detected, using UVCommandExecutor) return UVCommandExecutor() elif check_conda_available(): print(Conda environment detected, using CondaCommandExecutor) return CondaCommandExecutor() else: print(Standard Python environment detected, using VenvCommandExecutor) return VenvCommandExecutor() class CommandExecutor: 命令执行的统一接口 def __init__(self): self.executor get_appropriate_executor() def execute_bash_commands(self, commands: List[str]) - bool: 执行bash命令列表 for cmd in commands: print(f\nExecuting: {cmd}) # 处理pip install命令 if cmd.startswith(pip install): package cmd.replace(pip install, ).strip() success self.executor.install_package(package) else: success self.executor.run_command(cmd, shellTrue) if not success: print(fFailed to execute: {cmd}) return False return True4.2.3 代码块解析器import re from typing import List from langchain_core.output_parsers import BaseOutputParser from pydantic import BaseModel, Field class CodeBlocks(BaseModel): 保存按语言分类的代码块的字典结构 corrected: List[str] Field(default_factorylist, description修正后的代码) bash: List[str] Field(default_factorylist, descriptionBash命令) class CodeBlockParser(BaseOutputParser[CodeBlocks]): Markdown代码块的解析器 按语言分类并以字典格式返回代码块 def parse(self, text: str) - CodeBlocks: 从文本中提取并按语言分类代码块 # 初始化代码块字典 code_blocks { corrected: [], bash: [], } # 匹配language:type格式的模式 pattern r(\w):?(\w)?\n(.*?)\n matches re.finditer(pattern, text, re.DOTALL) for match in matches: language match.group(1).lower() block_type match.group(2) or language # 如果未指定类型则使用语言作为类型 code match.group(3).strip() if block_type in code_blocks: code_blocks[block_type].append(code) return CodeBlocks(**code_blocks) property def _type(self) - str: 指定解析器类型 return code_block_parser def get_format_instructions(self) - str: 解析的格式说明 return 代码应以markdown格式返回如下所示 language your_code_here 其中language应为python:corrected或bash。4.2.4 LLM组件from langchain_openai import ChatOpenAI from langchain_core.prompts import PromptTemplate model ChatOpenAI(modelgpt-4o-mini) prompt PromptTemplate.from_template(你得到了一个Python代码片段以及运行代码时发生的错误消息。分析代码识别错误原因并提供修正后的代码版本。 --- **上下文**: - 错误消息: {ERROR_MESSAGE} - 原始代码: {CODE_SNIPPET} **格式要求**: - 使用**确切**的这些代码块标签: - 修正后的代码: python:corrected - 包安装: bash - 不要使用通用的python代码块 **响应结构**: ### 1. 错误分析 - 识别导致错误的特定行号 - 使用Python运行机制解释根本原因 - 提供类似错误的预防策略 ### 2. 代码修正 python:corrected # 必须包含: # - 带行注释的错误修复 # - 增强的异常处理 # - PEP 8合规性 # - 关键函数的类型提示 ### 3. 推荐建议 - 包依赖(不要忘记) bash pip install [detected_packages] - 扫描import语句中的第三方包 - 排除标准库模块(例如os, sys) - 在一行安装命令中按字母顺序列出包 - 代码质量(PEP 8) - 运行时考虑 按照上面显示的清晰部分和标记代码块格式化你的响应。 ) chain prompt | model | CodeBlockParser()4.3 构建LangGraph工作流4.3.1 定义状态from typing import Optional, List from typing_extensions import TypedDict from langgraph.graph import StateGraph, END # 定义代理的状态 class AgentState(TypedDict): original_code: str # 原始代码 code: str error: Optional[str] dependencies: List[str] execution_result: Optional[ExecutionResult] debug_attempts: int fix_history: List[dict] # 构建图 builder StateGraph(AgentState)4.3.2 定义节点# 初始化组件 executor CommandExecutor() code_parser CodeBlockParser() llm_chain prompt | model # 定义节点 def execute_code_node(state: AgentState) - AgentState: if os.path.isfile(state[code]): with open(state[code], r, encodingutf-8) as f: file_content f.read() state[original_code] file_content result execute_code(state[code]) updated_attempts state[debug_attempts] 1 print(fDebug attempts updated to: {updated_attempts}) return { **state, execution_result: result, error: result[error] if not result[success] else None, debug_attempts: updated_attempts } def analyze_error_node(state: AgentState) - AgentState: if not state[error]: return state response llm_chain.invoke({ ERROR_MESSAGE: state[error], CODE_SNIPPET: state[code] }) parsed code_parser.parse(response.content) # 记录调试尝试详情: # - attempt: 当前尝试次数 # - error: 当前错误消息 # - fixed_code: AI提供的修正代码(如果可用) fix_record { attempt: state[debug_attempts], error: state[error], fixed_code: \n.join(parsed.corrected) if parsed.corrected else None } return { **state, code: \n.join(parsed.corrected) if parsed.corrected else state[code], # 用修正版本更新代码 dependencies: parsed.bash, fix_history: state.get(fix_history, []) [fix_record] } def install_deps_node(state: AgentState) - AgentState: 安装所需依赖 if state[dependencies]: for package in state[dependencies]: # 确保命令以pip install开头 if not package.startswith(pip install): package fpip install {package} executor.execute_bash_commands([package]) return state def decide_next_step(state: AgentState) - str: 决策节点确定工作流中的下一步 返回: str: 三种可能的结果之一: - end: 如果代码执行成功 - max_attempts_reached: 如果达到最大调试尝试次数(3) - retry: 如果需要进一步调试 print(fCurrent debug attempts: {state[debug_attempts]}) # 首先检查最大尝试次数以确保我们停止 if state[debug_attempts] 3: print(Max attempts reached, ending workflow) return max_attempts_reached # 然后检查成功执行 if state[execution_result][success] and not state.get(error): print(Successful execution, ending workflow) return end # 只有在尝试限制内且不成功时才继续 print(Continuing debug cycle) return retry # 添加节点 builder.add_node(execute, execute_code_node) builder.add_node(analyze, analyze_error_node) builder.add_node(install, install_deps_node)4.3.3 定义边# 添加边以创建更逻辑的流程 builder.set_entry_point(execute) builder.add_edge(analyze, install) builder.add_edge(install, execute) # 安装依赖后再次执行代码 # 从执行节点添加条件边 builder.add_conditional_edges( execute, decide_next_step, { end: END, # 如果执行成功结束工作流 max_attempts_reached: END, # 如果达到最大尝试次数结束工作流 retry: analyze # 如果错误持续继续调试 } ) # 编译图 app builder.compile()4.4 示例执行# 运行图 initial_state { original_code: sample_code, code: sample_code, error: None, dependencies: [], debug_attempts: 0, fix_history: [] } result app.invoke(initial_state) # 打印结果 print(fInstalled Dependencies: {result[dependencies]}) print(fDebug Attempts: {result[debug_attempts]}) print(\n\n----- Original Code -----) print(f{result[original_code]}) print(\n\n----- Final Code -----) print(f{result[code]}) print(\n\n----- Fix History -----) print(fFix History Length: {len(result[fix_history])}) for fix in result[fix_history]: print(fAttempt: {fix[attempt]}) print(----- Error -----) print(f{fix[error]}) print(----- Fixed Code -----) print(f{fix[fixed_code]})5. 案例效果5.1 自动错误检测系统能够自动检测Python代码中的语法错误、运行时错误和类型错误。5.2 智能错误分析系统能够分析错误的根本原因并提供详细的错误解释和修复建议。5.3 自动代码修复系统能够生成修复后的代码包括语法修正、类型转换和异常处理。5.4 依赖管理系统能够识别并安装代码所需的依赖包支持多种包管理器。5.5 迭代调试系统能够进行多轮调试尝试直到代码成功运行或达到最大尝试次数。5.6 调试历史记录系统能够记录每次调试尝试的详细信息包括错误和修复方案。6. 案例实现思路1 代码执行环境设计设计安全的代码执行环境能够捕获输出和错误信息支持文件和代码片段两种输入方式。2 命令执行器设计设计环境感知的命令执行器自动检测并适配不同的Python环境和包管理器。3 代码块解析器设计设计自定义的LangChain输出解析器能够从AI响应中提取和分类不同类型的代码块。4 LLM提示设计设计结构化的提示模板引导AI进行错误分析、代码修正和依赖识别。5 状态管理设计设计状态管理机制跟踪调试过程中的关键信息包括代码、错误、依赖和历史记录。6 工作流构建使用LangGraph构建调试工作流定义节点、边和条件路由实现自动化的调试流程。7. 扩展建议支持更多编程语言的调试如JavaScript、Java、C等添加更复杂的代码分析功能如性能分析、安全漏洞检测等实现代码质量评估功能提供代码风格和最佳实践建议集成版本控制系统支持Git仓库中的代码调试添加调试可视化界面展示调试过程和结果实现调试知识库积累常见错误和解决方案添加代码重构建议帮助改进代码结构和设计支持调试结果的导出和分享功能8. 总结本案例展示了如何使用LangChain和LangGraph构建一个AI驱动的代码调试系统。通过代码执行、错误分析、代码修正和依赖管理等核心功能系统能够自动化地检测和修复Python代码中的错误。该案例的核心价值在于展示了自动化代码调试的实现方法演示了多环境适配的命令执行机制提供了结构化AI响应解析的技术方案展示了状态管理和流程控制的应用这个案例为构建类似的智能开发工具提供了参考特别是在需要自动化代码分析和修复的场景中。通过扩展和优化该系统可以应用于更广泛的开发场景如代码审查、自动化测试、代码重构等。