你好我是专注于技术实战分享的博主。在探索AI工程化落地的过程中我发现一个普遍痛点许多开发者对“智能体Agents”的概念充满热情但一到具体实施就卡在了如何构建可维护的代码库、如何与团队协作以及如何评估其效果上。网上资料要么过于理论化要么是零散的代码片段缺乏一套从开发到部署、从个人到团队的完整工程实践指南。本文将围绕“智能体、代码库与团队”这一核心主题为你系统拆解如何作为一名AI工程师AI Engineer构建一个结构清晰、易于协作的智能体项目。我们将从核心概念入手逐步深入到项目架构设计、代码组织、团队协作流程并重点探讨如何对智能体进行系统化评估Evals。无论你是想将LLM能力集成到现有业务中的开发者还是希望构建自主智能体AI Agents的探索者这篇文章都将提供一套可直接复用的实战方案。1. 智能体Agents的核心概念与工程价值在深入代码之前我们必须统一认知什么是智能体为什么它需要专门的工程实践1.1 智能体是什么超越简单API调用简单来说一个智能体是一个能够感知环境、进行决策并执行行动以实现目标的系统。在LLM的语境下智能体通常指一个由大语言模型驱动LLM-powered的程序它能够利用工具Tools、访问外部知识如代码库并在多轮交互中完成复杂任务。与一次性的提示词Prompt调用不同智能体具备几个关键特征状态管理State Management能够记住对话历史、中间结果和任务目标。工具使用Tool Use可以调用外部函数、API或数据库来获取信息或执行操作如运行代码、查询网络、操作文件。自主规划Planning能够将复杂目标拆解为可执行的子步骤序列。迭代与反思Iteration Reflection根据执行结果调整策略甚至进行自我改进Self-improving。1.2 为什么需要关注代码库与团队当智能体从演示原型Demo走向生产环境Production时挑战随之而来代码混乱提示词、工具函数、逻辑控制、状态存储混杂在一个文件中难以阅读和维护。协作困难团队成员不清楚如何添加新工具、修改提示词或理解智能体的决策逻辑。评估缺失无法量化智能体的表现迭代优化如同“黑箱”操作严重依赖人工测试。部署复杂如何将智能体集成到现有系统如Kubernetes集群并确保其稳定、可观测。因此AI Engineer的角色应运而生他们不仅需要理解AI模型更需要具备软件工程能力来构建可靠、可扩展、可协作的智能体系统。本文将聚焦于如何构建智能体的Codebases代码库以及适应Teams团队开发的工程实践。2. 环境准备与项目初始化我们将使用Python作为主要语言并围绕目前流行的智能体开发框架来构建。请注意AI领域工具迭代迅速本文重点阐述设计思想和通用模式具体版本请根据项目实际情况调整。2.1 基础环境与工具Python: 推荐使用 3.9 或 3.10 版本。包管理: 使用pip或更推荐的poetry/uv进行依赖管理。版本控制: Git 是团队协作的基石。LLM API: 准备一个LLM提供商的API密钥如OpenAI, Anthropic, 国内合规平台等。本文示例将使用OpenAI格式的API但架构是通用的。IDE: 任何你熟悉的代码编辑器如VSCode、PyCharm。Cursor或Claude for VS Code等AI编程助手能极大提升效率。2.2 初始化项目结构一个清晰的目录结构是良好代码库的开始。我们创建一个名为ai_agent_project的项目。mkdir ai_agent_project cd ai_agent_project使用poetry初始化项目如果未安装请先pip install poetrypoetry new . # 或者手动创建核心目录 mkdir -p src/agents src/tools src/evaluations tests docs touch README.md requirements.txt .env.example最终建议的项目结构如下ai_agent_project/ ├── pyproject.toml # 项目依赖和配置 (poetry) ├── README.md ├── .env # 环境变量本地不提交 ├── .env.example # 环境变量示例 ├── src/ │ ├── __init__.py │ ├── agents/ # 智能体核心逻辑 │ │ ├── __init__.py │ │ ├── base_agent.py # 基础智能体类 │ │ ├── coding_agent.py # 代码智能体 │ │ └── orchestrator.py # 智能体编排器 │ ├── tools/ # 工具集 │ │ ├── __init__.py │ │ ├── base_tool.py │ │ ├── web_search.py │ │ ├── code_executor.py │ │ └── file_ops.py │ ├── memory/ # 记忆/状态管理 │ │ ├── __init__.py │ │ ├── base_memory.py │ │ └── conversation_memory.py │ ├── evaluations/ # 评估模块 │ │ ├── __init__.py │ │ ├── eval_runner.py │ │ └── test_cases/ │ └── utils/ # 通用工具函数 │ ├── __init__.py │ └── logger.py ├── tests/ # 单元测试和集成测试 ├── docs/ # 项目文档 ├── scripts/ # 部署或实用脚本 └── examples/ # 使用示例3. 构建智能体代码库核心模块拆解接下来我们深入每个模块编写可复用的代码。3.1 定义基础工具Tools工具是智能体的“手脚”。所有工具应继承一个基类确保接口统一。# 文件路径src/tools/base_tool.py from abc import ABC, abstractmethod from typing import Any, Dict, Optional from pydantic import BaseModel, Field class ToolInputSchema(BaseModel): 工具输入参数的Pydantic模型用于验证和自描述。 # 示例字段具体由工具定义 query: str Field(..., description搜索查询词) class BaseTool(ABC): 工具基类 name: str base_tool description: str 基础工具描述 input_schema: Optional[type[BaseModel]] None def __init__(self, **kwargs): # 可在此初始化工具所需资源如API客户端 pass abstractmethod async def execute(self, input_data: Dict[str, Any]) - Dict[str, Any]: 执行工具的核心方法 pass def get_schema(self) - Optional[type[BaseModel]]: 返回工具的输入模式用于让LLM理解如何调用。 return self.input_schema一个具体的网页搜索工具实现# 文件路径src/tools/web_search.py import asyncio from typing import Dict, Any # 假设使用DuckDuckGo搜索需安装pip install duckduckgo-search from duckduckgo_search import DDGS from .base_tool import BaseTool, ToolInputSchema from pydantic import Field class WebSearchInput(ToolInputSchema): query: str Field(..., description需要搜索的关键词或问题) class WebSearchTool(BaseTool): name web_search description 使用DuckDuckGo在互联网上搜索最新信息。 input_schema WebSearchInput def __init__(self, max_results: int 5): super().__init__() self.max_results max_results self.ddgs DDGS() async def execute(self, input_data: Dict[str, Any]) - Dict[str, Any]: schema self.input_schema(**input_data) query schema.query try: # 注意DDGS.search是同步方法在异步环境中需使用run_in_executor loop asyncio.get_event_loop() results await loop.run_in_executor( None, lambda: list(self.ddgs.text(query, max_resultsself.max_results)) ) # 格式化结果 formatted_results [] for r in results: formatted_results.append({ title: r.get(title, ), body: r.get(body, ), href: r.get(href, ) }) return { status: success, results: formatted_results, query: query } except Exception as e: return { status: error, message: f搜索执行失败: {str(e)} }3.2 实现记忆管理Memory记忆模块负责存储和检索对话历史、上下文信息。# 文件路径src/memory/conversation_memory.py from typing import List, Dict, Any from .base_memory import BaseMemory class ConversationMemory(BaseMemory): 基于列表的简单对话记忆 def __init__(self, max_turns: int 20): self.messages: List[Dict[str, Any]] [] self.max_turns max_turns def add_message(self, role: str, content: str, **kwargs): 添加一条消息到历史记录。 message {role: role, content: content, **kwargs} self.messages.append(message) # 限制历史长度防止上下文过长 if len(self.messages) self.max_turns * 2: # 假设每轮包含user和assistant self.messages self.messages[-self.max_turns*2:] def get_context(self, recent_n: int 10) - List[Dict[str, Any]]: 获取最近的N条消息作为上下文。 return self.messages[-recent_n*2:] if recent_n 0 else self.messages.copy() def clear(self): 清空记忆。 self.messages.clear()3.3 构建基础智能体Base Agent这是智能体的核心大脑整合了LLM调用、工具选择和记忆。# 文件路径src/agents/base_agent.py import json import asyncio from typing import List, Dict, Any, Optional from openai import AsyncOpenAI # 或其他兼容OpenAI API的客户端 from src.tools.base_tool import BaseTool from src.memory.conversation_memory import ConversationMemory class BaseAgent: 基础智能体类封装了与LLM的交互和工具调用循环。 def __init__( self, llm_client: Any, system_prompt: str, tools: List[BaseTool], memory: Optional[ConversationMemory] None ): self.llm_client llm_client self.system_prompt system_prompt self.tools {tool.name: tool for tool in tools} self.memory memory or ConversationMemory() # 将工具描述格式化供LLM理解 self._formatted_tools self._format_tools_for_llm() def _format_tools_for_llm(self) - List[Dict[str, Any]]: 将工具列表转换为OpenAI工具调用格式。 formatted [] for tool in self.tools.values(): schema tool.get_schema() if schema: # 构建符合OpenAI Function Calling格式的描述 properties {k: {type: string, description: v.field_info.description} for k, v in schema.model_fields.items()} formatted.append({ type: function, function: { name: tool.name, description: tool.description, parameters: { type: object, properties: properties, required: list(schema.model_fields.keys()) } } }) return formatted async def process_query(self, user_input: str) - Dict[str, Any]: 处理用户输入的主循环。 # 1. 将用户输入加入记忆 self.memory.add_message(user, user_input) # 2. 准备上下文消息 messages [{role: system, content: self.system_prompt}] messages.extend(self.memory.get_context()) # 3. 调用LLM允许其选择工具 response await self.llm_client.chat.completions.create( modelgpt-4-turbo-preview, # 根据实际情况选择模型 messagesmessages, toolsself._formatted_tools if self._formatted_tools else None, tool_choiceauto if self._formatted_tools else None, ) message response.choices[0].message tool_calls message.tool_calls # 4. 处理LLM响应 final_response_content message.content or if tool_calls: # LLM要求调用工具 for tool_call in tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) if tool_name in self.tools: # 执行工具 tool_result await self.tools[tool_name].execute(tool_args) # 将工具执行结果作为消息加入上下文让LLM继续处理 messages.append(message) # 添加LLM的消息 messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(tool_result, ensure_asciiFalse), name: tool_name }) # 再次调用LLM让其基于工具结果生成最终回复 second_response await self.llm_client.chat.completions.create( modelgpt-4-turbo-preview, messagesmessages, ) final_response_content second_response.choices[0].message.content message second_response.choices[0].message # 更新最终消息 else: final_response_content f错误请求了未知工具 {tool_name}。 # 5. 将智能体的最终回复加入记忆 self.memory.add_message(assistant, final_response_content) # 6. 返回结果 return { response: final_response_content, tool_calls_used: bool(tool_calls), memory_length: len(self.memory.messages) }4. 完整实战案例构建一个代码分析智能体现在我们整合上述模块创建一个能够分析GitHub仓库代码的智能体。这个智能体需要用到代码读取工具。4.1 创建代码读取工具首先添加一个从本地文件系统模拟从仓库克隆的代码读取代码的工具。# 文件路径src/tools/code_reader.py import os from pathlib import Path from typing import Dict, Any from .base_tool import BaseTool, ToolInputSchema from pydantic import Field, field_validator class CodeReaderInput(ToolInputSchema): file_path: str Field(..., description相对于项目根目录的代码文件路径) max_lines: int Field(100, description最大读取行数防止文件过大) field_validator(file_path) classmethod def validate_path(cls, v): # 简单的路径安全校验防止目录遍历攻击 if .. in v or v.startswith(/): raise ValueError(文件路径不安全或不允许。) return v class CodeReaderTool(BaseTool): name read_code_file description 读取指定路径的源代码文件内容。 input_schema CodeReaderInput def __init__(self, base_code_dir: str ./example_code): super().__init__() self.base_dir Path(base_code_dir).resolve() async def execute(self, input_data: Dict[str, Any]) - Dict[str, Any]: schema self.input_schema(**input_data) file_path schema.file_path max_lines schema.max_lines full_path self.base_dir / file_path try: if not full_path.exists(): return {status: error, message: f文件不存在: {file_path}} if not full_path.is_file(): return {status: error, message: f路径不是文件: {file_path}} # 安全检查确保文件在允许的基目录下 if not str(full_path).startswith(str(self.base_dir)): return {status: error, message: 访问越界。} with open(full_path, r, encodingutf-8, errorsignore) as f: lines [] for i, line in enumerate(f): if i max_lines: lines.append(f... (文件超过{max_lines}行已截断)) break lines.append(line) content .join(lines) return { status: success, content: content, file_path: file_path, lines_read: len(lines) if len(lines) max_lines else max_lines } except Exception as e: return {status: error, message: f读取文件失败: {str(e)}}4.2 创建智能体并运行创建一个主程序来实例化和运行我们的代码分析智能体。# 文件路径examples/run_code_agent.py import asyncio import os from dotenv import load_dotenv from openai import AsyncOpenAI from src.agents.base_agent import BaseAgent from src.tools.web_search import WebSearchTool from src.tools.code_reader import CodeReaderTool # 加载环境变量如OPENAI_API_KEY load_dotenv() async def main(): # 1. 初始化LLM客户端 client AsyncOpenAI(api_keyos.getenv(OPENAI_API_KEY)) # 2. 定义系统提示词明确智能体的角色和能力 system_prompt 你是一个专业的代码分析助手。你可以读取用户指定的源代码文件并对其进行分析。 你可以回答关于代码结构、函数功能、潜在问题等方面的问题。 如果遇到你不确定的概念你可以使用网络搜索工具获取最新信息。 请专注于提供准确、有帮助的代码分析。 # 3. 实例化工具 search_tool WebSearchTool(max_results3) code_reader_tool CodeReaderTool(base_code_dir./example_code) # 假设有个示例代码目录 # 4. 创建智能体 agent BaseAgent( llm_clientclient, system_promptsystem_prompt, tools[search_tool, code_reader_tool] ) # 5. 运行交互循环 print(代码分析智能体已启动。输入 quit 退出。) while True: try: user_input input(\n用户: ).strip() if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input: continue print(智能体思考中...) response await agent.process_query(user_input) print(f\n助手: {response[response]}) if response[tool_calls_used]: print((本次回答使用了工具)) except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f\n发生错误: {e}) if __name__ __main__: asyncio.run(main())4.3 准备示例代码并运行在项目根目录创建example_code文件夹并放入一个示例Python文件。mkdir -p example_code cat example_code/example.py EOF 这是一个示例计算器模块。 def add(a: float, b: float) - float: 返回两个数的和。 return a b def multiply(a: float, b: float) - float: 返回两个数的乘积。 return a * b def main(): 主函数演示功能。 print(计算器演示) result add(5, 3) print(f5 3 {result}) result2 multiply(4, 2) print(f4 * 2 {result2}) if __name__ __main__: main() EOF安装依赖并运行# 安装核心依赖 pip install openai duckduckgo-search python-dotenv pydantic # 设置你的API密钥到.env文件 echo OPENAI_API_KEYyour_api_key_here .env # 运行智能体 python examples/run_code_agent.py交互示例用户: 请读取并分析 example.py 这个文件。 智能体思考中... 助手: 我已经读取了 example.py 文件。这是一个简单的Python计算器模块包含以下内容 1. **函数 add(a: float, b: float) - float**: 接收两个浮点数返回它们的和。有清晰的类型注解和文档字符串。 2. **函数 multiply(a: float, b: float) - float**: 接收两个浮点数返回它们的乘积。同样有良好的文档。 3. **函数 main()**: 一个演示函数调用 add 和 multiply 并打印结果。 4. **模块入口**: 使用 if __name__ __main__: 来确保演示代码只在直接运行该脚本时执行。 **代码质量评价** - 优点结构清晰有文档字符串和类型提示符合Python PEP 8风格。 - 潜在改进可以增加错误处理例如输入非数字或者将函数扩展到更复杂的数学运算。这是一个很好的教学示例。 (本次回答使用了工具)5. 面向团队的工程实践协作、评估与部署个人项目可以快速迭代但团队协作需要规范、流程和度量。5.1 代码库协作规范清晰的模块化如前所示按agents、tools、memory、evaluations分离关注点。统一的接口所有工具继承BaseTool所有记忆模块继承BaseMemory确保新成员能快速添加功能。详尽的文档每个模块、类、重要函数都应包含Docstring。使用docs目录存放架构设计、API说明和 onboarding 指南。配置化管理将模型名称、API端点、超时时间、工具开关等提取到配置文件如config.yaml或环境变量中避免硬编码。版本控制提示词将重要的系统提示词system_prompt保存在版本控制的文件中如prompts/目录方便追溯和对比优化。5.2 智能体评估Demystifying Evals for AI Agents评估是衡量智能体性能、驱动迭代的关键。不能只靠人工测试。一个简单的评估运行器示例# 文件路径src/evaluations/eval_runner.py import asyncio import json from typing import List, Dict, Any from src.agents.base_agent import BaseAgent class EvaluationRunner: 运行评估测试套件。 def __init__(self, agent: BaseAgent, test_cases_path: str): self.agent agent with open(test_cases_path, r, encodingutf-8) as f: self.test_cases: List[Dict[str, Any]] json.load(f) async def run_single(self, test_case: Dict[str, Any]) - Dict[str, Any]: 运行单个测试用例。 query test_case[query] expected_keywords test_case.get(expected_keywords, []) # 简单关键词匹配 max_tool_calls test_case.get(max_tool_calls, 5) original_memory_len len(self.agent.memory.messages) self.agent.memory.clear() # 每个测试用例使用干净的记忆 try: response_data await self.agent.process_query(query) actual_response response_data[response] # 简单的评估逻辑检查响应中是否包含预期关键词 keyword_score 0 for keyword in expected_keywords: if keyword.lower() in actual_response.lower(): keyword_score 1 score keyword_score / len(expected_keywords) if expected_keywords else 0.0 return { test_id: test_case[id], query: query, expected_keywords: expected_keywords, actual_response: actual_response, score: score, passed: score 0.8, # 假设80%关键词匹配即为通过 tool_calls_used: response_data[tool_calls_used], error: None } except Exception as e: return { test_id: test_case[id], query: query, error: str(e), passed: False, score: 0.0 } finally: # 恢复记忆可选 while len(self.agent.memory.messages) original_memory_len: self.agent.memory.messages.pop() async def run_all(self) - List[Dict[str, Any]]: 运行所有测试用例。 results [] for test_case in self.test_cases: result await self.run_single(test_case) results.append(result) return results def generate_report(self, results: List[Dict[str, Any]]) - Dict[str, Any]: 生成评估报告。 total len(results) passed sum(1 for r in results if r.get(passed, False)) avg_score sum(r.get(score, 0) for r in results) / total if total 0 else 0 return { total_tests: total, passed_tests: passed, pass_rate: passed / total if total 0 else 0, average_score: avg_score, detailed_results: results }创建测试用例文件// 文件路径src/evaluations/test_cases/code_analysis_tests.json [ { id: test_01, query: 请分析 example.py 中 add 函数的作用。, expected_keywords: [add, 和, 相加, float, 返回], category: code_analysis }, { id: test_02, query: 这个文件里有什么函数, expected_keywords: [add, multiply, main, 函数], category: code_analysis }, { id: test_03, query: Python中的装饰器是什么, expected_keywords: [, decorator, 修饰, 函数, 语法糖], category: knowledge_query, max_tool_calls: 2 // 此问题可能触发搜索工具 } ]运行评估# 文件路径examples/run_evaluation.py import asyncio import json from src.agents.base_agent import BaseAgent from src.tools.code_reader import CodeReaderTool from src.evaluations.eval_runner import EvaluationRunner # ... 省略初始化agent的代码同run_code_agent.py async def main(): agent ... # 初始化agent同上 runner EvaluationRunner( agentagent, test_cases_pathsrc/evaluations/test_cases/code_analysis_tests.json ) results await runner.run_all() report runner.generate_report(results) print(json.dumps(report, indent2, ensure_asciiFalse)) print(f\n通过率: {report[pass_rate]:.2%}) print(f平均分: {report[average_score]:.2f}) if __name__ __main__: asyncio.run(main())5.3 部署与运维考量当智能体需要服务化或集成到更大系统时API服务化使用FastAPI或Flask将智能体封装成HTTP API提供/chat或/process端点。配置隔离使用环境变量或配置中心管理不同环境开发、测试、生产的API密钥、模型参数。日志与监控在关键节点工具调用、LLM请求、最终响应记录结构化日志。集成监控如Prometheus跟踪耗时、错误率和Token使用量。错误处理与降级为LLM调用和工具调用设置超时和重试机制。在LLM服务不可用时提供降级策略如返回缓存答案或友好错误信息。在Kubernetes中运行将智能体服务容器化利用K8s的Deployment进行部署配合HPA实现弹性伸缩。需要特别注意智能体通常是有状态的记忆在多副本时需要将会话状态外存到Redis等共享存储中。6. 常见问题与排查思路在开发智能体过程中你会遇到一些典型问题。问题现象可能原因排查思路与解决方案LLM不调用工具1. 工具描述不清晰。2. 系统提示词未鼓励使用工具。3. LLM模型不支持或未开启函数调用。1. 检查description和parameters的描述是否准确易懂。2. 在system_prompt中明确指令如“你必须使用提供的工具来获取信息”。3. 确认使用的模型如gpt-3.5-turbo与gpt-4-turbo-preview是否支持并启用了tools参数。工具调用参数错误1. LLM生成的参数格式不符合Pydantic模型。2. 参数类型不匹配。1. 在工具execute方法开始时使用schema self.input_schema(**input_data)进行验证和转换。2. 在工具描述中明确参数类型和示例。智能体陷入循环或动作过多1. 缺乏停止条件。2. 工具执行结果误导LLM再次调用工具。1. 在BaseAgent的主循环中设置最大工具调用次数限制。2. 优化工具返回的结果格式使其更简洁、信息明确。记忆上下文过长导致Token超限或性能下降1. 对话轮次过多未清理。2. 记忆模块未做摘要或压缩。1. 在ConversationMemory中实现max_turns限制。2. 实现记忆摘要功能定期让LLM总结之前的对话用摘要替换详细历史。评估结果不稳定1. 评估标准过于主观如关键词匹配。2. LLM生成具有随机性。1. 采用更复杂的评估方法如使用另一个LLM作为裁判LLM-as-a-Judge或基于真实业务指标任务完成率。2. 在评估中设置固定的随机种子并对每个测试用例运行多次取平均。7. 最佳实践与工程建议提示词工程化将提示词视为代码。使用模板引擎如Jinja2管理提示词将变量如工具列表、当前日期动态注入。为不同任务创建专用的提示词模板文件。工具设计原则单一职责每个工具只做一件事并做好。健壮性工具内部要有充分的错误处理和边界检查避免因单个工具失败导致整个智能体崩溃。安全性对用户输入进行严格的验证和清理防止路径遍历、命令注入等攻击。测试驱动开发为每个工具编写单元测试。为智能体的核心流程编写集成测试。评估套件Evals应作为CI/CD流水线的一部分在每次代码变更后自动运行防止性能回归。成本与性能监控记录每次LLM调用的模型、Token使用量和耗时。设置预算告警。对于内部工具考虑使用缓存如对相似的代码分析请求缓存结果来减少不必要的LLM调用和工具执行。团队知识共享建立团队内部的“智能体模式库”收集和分享有效的提示词模板、工具设计模式、评估用例。定期进行代码评审特别关注提示词和工具描述的清晰度。构建一个面向团队、可维护、可评估的智能体代码库是将AI从炫酷演示转化为实际生产力的关键一步。它要求开发者兼具AI理解力和软件工程素养。本文提供的架构和代码只是一个起点你可以在此基础上根据具体业务需求扩展更复杂的工具如数据库查询、内部API调用、实现更高级的记忆机制如向量数据库检索并设计更科学的评估体系。记住好的智能体系统是迭代出来的。从一个小而美的原型开始建立评估基线然后在一个坚实的代码库基础上与你的团队一起持续改进。 AI智能体工程化实战：构建可维护、可协作的代码库与评估体系
你好我是专注于技术实战分享的博主。在探索AI工程化落地的过程中我发现一个普遍痛点许多开发者对“智能体Agents”的概念充满热情但一到具体实施就卡在了如何构建可维护的代码库、如何与团队协作以及如何评估其效果上。网上资料要么过于理论化要么是零散的代码片段缺乏一套从开发到部署、从个人到团队的完整工程实践指南。本文将围绕“智能体、代码库与团队”这一核心主题为你系统拆解如何作为一名AI工程师AI Engineer构建一个结构清晰、易于协作的智能体项目。我们将从核心概念入手逐步深入到项目架构设计、代码组织、团队协作流程并重点探讨如何对智能体进行系统化评估Evals。无论你是想将LLM能力集成到现有业务中的开发者还是希望构建自主智能体AI Agents的探索者这篇文章都将提供一套可直接复用的实战方案。1. 智能体Agents的核心概念与工程价值在深入代码之前我们必须统一认知什么是智能体为什么它需要专门的工程实践1.1 智能体是什么超越简单API调用简单来说一个智能体是一个能够感知环境、进行决策并执行行动以实现目标的系统。在LLM的语境下智能体通常指一个由大语言模型驱动LLM-powered的程序它能够利用工具Tools、访问外部知识如代码库并在多轮交互中完成复杂任务。与一次性的提示词Prompt调用不同智能体具备几个关键特征状态管理State Management能够记住对话历史、中间结果和任务目标。工具使用Tool Use可以调用外部函数、API或数据库来获取信息或执行操作如运行代码、查询网络、操作文件。自主规划Planning能够将复杂目标拆解为可执行的子步骤序列。迭代与反思Iteration Reflection根据执行结果调整策略甚至进行自我改进Self-improving。1.2 为什么需要关注代码库与团队当智能体从演示原型Demo走向生产环境Production时挑战随之而来代码混乱提示词、工具函数、逻辑控制、状态存储混杂在一个文件中难以阅读和维护。协作困难团队成员不清楚如何添加新工具、修改提示词或理解智能体的决策逻辑。评估缺失无法量化智能体的表现迭代优化如同“黑箱”操作严重依赖人工测试。部署复杂如何将智能体集成到现有系统如Kubernetes集群并确保其稳定、可观测。因此AI Engineer的角色应运而生他们不仅需要理解AI模型更需要具备软件工程能力来构建可靠、可扩展、可协作的智能体系统。本文将聚焦于如何构建智能体的Codebases代码库以及适应Teams团队开发的工程实践。2. 环境准备与项目初始化我们将使用Python作为主要语言并围绕目前流行的智能体开发框架来构建。请注意AI领域工具迭代迅速本文重点阐述设计思想和通用模式具体版本请根据项目实际情况调整。2.1 基础环境与工具Python: 推荐使用 3.9 或 3.10 版本。包管理: 使用pip或更推荐的poetry/uv进行依赖管理。版本控制: Git 是团队协作的基石。LLM API: 准备一个LLM提供商的API密钥如OpenAI, Anthropic, 国内合规平台等。本文示例将使用OpenAI格式的API但架构是通用的。IDE: 任何你熟悉的代码编辑器如VSCode、PyCharm。Cursor或Claude for VS Code等AI编程助手能极大提升效率。2.2 初始化项目结构一个清晰的目录结构是良好代码库的开始。我们创建一个名为ai_agent_project的项目。mkdir ai_agent_project cd ai_agent_project使用poetry初始化项目如果未安装请先pip install poetrypoetry new . # 或者手动创建核心目录 mkdir -p src/agents src/tools src/evaluations tests docs touch README.md requirements.txt .env.example最终建议的项目结构如下ai_agent_project/ ├── pyproject.toml # 项目依赖和配置 (poetry) ├── README.md ├── .env # 环境变量本地不提交 ├── .env.example # 环境变量示例 ├── src/ │ ├── __init__.py │ ├── agents/ # 智能体核心逻辑 │ │ ├── __init__.py │ │ ├── base_agent.py # 基础智能体类 │ │ ├── coding_agent.py # 代码智能体 │ │ └── orchestrator.py # 智能体编排器 │ ├── tools/ # 工具集 │ │ ├── __init__.py │ │ ├── base_tool.py │ │ ├── web_search.py │ │ ├── code_executor.py │ │ └── file_ops.py │ ├── memory/ # 记忆/状态管理 │ │ ├── __init__.py │ │ ├── base_memory.py │ │ └── conversation_memory.py │ ├── evaluations/ # 评估模块 │ │ ├── __init__.py │ │ ├── eval_runner.py │ │ └── test_cases/ │ └── utils/ # 通用工具函数 │ ├── __init__.py │ └── logger.py ├── tests/ # 单元测试和集成测试 ├── docs/ # 项目文档 ├── scripts/ # 部署或实用脚本 └── examples/ # 使用示例3. 构建智能体代码库核心模块拆解接下来我们深入每个模块编写可复用的代码。3.1 定义基础工具Tools工具是智能体的“手脚”。所有工具应继承一个基类确保接口统一。# 文件路径src/tools/base_tool.py from abc import ABC, abstractmethod from typing import Any, Dict, Optional from pydantic import BaseModel, Field class ToolInputSchema(BaseModel): 工具输入参数的Pydantic模型用于验证和自描述。 # 示例字段具体由工具定义 query: str Field(..., description搜索查询词) class BaseTool(ABC): 工具基类 name: str base_tool description: str 基础工具描述 input_schema: Optional[type[BaseModel]] None def __init__(self, **kwargs): # 可在此初始化工具所需资源如API客户端 pass abstractmethod async def execute(self, input_data: Dict[str, Any]) - Dict[str, Any]: 执行工具的核心方法 pass def get_schema(self) - Optional[type[BaseModel]]: 返回工具的输入模式用于让LLM理解如何调用。 return self.input_schema一个具体的网页搜索工具实现# 文件路径src/tools/web_search.py import asyncio from typing import Dict, Any # 假设使用DuckDuckGo搜索需安装pip install duckduckgo-search from duckduckgo_search import DDGS from .base_tool import BaseTool, ToolInputSchema from pydantic import Field class WebSearchInput(ToolInputSchema): query: str Field(..., description需要搜索的关键词或问题) class WebSearchTool(BaseTool): name web_search description 使用DuckDuckGo在互联网上搜索最新信息。 input_schema WebSearchInput def __init__(self, max_results: int 5): super().__init__() self.max_results max_results self.ddgs DDGS() async def execute(self, input_data: Dict[str, Any]) - Dict[str, Any]: schema self.input_schema(**input_data) query schema.query try: # 注意DDGS.search是同步方法在异步环境中需使用run_in_executor loop asyncio.get_event_loop() results await loop.run_in_executor( None, lambda: list(self.ddgs.text(query, max_resultsself.max_results)) ) # 格式化结果 formatted_results [] for r in results: formatted_results.append({ title: r.get(title, ), body: r.get(body, ), href: r.get(href, ) }) return { status: success, results: formatted_results, query: query } except Exception as e: return { status: error, message: f搜索执行失败: {str(e)} }3.2 实现记忆管理Memory记忆模块负责存储和检索对话历史、上下文信息。# 文件路径src/memory/conversation_memory.py from typing import List, Dict, Any from .base_memory import BaseMemory class ConversationMemory(BaseMemory): 基于列表的简单对话记忆 def __init__(self, max_turns: int 20): self.messages: List[Dict[str, Any]] [] self.max_turns max_turns def add_message(self, role: str, content: str, **kwargs): 添加一条消息到历史记录。 message {role: role, content: content, **kwargs} self.messages.append(message) # 限制历史长度防止上下文过长 if len(self.messages) self.max_turns * 2: # 假设每轮包含user和assistant self.messages self.messages[-self.max_turns*2:] def get_context(self, recent_n: int 10) - List[Dict[str, Any]]: 获取最近的N条消息作为上下文。 return self.messages[-recent_n*2:] if recent_n 0 else self.messages.copy() def clear(self): 清空记忆。 self.messages.clear()3.3 构建基础智能体Base Agent这是智能体的核心大脑整合了LLM调用、工具选择和记忆。# 文件路径src/agents/base_agent.py import json import asyncio from typing import List, Dict, Any, Optional from openai import AsyncOpenAI # 或其他兼容OpenAI API的客户端 from src.tools.base_tool import BaseTool from src.memory.conversation_memory import ConversationMemory class BaseAgent: 基础智能体类封装了与LLM的交互和工具调用循环。 def __init__( self, llm_client: Any, system_prompt: str, tools: List[BaseTool], memory: Optional[ConversationMemory] None ): self.llm_client llm_client self.system_prompt system_prompt self.tools {tool.name: tool for tool in tools} self.memory memory or ConversationMemory() # 将工具描述格式化供LLM理解 self._formatted_tools self._format_tools_for_llm() def _format_tools_for_llm(self) - List[Dict[str, Any]]: 将工具列表转换为OpenAI工具调用格式。 formatted [] for tool in self.tools.values(): schema tool.get_schema() if schema: # 构建符合OpenAI Function Calling格式的描述 properties {k: {type: string, description: v.field_info.description} for k, v in schema.model_fields.items()} formatted.append({ type: function, function: { name: tool.name, description: tool.description, parameters: { type: object, properties: properties, required: list(schema.model_fields.keys()) } } }) return formatted async def process_query(self, user_input: str) - Dict[str, Any]: 处理用户输入的主循环。 # 1. 将用户输入加入记忆 self.memory.add_message(user, user_input) # 2. 准备上下文消息 messages [{role: system, content: self.system_prompt}] messages.extend(self.memory.get_context()) # 3. 调用LLM允许其选择工具 response await self.llm_client.chat.completions.create( modelgpt-4-turbo-preview, # 根据实际情况选择模型 messagesmessages, toolsself._formatted_tools if self._formatted_tools else None, tool_choiceauto if self._formatted_tools else None, ) message response.choices[0].message tool_calls message.tool_calls # 4. 处理LLM响应 final_response_content message.content or if tool_calls: # LLM要求调用工具 for tool_call in tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) if tool_name in self.tools: # 执行工具 tool_result await self.tools[tool_name].execute(tool_args) # 将工具执行结果作为消息加入上下文让LLM继续处理 messages.append(message) # 添加LLM的消息 messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(tool_result, ensure_asciiFalse), name: tool_name }) # 再次调用LLM让其基于工具结果生成最终回复 second_response await self.llm_client.chat.completions.create( modelgpt-4-turbo-preview, messagesmessages, ) final_response_content second_response.choices[0].message.content message second_response.choices[0].message # 更新最终消息 else: final_response_content f错误请求了未知工具 {tool_name}。 # 5. 将智能体的最终回复加入记忆 self.memory.add_message(assistant, final_response_content) # 6. 返回结果 return { response: final_response_content, tool_calls_used: bool(tool_calls), memory_length: len(self.memory.messages) }4. 完整实战案例构建一个代码分析智能体现在我们整合上述模块创建一个能够分析GitHub仓库代码的智能体。这个智能体需要用到代码读取工具。4.1 创建代码读取工具首先添加一个从本地文件系统模拟从仓库克隆的代码读取代码的工具。# 文件路径src/tools/code_reader.py import os from pathlib import Path from typing import Dict, Any from .base_tool import BaseTool, ToolInputSchema from pydantic import Field, field_validator class CodeReaderInput(ToolInputSchema): file_path: str Field(..., description相对于项目根目录的代码文件路径) max_lines: int Field(100, description最大读取行数防止文件过大) field_validator(file_path) classmethod def validate_path(cls, v): # 简单的路径安全校验防止目录遍历攻击 if .. in v or v.startswith(/): raise ValueError(文件路径不安全或不允许。) return v class CodeReaderTool(BaseTool): name read_code_file description 读取指定路径的源代码文件内容。 input_schema CodeReaderInput def __init__(self, base_code_dir: str ./example_code): super().__init__() self.base_dir Path(base_code_dir).resolve() async def execute(self, input_data: Dict[str, Any]) - Dict[str, Any]: schema self.input_schema(**input_data) file_path schema.file_path max_lines schema.max_lines full_path self.base_dir / file_path try: if not full_path.exists(): return {status: error, message: f文件不存在: {file_path}} if not full_path.is_file(): return {status: error, message: f路径不是文件: {file_path}} # 安全检查确保文件在允许的基目录下 if not str(full_path).startswith(str(self.base_dir)): return {status: error, message: 访问越界。} with open(full_path, r, encodingutf-8, errorsignore) as f: lines [] for i, line in enumerate(f): if i max_lines: lines.append(f... (文件超过{max_lines}行已截断)) break lines.append(line) content .join(lines) return { status: success, content: content, file_path: file_path, lines_read: len(lines) if len(lines) max_lines else max_lines } except Exception as e: return {status: error, message: f读取文件失败: {str(e)}}4.2 创建智能体并运行创建一个主程序来实例化和运行我们的代码分析智能体。# 文件路径examples/run_code_agent.py import asyncio import os from dotenv import load_dotenv from openai import AsyncOpenAI from src.agents.base_agent import BaseAgent from src.tools.web_search import WebSearchTool from src.tools.code_reader import CodeReaderTool # 加载环境变量如OPENAI_API_KEY load_dotenv() async def main(): # 1. 初始化LLM客户端 client AsyncOpenAI(api_keyos.getenv(OPENAI_API_KEY)) # 2. 定义系统提示词明确智能体的角色和能力 system_prompt 你是一个专业的代码分析助手。你可以读取用户指定的源代码文件并对其进行分析。 你可以回答关于代码结构、函数功能、潜在问题等方面的问题。 如果遇到你不确定的概念你可以使用网络搜索工具获取最新信息。 请专注于提供准确、有帮助的代码分析。 # 3. 实例化工具 search_tool WebSearchTool(max_results3) code_reader_tool CodeReaderTool(base_code_dir./example_code) # 假设有个示例代码目录 # 4. 创建智能体 agent BaseAgent( llm_clientclient, system_promptsystem_prompt, tools[search_tool, code_reader_tool] ) # 5. 运行交互循环 print(代码分析智能体已启动。输入 quit 退出。) while True: try: user_input input(\n用户: ).strip() if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input: continue print(智能体思考中...) response await agent.process_query(user_input) print(f\n助手: {response[response]}) if response[tool_calls_used]: print((本次回答使用了工具)) except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f\n发生错误: {e}) if __name__ __main__: asyncio.run(main())4.3 准备示例代码并运行在项目根目录创建example_code文件夹并放入一个示例Python文件。mkdir -p example_code cat example_code/example.py EOF 这是一个示例计算器模块。 def add(a: float, b: float) - float: 返回两个数的和。 return a b def multiply(a: float, b: float) - float: 返回两个数的乘积。 return a * b def main(): 主函数演示功能。 print(计算器演示) result add(5, 3) print(f5 3 {result}) result2 multiply(4, 2) print(f4 * 2 {result2}) if __name__ __main__: main() EOF安装依赖并运行# 安装核心依赖 pip install openai duckduckgo-search python-dotenv pydantic # 设置你的API密钥到.env文件 echo OPENAI_API_KEYyour_api_key_here .env # 运行智能体 python examples/run_code_agent.py交互示例用户: 请读取并分析 example.py 这个文件。 智能体思考中... 助手: 我已经读取了 example.py 文件。这是一个简单的Python计算器模块包含以下内容 1. **函数 add(a: float, b: float) - float**: 接收两个浮点数返回它们的和。有清晰的类型注解和文档字符串。 2. **函数 multiply(a: float, b: float) - float**: 接收两个浮点数返回它们的乘积。同样有良好的文档。 3. **函数 main()**: 一个演示函数调用 add 和 multiply 并打印结果。 4. **模块入口**: 使用 if __name__ __main__: 来确保演示代码只在直接运行该脚本时执行。 **代码质量评价** - 优点结构清晰有文档字符串和类型提示符合Python PEP 8风格。 - 潜在改进可以增加错误处理例如输入非数字或者将函数扩展到更复杂的数学运算。这是一个很好的教学示例。 (本次回答使用了工具)5. 面向团队的工程实践协作、评估与部署个人项目可以快速迭代但团队协作需要规范、流程和度量。5.1 代码库协作规范清晰的模块化如前所示按agents、tools、memory、evaluations分离关注点。统一的接口所有工具继承BaseTool所有记忆模块继承BaseMemory确保新成员能快速添加功能。详尽的文档每个模块、类、重要函数都应包含Docstring。使用docs目录存放架构设计、API说明和 onboarding 指南。配置化管理将模型名称、API端点、超时时间、工具开关等提取到配置文件如config.yaml或环境变量中避免硬编码。版本控制提示词将重要的系统提示词system_prompt保存在版本控制的文件中如prompts/目录方便追溯和对比优化。5.2 智能体评估Demystifying Evals for AI Agents评估是衡量智能体性能、驱动迭代的关键。不能只靠人工测试。一个简单的评估运行器示例# 文件路径src/evaluations/eval_runner.py import asyncio import json from typing import List, Dict, Any from src.agents.base_agent import BaseAgent class EvaluationRunner: 运行评估测试套件。 def __init__(self, agent: BaseAgent, test_cases_path: str): self.agent agent with open(test_cases_path, r, encodingutf-8) as f: self.test_cases: List[Dict[str, Any]] json.load(f) async def run_single(self, test_case: Dict[str, Any]) - Dict[str, Any]: 运行单个测试用例。 query test_case[query] expected_keywords test_case.get(expected_keywords, []) # 简单关键词匹配 max_tool_calls test_case.get(max_tool_calls, 5) original_memory_len len(self.agent.memory.messages) self.agent.memory.clear() # 每个测试用例使用干净的记忆 try: response_data await self.agent.process_query(query) actual_response response_data[response] # 简单的评估逻辑检查响应中是否包含预期关键词 keyword_score 0 for keyword in expected_keywords: if keyword.lower() in actual_response.lower(): keyword_score 1 score keyword_score / len(expected_keywords) if expected_keywords else 0.0 return { test_id: test_case[id], query: query, expected_keywords: expected_keywords, actual_response: actual_response, score: score, passed: score 0.8, # 假设80%关键词匹配即为通过 tool_calls_used: response_data[tool_calls_used], error: None } except Exception as e: return { test_id: test_case[id], query: query, error: str(e), passed: False, score: 0.0 } finally: # 恢复记忆可选 while len(self.agent.memory.messages) original_memory_len: self.agent.memory.messages.pop() async def run_all(self) - List[Dict[str, Any]]: 运行所有测试用例。 results [] for test_case in self.test_cases: result await self.run_single(test_case) results.append(result) return results def generate_report(self, results: List[Dict[str, Any]]) - Dict[str, Any]: 生成评估报告。 total len(results) passed sum(1 for r in results if r.get(passed, False)) avg_score sum(r.get(score, 0) for r in results) / total if total 0 else 0 return { total_tests: total, passed_tests: passed, pass_rate: passed / total if total 0 else 0, average_score: avg_score, detailed_results: results }创建测试用例文件// 文件路径src/evaluations/test_cases/code_analysis_tests.json [ { id: test_01, query: 请分析 example.py 中 add 函数的作用。, expected_keywords: [add, 和, 相加, float, 返回], category: code_analysis }, { id: test_02, query: 这个文件里有什么函数, expected_keywords: [add, multiply, main, 函数], category: code_analysis }, { id: test_03, query: Python中的装饰器是什么, expected_keywords: [, decorator, 修饰, 函数, 语法糖], category: knowledge_query, max_tool_calls: 2 // 此问题可能触发搜索工具 } ]运行评估# 文件路径examples/run_evaluation.py import asyncio import json from src.agents.base_agent import BaseAgent from src.tools.code_reader import CodeReaderTool from src.evaluations.eval_runner import EvaluationRunner # ... 省略初始化agent的代码同run_code_agent.py async def main(): agent ... # 初始化agent同上 runner EvaluationRunner( agentagent, test_cases_pathsrc/evaluations/test_cases/code_analysis_tests.json ) results await runner.run_all() report runner.generate_report(results) print(json.dumps(report, indent2, ensure_asciiFalse)) print(f\n通过率: {report[pass_rate]:.2%}) print(f平均分: {report[average_score]:.2f}) if __name__ __main__: asyncio.run(main())5.3 部署与运维考量当智能体需要服务化或集成到更大系统时API服务化使用FastAPI或Flask将智能体封装成HTTP API提供/chat或/process端点。配置隔离使用环境变量或配置中心管理不同环境开发、测试、生产的API密钥、模型参数。日志与监控在关键节点工具调用、LLM请求、最终响应记录结构化日志。集成监控如Prometheus跟踪耗时、错误率和Token使用量。错误处理与降级为LLM调用和工具调用设置超时和重试机制。在LLM服务不可用时提供降级策略如返回缓存答案或友好错误信息。在Kubernetes中运行将智能体服务容器化利用K8s的Deployment进行部署配合HPA实现弹性伸缩。需要特别注意智能体通常是有状态的记忆在多副本时需要将会话状态外存到Redis等共享存储中。6. 常见问题与排查思路在开发智能体过程中你会遇到一些典型问题。问题现象可能原因排查思路与解决方案LLM不调用工具1. 工具描述不清晰。2. 系统提示词未鼓励使用工具。3. LLM模型不支持或未开启函数调用。1. 检查description和parameters的描述是否准确易懂。2. 在system_prompt中明确指令如“你必须使用提供的工具来获取信息”。3. 确认使用的模型如gpt-3.5-turbo与gpt-4-turbo-preview是否支持并启用了tools参数。工具调用参数错误1. LLM生成的参数格式不符合Pydantic模型。2. 参数类型不匹配。1. 在工具execute方法开始时使用schema self.input_schema(**input_data)进行验证和转换。2. 在工具描述中明确参数类型和示例。智能体陷入循环或动作过多1. 缺乏停止条件。2. 工具执行结果误导LLM再次调用工具。1. 在BaseAgent的主循环中设置最大工具调用次数限制。2. 优化工具返回的结果格式使其更简洁、信息明确。记忆上下文过长导致Token超限或性能下降1. 对话轮次过多未清理。2. 记忆模块未做摘要或压缩。1. 在ConversationMemory中实现max_turns限制。2. 实现记忆摘要功能定期让LLM总结之前的对话用摘要替换详细历史。评估结果不稳定1. 评估标准过于主观如关键词匹配。2. LLM生成具有随机性。1. 采用更复杂的评估方法如使用另一个LLM作为裁判LLM-as-a-Judge或基于真实业务指标任务完成率。2. 在评估中设置固定的随机种子并对每个测试用例运行多次取平均。7. 最佳实践与工程建议提示词工程化将提示词视为代码。使用模板引擎如Jinja2管理提示词将变量如工具列表、当前日期动态注入。为不同任务创建专用的提示词模板文件。工具设计原则单一职责每个工具只做一件事并做好。健壮性工具内部要有充分的错误处理和边界检查避免因单个工具失败导致整个智能体崩溃。安全性对用户输入进行严格的验证和清理防止路径遍历、命令注入等攻击。测试驱动开发为每个工具编写单元测试。为智能体的核心流程编写集成测试。评估套件Evals应作为CI/CD流水线的一部分在每次代码变更后自动运行防止性能回归。成本与性能监控记录每次LLM调用的模型、Token使用量和耗时。设置预算告警。对于内部工具考虑使用缓存如对相似的代码分析请求缓存结果来减少不必要的LLM调用和工具执行。团队知识共享建立团队内部的“智能体模式库”收集和分享有效的提示词模板、工具设计模式、评估用例。定期进行代码评审特别关注提示词和工具描述的清晰度。构建一个面向团队、可维护、可评估的智能体代码库是将AI从炫酷演示转化为实际生产力的关键一步。它要求开发者兼具AI理解力和软件工程素养。本文提供的架构和代码只是一个起点你可以在此基础上根据具体业务需求扩展更复杂的工具如数据库查询、内部API调用、实现更高级的记忆机制如向量数据库检索并设计更科学的评估体系。记住好的智能体系统是迭代出来的。从一个小而美的原型开始建立评估基线然后在一个坚实的代码库基础上与你的团队一起持续改进。