875-LangChain框架Use-Cases - 代码调试系统

1. 案例目标

本案例旨在构建一个基于LangGraph的AI驱动Python代码调试系统，通过自动化流程执行代码、分析错误、建议修复并验证修正。

系统主要实现以下目标：

自动执行Python代码并捕获错误
使用AI分析错误并识别根本原因
生成修复后的代码和单元测试
验证修正后的代码是否正常工作
展示LangGraph框架在构建复杂工作流中的应用

2. 技术栈与核心依赖

LangChain

LangGraph

OpenAI GPT

Python

Jupyter Notebook

Pydantic

Subprocess

3. 环境配置

3.1 安装依赖

# 安装所需包 %pip install langchain-opentutorial from langchain_opentutorial import package package.install( [ "langsmith", "langchain", "langchain_core", "langchain_community", "langchain_openai", "langgraph", ], verbose=False, upgrade=False, )

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', encoding='utf-8') as f: code = f.read() # 编译文件内容以预先检查语法错误 try: compile(code, file_path, 'exec') except SyntaxError as se: return ExecutionResult( success=False, output="", error=f"SyntaxError: {se}", code=code ) # 通过subprocess执行文件（捕获运行时错误） result = subprocess.run( [sys.executable, file_path], capture_output=True, text=True ) success = result.returncode == 0 error_msg = result.stderr if not success else None return ExecutionResult( success=success, output=result.stdout, error=error_msg, code=code ) else: # 否则，假设源是代码字符串并在内部执行 code = source try: compiled_code = compile(code, "", "exec") except SyntaxError as se: return ExecutionResult( success=False, output="", error=f"SyntaxError: {se}", code=code ) # 捕获exec()执行期间的输出 output_capture = io.StringIO() try: with contextlib.redirect_stdout(output_capture): exec(compiled_code, {}) except Exception as e: return ExecutionResult( success=False, output=output_capture.getvalue(), error=str(e), code=code ) return ExecutionResult( success=True, output=output_capture.getvalue(), error=None, code=code )

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, shell=shell, text=True, capture_output=True, check=True ) if result.stdout: print("Output:", result.stdout) if result.stderr: print("Stderr:", result.stderr) return True except subprocess.CalledProcessError as e: print(f"Command failed: {e}") print("Error output:", e.stderr) return False except Exception as e: print(f"Unexpected error: {e}") return False class UVCommandExecutor(BaseCommandExecutor): """UV环境执行器""" def install_package(self, package: str) -> bool: cmd = ["uv", "pip", "install", package] print(f"Installing 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(f"Installing 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(f"Installing 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(f"Installing 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_output=True, text=True ) 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_output=True, text=True) 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, shell=True) if not success: print(f"Failed to execute: {cmd}") return False return True

4.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_factory=list, description="修正后的代码") bash: List[str] = Field(default_factory=list, description="Bash命令") 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(model="gpt-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", encoding="utf-8") as f: file_content = f.read() state["original_code"] = file_content result = execute_code(state["code"]) updated_attempts = state["debug_attempts"] + 1 print(f"Debug 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 = f"pip install {package}" executor.execute_bash_commands([package]) return state def decide_next_step(state: AgentState) -> str: """决策节点，确定工作流中的下一步 返回: str: 三种可能的结果之一: - "end": 如果代码执行成功 - "max_attempts_reached": 如果达到最大调试尝试次数(3) - "retry": 如果需要进一步调试 """ print(f"Current 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(f"Installed Dependencies: {result['dependencies']}") print(f"Debug 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(f"Fix History Length: {len(result['fix_history'])}") for fix in result['fix_history']: print(f"Attempt: {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响应解析的技术方案
展示了状态管理和流程控制的应用

这个案例为构建类似的智能开发工具提供了参考，特别是在需要自动化代码分析和修复的场景中。通过扩展和优化，该系统可以应用于更广泛的开发场景，如代码审查、自动化测试、代码重构等。

875-LangChain框架Use-Cases - 代码调试系统 - 案例分析