【DeepSeek 实战】打造全能编程助手：DeepSeek V4 Agent 开发与工具调用

from typing import Dict, List
import subprocess
import os
# 工具定义列表
TOOLS = [
    {
        "name": "execute_python_code",
        "description": "执行 Python 代码并返回输出结果。适用于测试代码片段或验证逻辑。",
        "parameters": {
            "type": "object",
            "properties": {
                "code": {
                    "type": "string",
                    "description": "要执行的 Python 代码"
                }
            },
            "required": ["code"]
        }
    },
    {
        "name": "read_file",
        "description": "读取指定文件的内容",
        "parameters": {
            "type": "object",
            "properties": {
                "file_path": {
                    "type": "string",
                    "description": "文件的绝对路径或相对路径"
                }
            },
            "required": ["file_path"]
        }
    },
    {
        "name": "run_unit_tests",
        "description": "运行指定文件或目录的单元测试",
        "parameters": {
            "type": "object",
            "properties": {
                "test_path": {
                    "type": "string",
                    "description": "测试文件或目录路径"
                }
            },
            "required": ["test_path"]
        }
    }
]

class ToolExecutor:
    """
    工具执行引擎
    """
    @staticmethod
    def execute_python_code(code: str) -> Dict:
        """
        安全地执行 Python 代码
        """
        try:
            # 使用 subprocess 隔离执行环境
            result = subprocess.run(
                ["python3", "-c", code],
                capture_output=True,
                text=True,
                timeout=10  # 超时保护
            )
            if result.returncode == 0:
                return {
                    "success": True,
                    "output": result.stdout.strip()
                }
            else:
                return {
                    "success": False,
                    "error": result.stderr.strip()
                }
        except subprocess.TimeoutExpired:
            return {"success": False, "error": "代码执行超时"}
        except Exception as e:
            return {"success": False, "error": str(e)}
    @staticmethod
    def read_file(file_path: str) -> Dict:
        """
        读取文件内容
        """
        try:
            if not os.path.exists(file_path):
                return {"success": False, "error": f"文件不存在: {file_path}"}
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()
            return {
                "success": True,
                "content": content[:5000]  # 限制返回长度
            }
        except Exception as e:
            return {"success": False, "error": str(e)}
    @staticmethod
    def run_unit_tests(test_path: str) -> Dict:
        """
        运行单元测试
        """
        try:
            result = subprocess.run(
                ["pytest", test_path, "-v"],
                capture_output=True,
                text=True,
                timeout=60
            )
            return {
                "success": result.returncode == 0,
                "output": result.stdout + result.stderr
            }
        except Exception as e:
            return {"success": False, "error": str(e)}
# 工具映射表
TOOL_FUNCTIONS = {
    "execute_python_code": ToolExecutor.execute_python_code,
    "read_file": ToolExecutor.read_file,
    "run_unit_tests": ToolExecutor.run_unit_tests
}

from deepseek import AsyncDeepSeek
import json
import asyncio
class DeepSeekAgent:
    def __init__(self, api_key: str):
        self.client = AsyncDeepSeek(api_key=api_key)
        self.tools = TOOLS
        self.tool_functions = TOOL_FUNCTIONS
        self.conversation_history = []
    async def run(self, user_input: str, max_iterations: int = 5) -> str:
        """
        运行 Agent，处理用户任务
        :param user_input: 用户输入的任务描述
        :param max_iterations: 最大迭代次数，防止死循环
        :return: 最终答案
        """
        self.conversation_history.append({
            "role": "user",
            "content": user_input
        })
        for iteration in range(max_iterations):
            # 1. 调用 LLM，传入工具定义
            response = await self.client.chat.completions.create(
                model="deepseek-chat",
                messages=self.conversation_history,
                tools=self.tools,
                tool_choice="auto"  # 让模型自主决定是否调用工具
            )
            assistant_message = response.choices[0].message
            # 2. 检查是否有工具调用
            if assistant_message.tool_calls:
                tool_calls = assistant_message.tool_calls
                # 3. 执行工具
                tool_results = []
                for tool_call in tool_calls:
                    function_name = tool_call.function.name
                    function_args = json.loads(tool_call.function.arguments)
                    print(f"[Agent] 调用工具: {function_name}, 参数: {function_args}")
                    # 执行对应工具
                    if function_name in self.tool_functions:
                        result = self.tool_functions[function_name](**function_args)
                        tool_results.append({
                            "tool_call_id": tool_call.id,
                            "name": function_name,
                            "result": result
                        })
                # 4. 将工具结果反馈给 LLM
                self.conversation_history.append(assistant_message)
                for tool_result in tool_results:
                    self.conversation_history.append({
                        "role": "tool",
                        "tool_call_id": tool_result["tool_call_id"],
                        "name": tool_result["name"],
                        "content": json.dumps(tool_result["result"], ensure_ascii=False)
                    })
                # 继续下一轮迭代
                continue
            else:
                # 没有工具调用，直接返回答案
                final_answer = assistant_message.content
                self.conversation_history.append(assistant_message)
                return final_answer
        return "任务未完成，已达到最大迭代次数"
# 使用示例
async def main():
    agent = DeepSeekAgent(api_key=os.getenv("DEEPSEEK_API_KEY"))
    # 任务：读取某个文件，检查其中的代码是否有问题，并运行测试验证
    task = """
    请帮我检查 ./src/utils.py 文件中的代码：
    1. 读取文件内容
    2. 分析是否存在潜在 Bug
    3. 如果有问题，生成修复后的代码并执行验证
    4. 运行该文件的单元测试
    """
    result = await agent.run(task)
    print(f"最终答案:\n{result}")
asyncio.run(main())

## 代码审查报告
### 发现的问题
1. **竞态条件 (Race Condition)**
- 位置: `order_service.py` 第 45 行
- 问题: 在检查库存和扣减库存之间存在时间窗口，可能导致超卖
### 修复建议
使用数据库事务或分布式锁确保原子性：
```python
# 修复前
if stock > 0:
    stock -= 1
    save(stock)
# 修复后
with transaction.atomic():
    product = Product.objects.select_for_update().get(id=product_id)
    if product.stock > 0:
        product.stock -= 1
        product.save()

传统人工处理年度成本:
├── 人力成本: 20人 × ¥12,000/月 × 12 = ¥2,880,000
├── 培训费用: ¥80,000/年
├── 管理成本: ¥150,000/年
└── 总计:       ¥3,110,000
Agent 智能处理年度成本:
├── 服务器费用: ¥12,000/月 × 12 = ¥144,000
├── API 费用:   7000次 × 250天 × ¥0.4/次 = ¥700,000
├── 维护人力:   2人 × ¥12,000/月 × 12 = ¥288,000
└── 总计:       ¥1,132,000
🎉 年度节省: ¥1,978,000 (约 198 万元)