企业官网建设流程全解析

最近在探索AI智能体（Agent）的落地应用时，发现很多开发者对如何构建一个具备复杂技能（Skills）的智能体系统感到困惑。网上资料要么过于理论化，要么代码片段零散，难以整合成一个可运行、可扩展的项目。本文将基于一个名为“Agent Skills”的架构项目，手把手带你从零开始，构建一个具备多技能协同与深度研究（DeepResearch）能力的智能体系统。无论你是想入门AI Agent开发，还是希望将智能体技术应用到实际业务中，这篇实战教程都能提供一条清晰的路径和一套完整的代码。

1. 背景与核心概念：什么是Agent Skills架构？

在深入代码之前，我们有必要厘清几个核心概念。这能帮助我们在后续开发中做出正确的设计决策。

1.1 智能体（Agent）与技能（Skills）

智能体（Agent）在AI语境下，通常指一个能够感知环境、进行决策并执行行动以达成目标的自治软件实体。它不仅仅是调用一次大模型API，而是一个具备记忆、规划和工具使用能力的持续运行系统。

技能（Skills）是智能体能力的具象化单元。一个复杂的任务可以被拆解为多个基础技能的组合。例如，一个“市场分析Agent”可能由“网络搜索Skill”、“数据提取Skill”、“报告生成Skill”等构成。将功能模块化为Skill，极大地提升了系统的可维护性和可扩展性。

1.2 Agent Skills 架构的核心思想

Agent Skills架构是一种设计模式，它强调：

1. 模块化：每个Skill是独立的、功能单一的模块，通过清晰的接口与Agent核心交互。
2. 可编排：Agent核心（或称“大脑”）负责根据目标，动态地选择、组合和调用不同的Skills。
3. 状态管理：Agent需要维护对话历史、任务上下文、工具调用结果等状态，供后续决策使用。
4. 工具集成：Skills的本质是让Agent能够安全、有效地使用外部工具（如搜索引擎、数据库、API）。

1.3 DeepResearch：深度研究能力

DeepResearch（深度研究）是本项目要实现的一个高级技能示例。它模拟了一个研究员的完整工作流：理解复杂问题 -> 拆解子问题 -> 并行搜索与信息收集 -> 综合分析与验证 -> 产出结构化报告。这远非一次简单的网络搜索，而是体现了智能体在复杂任务上的规划、执行与反思能力。

1.4 与相关概念的区分

• Skills vs. MCP (Model Context Protocol)：MCP是Anthropic提出的一种协议，用于让大模型安全、标准化地使用服务器端工具和资源。你可以把Skills看作是应用层的功能实现，而MCP是底层通信协议的一种选择。我们的项目侧重于应用层Skill的构建与编排，可以兼容不同的底层协议（如OpenAI的Function Calling，或MCP）。
• Agent vs. 简单提示工程：传统的提示工程是静态的、一次性的交互。Agent是动态的、有状态的，它可以根据历史交互调整策略，主动规划步骤，形成一个闭环。

2. 环境准备与版本说明

我们将使用Python作为开发语言，并依托于LangChain这一流行的AI应用开发框架来构建我们的Agent系统。LangChain提供了丰富的组件来简化Agent、Tools、Memory的开发。

核心环境要求：

• 操作系统：Windows 10/11, macOS, 或 Linux (Ubuntu 20.04+)。本文命令以macOS/Linux的bash为例，Windows用户可在PowerShell或WSL中运行。
• Python：版本 3.9 或 3.10。推荐使用3.10以获得最佳兼容性。

  python --version # 检查版本

• 包管理工具：pip或poetry。本文使用pip。
• 大模型API：你需要一个可用的大模型API密钥，例如：
  • OpenAI GPT-3.5/4
  • Anthropic Claude
  • 或国内可用的智谱、月之暗面等平台的API。
  • 注意：我们将使用OpenAI格式的接口进行演示，其他兼容API（如OpenRouter、Ollama本地模型）只需调整基础URL和API Key即可。

项目初始化：

1. 创建项目目录并进入：

   mkdir agent-skills-tutorial && cd agent-skills-tutorial

2. 创建虚拟环境（强烈推荐）：

   python -m venv venv # 激活虚拟环境 # macOS/Linux: source venv/bin/activate # Windows: # venv\Scripts\activate

3. 创建核心依赖文件requirements.txt：

   langchain>=0.1.0 langchain-openai>=0.0.5 # 用于集成OpenAI模型 langchain-community>=0.0.10 # 包含许多社区工具和集成 python-dotenv>=1.0.0 # 用于管理环境变量 duckduckgo-search>=4.1.0 # 用于实现搜索技能 beautifulsoup4>=4.12.0 # 用于网页内容解析 requests>=2.31.0 # 用于HTTP请求 pydantic>=2.0.0 # 用于数据验证和设置管理

4. 安装依赖：

   pip install -r requirements.txt

5. 创建环境变量文件.env，用于安全存储API密钥：

   # .env OPENAI_API_KEY="your-openai-api-key-here" # 如果你使用其他模型，例如Ollama # OLLAMA_BASE_URL="http://localhost:11434" # OLLAMA_MODEL="llama3.2"

   重要：请将.env添加到.gitignore文件中，避免密钥泄露。

3. 项目架构设计与核心模块拆解

在写代码前，我们先设计整个项目的结构。一个清晰的架构是项目可维护性的基石。

agent-skills-tutorial/ ├── .env # 环境变量（密钥等） ├── .gitignore ├── requirements.txt # Python依赖 ├── main.py # 主程序入口 ├── core/ # 核心Agent逻辑 │ ├── __init__.py │ ├── agent.py # Agent核心类定义 │ └── memory.py # 记忆管理（对话历史） ├── skills/ # 技能模块目录 │ ├── __init__.py │ ├── base_skill.py # 技能基类 │ ├── web_search.py # 网络搜索技能 │ ├── calculator.py # 计算器技能 │ └── deep_research.py # 深度研究技能（复合技能） ├── tools/ # 底层工具（可被技能调用） │ ├── __init__.py │ └── web_scraper.py # 网页抓取工具 └── utils/ # 工具函数 ├── __init__.py └── config.py # 配置加载

各模块职责说明：

• core/agent.py：定义Agent类。它是系统的大脑，负责加载技能、解析用户意图、维护执行状态、并调用相应的技能。
• core/memory.py：管理Agent的对话历史和工作记忆，通常使用ConversationBufferMemory。
• skills/base_skill.py：定义所有技能的抽象基类（BaseSkill），规定每个技能必须实现的接口（如execute方法）。
• skills/*.py：具体的技能实现。每个技能文件都是一个独立的、可插拔的功能模块。
• tools/：存放更底层的、单一功能的工具，例如一个纯粹的网页抓取函数。技能可以组合调用多个工具。
• utils/config.py：统一加载环境配置和模型设置。

4. 核心代码实战：一步步构建Agent系统

接下来，我们将从底层到上层，逐一实现上述模块。

4.1 实现技能基类与配置加载

首先，定义技能的契约。创建skills/base_skill.py：

# skills/base_skill.py from abc import ABC, abstractmethod from typing import Any, Dict from pydantic import BaseModel, Field class SkillInput(BaseModel): """技能的输入参数模型""" input_text: str = Field(description="用户输入或上级指令") context: Dict[str, Any] = Field(default_factory=dict, description="执行上下文信息") class BaseSkill(ABC): """所有技能的抽象基类""" name: str = "base_skill" description: str = "一个基础的技能" def __init__(self, config: Dict[str, Any] = None): self.config = config or {} @abstractmethod async def execute(self, skill_input: SkillInput) -> Dict[str, Any]: """ 执行技能的核心方法。 返回一个字典，至少包含 `output` (技能输出) 和 `status` (执行状态)。 """ pass def get_description(self) -> str: """获取技能的描述，用于Agent决策""" return f"{self.name}: {self.description}"

接着，创建配置加载工具utils/config.py：

# utils/config.py import os from dotenv import load_dotenv from typing import Optional load_dotenv() # 加载 .env 文件中的环境变量 class Config: """全局配置类""" OPENAI_API_KEY: Optional[str] = os.getenv("OPENAI_API_KEY") # 可以在这里添加其他配置，如模型名称、温度等 MODEL_NAME: str = "gpt-3.5-turbo" MODEL_TEMPERATURE: float = 0.1 # 低温度使输出更确定 @classmethod def validate(cls): """验证必要配置是否存在""" if not cls.OPENAI_API_KEY: raise ValueError("OPENAI_API_KEY 未在环境变量中设置。请检查 .env 文件。") config = Config() config.validate()

4.2 实现具体技能：Web搜索与计算器

让我们实现两个基础技能，感受一下技能模块的独立性。

技能一：网络搜索 (skills/web_search.py)这个技能利用duckduckgo-search库进行实时网络搜索。

# skills/web_search.py from duckduckgo_search import DDGS from .base_skill import BaseSkill, SkillInput from typing import Dict, Any import asyncio class WebSearchSkill(BaseSkill): name = "web_search" description = "使用DuckDuckGo进行网络搜索，获取最新信息。" def __init__(self, config: Dict[str, Any] = None): super().__init__(config) self.max_results = self.config.get("max_results", 3) async def execute(self, skill_input: SkillInput) -> Dict[str, Any]: query = skill_input.input_text if not query: return {"output": "搜索查询不能为空。", "status": "error"} try: # 注意：DDGS() 同步调用，我们用线程池包装以避免阻塞事件循环 loop = asyncio.get_event_loop() results = await loop.run_in_executor( None, self._sync_search, query ) formatted_results = [] for r in results[:self.max_results]: formatted_results.append(f"标题: {r['title']}\n链接: {r['href']}\n摘要: {r['body']}\n") output = f"关于 '{query}' 的搜索结果（共{len(formatted_results)}条）：\n" + "-"*50 + "\n" output += "\n".join(formatted_results) return { "output": output, "status": "success", "raw_results": results[:self.max_results] # 保留原始数据供其他技能使用 } except Exception as e: return {"output": f"搜索过程中发生错误：{str(e)}", "status": "error"} def _sync_search(self, query: str): """同步执行搜索的函数""" with DDGS() as ddgs: return list(ddgs.text(query, max_results=self.max_results+2))

技能二：计算器 (skills/calculator.py)这是一个安全的、受限的计算技能，使用eval但进行了严格限制。

# skills/calculator.py import ast import operator as op from .base_skill import BaseSkill, SkillInput from typing import Dict, Any # 定义允许的运算符 allowed_operators = { ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv, ast.Pow: op.pow, ast.USub: op.neg, } class CalculatorSkill(BaseSkill): name = "calculator" description = "执行安全的数学表达式计算（支持 +, -, *, /, **, 括号）。" async def execute(self, skill_input: SkillInput) -> Dict[str, Any]: expression = skill_input.input_text.strip() if not expression: return {"output": "计算表达式不能为空。", "status": "error"} try: # 使用抽象语法树进行安全评估 node = ast.parse(expression, mode='eval').body result = self._eval_expr(node) return {"output": f"{expression} = {result}", "status": "success", "result": result} except (SyntaxError, TypeError, KeyError, ZeroDivisionError) as e: return {"output": f"计算错误或表达式不安全：{str(e)}", "status": "error"} def _eval_expr(self, node): """递归安全地评估AST节点""" if isinstance(node, ast.Num): # 数字 return node.n elif isinstance(node, ast.BinOp): # 二元操作 left_val = self._eval_expr(node.left) right_val = self._eval_expr(node.right) operator_func = allowed_operators.get(type(node.op)) if operator_func is None: raise TypeError(f"不允许的操作符: {type(node.op)}") return operator_func(left_val, right_val) elif isinstance(node, ast.UnaryOp): # 一元操作（如负数） operand_val = self._eval_expr(node.operand) operator_func = allowed_operators.get(type(node.op)) if operator_func is None: raise TypeError(f"不允许的操作符: {type(node.op)}") return operator_func(operand_val) else: raise TypeError(f"不支持的表达式类型: {type(node)}")

4.3 实现复合技能：DeepResearch（深度研究）

这是本项目的亮点。DeepResearchSkill本身不直接做具体工作，而是扮演一个“子项目经理”的角色，它利用已有的WebSearchSkill和规划能力，完成一个多步骤的研究任务。

创建skills/deep_research.py：

# skills/deep_research.py import asyncio from typing import Dict, Any, List from .base_skill import BaseSkill, SkillInput from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema import SystemMessage, HumanMessage from utils.config import config class DeepResearchSkill(BaseSkill): name = "deep_research" description = "对一个复杂主题进行深度研究：分解问题、并行搜索、综合分析并生成报告。" def __init__(self, config: Dict[str, Any] = None): super().__init__(config) # 初始化一个大语言模型，用于规划和分析 self.llm = ChatOpenAI( api_key=config.OPENAI_API_KEY, model=config.MODEL_NAME, temperature=0.2, ) # 这里假设我们已经有了一个搜索技能实例，实际中可通过依赖注入传入 from .web_search import WebSearchSkill self.search_skill = WebSearchSkill(config={"max_results": 4}) async def execute(self, skill_input: SkillInput) -> Dict[str, Any]: research_topic = skill_input.input_text print(f"[DeepResearch] 开始研究主题: {research_topic}") # 步骤1：规划研究子问题 sub_questions = await self._plan_research(research_topic) print(f"[DeepResearch] 生成的子问题: {sub_questions}") # 步骤2：并行搜索所有子问题 search_tasks = [] for q in sub_questions: task = self.search_skill.execute(SkillInput(input_text=q)) search_tasks.append(task) search_results = await asyncio.gather(*search_tasks) print(f"[DeepResearch] 并行搜索完成，共 {len(search_results)} 组结果。") # 步骤3：综合分析与报告生成 final_report = await self._synthesize_report(research_topic, sub_questions, search_results) return { "output": final_report, "status": "success", "metadata": { "topic": research_topic, "sub_questions": sub_questions, "search_summary": [r.get('status') for r in search_results] } } async def _plan_research(self, topic: str) -> List[str]: """使用LLM将宽泛的研究主题分解为3-5个具体的子问题""" prompt = ChatPromptTemplate.from_messages([ SystemMessage(content="你是一个资深研究助理。请将用户给出的宽泛研究主题，分解为3到5个具体、可搜索的关键子问题。这些子问题应覆盖主题的不同方面，并有助于形成全面的理解。请直接输出问题列表，每行一个，不要编号以外的额外解释。"), HumanMessage(content=f"研究主题：{topic}") ]) messages = prompt.format_messages() response = await self.llm.ainvoke(messages) # 解析LLM返回的文本，拆分成问题列表 questions_text = response.content.strip() questions = [q.strip() for q in questions_text.split('\n') if q.strip()] # 清理可能存在的编号前缀（如“1. ”，“- ”） cleaned_questions = [] for q in questions: # 移除常见的列表标记 for prefix in ['1.', '2.', '3.', '4.', '5.', '- ', '* ']: if q.startswith(prefix): q = q[len(prefix):].strip() break cleaned_questions.append(q) return cleaned_questions[:5] # 最多返回5个问题 async def _synthesize_report(self, topic: str, sub_questions: List[str], search_results: List[Dict]) -> str: """综合搜索结果为一份结构化报告""" # 准备给LLM的上下文 context_parts = [] for i, (q, result) in enumerate(zip(sub_questions, search_results)): if result['status'] == 'success': context_parts.append(f"## 子问题 {i+1}: {q}\n搜索到的相关信息：\n{result['output'][:1500]}...\n") # 截取部分内容避免token超限 else: context_parts.append(f"## 子问题 {i+1}: {q}\n[信息获取失败: {result['output']}]\n") research_context = "\n".join(context_parts) synthesis_prompt = ChatPromptTemplate.from_messages([ SystemMessage(content="你是一个分析专家。请基于以下关于某个主题的多个子问题的搜索信息，撰写一份简洁、清晰、结构化的研究报告。报告应包含：1) 概述；2) 针对每个子问题的关键发现；3) 综合结论与见解。确保信息准确，如果某些部分信息不足请注明。"), HumanMessage(content=f"研究主题：{topic}\n\n以下是各子问题的搜索信息：\n{research_context}\n\n请生成研究报告：") ]) messages = synthesis_prompt.format_messages() response = await self.llm.ainvoke(messages) return response.content

4.4 实现Agent核心与记忆管理

现在，我们来创建Agent的大脑。创建core/agent.py：

# core/agent.py from typing import Dict, Any, List, Optional from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema import SystemMessage, HumanMessage, AIMessage from skills.base_skill import BaseSkill, SkillInput from utils.config import config import json class Agent: """智能体核心，负责技能管理和任务调度""" def __init__(self, skills: List[BaseSkill], model_name: str = None): self.skills = {skill.name: skill for skill in skills} self.llm = ChatOpenAI( api_key=config.OPENAI_API_KEY, model=model_name or config.MODEL_NAME, temperature=config.MODEL_TEMPERATURE, ) # 简单的对话历史记忆 self.conversation_history: List[Dict[str, str]] = [] def _build_system_prompt(self) -> str: """构建系统提示词，描述Agent的能力和规则""" skill_descriptions = "\n".join([skill.get_description() for skill in self.skills.values()]) return f"""你是一个多功能AI助手，可以调用以下技能来帮助用户： {skill_descriptions} 请遵循以下规则： 1. 仔细分析用户的请求。 2. 如果请求可以通过直接对话回答，请直接回答。 3. 如果需要使用上述技能，请严格按照以下JSON格式回复： ```json {{"action": "use_skill", "skill_name": "技能名称", "input": "传递给技能的输入文本"}}

4. 不要解释你将做什么，直接输出JSON。 """

   async def process_query(self, user_input: str) -> str: """处理用户输入的主要方法""" # 将用户输入加入历史 self.conversation_history.append({"role": "user", "content": user_input})

   # 准备包含历史的对话上下文 messages = [SystemMessage(content=self._build_system_prompt())] for msg in self.conversation_history[-6:]: # 保留最近6轮对话作为上下文 if msg["role"] == "user": messages.append(HumanMessage(content=msg["content"])) else: messages.append(AIMessage(content=msg["content"])) # 调用LLM进行决策 llm_response = await self.llm.ainvoke(messages) response_text = llm_response.content.strip() # 尝试解析LLM的响应，看是否是技能调用指令 skill_result = None try: # 尝试从响应中提取JSON代码块 if "```json" in response_text: json_str = response_text.split("```json")[1].split("```")[0].strip() elif response_text.startswith("{") and response_text.endswith("}"): json_str = response_text else: json_str = None if json_str: action_data = json.loads(json_str) if action_data.get("action") == "use_skill": skill_name = action_data["skill_name"] skill_input_text = action_data["input"] if skill_name in self.skills: print(f"[Agent] 决定调用技能: {skill_name}, 输入: {skill_input_text}") skill = self.skills[skill_name] skill_result = await skill.execute(SkillInput(input_text=skill_input_text, context={"user_query": user_input})) # 将技能执行结果格式化后，作为Agent的最终回复 if skill_result["status"] == "success": final_response = f"【{skill.name} 执行结果】\n{skill_result['output']}" else: final_response = f"【技能执行出错】\n{skill_result['output']}" else: final_response = f"未知的技能名称: {skill_name}。可用技能: {list(self.skills.keys())}" else: final_response = response_text # LLM直接回复了文本 else: final_response = response_text # LLM直接回复了文本 except json.JSONDecodeError: # 如果响应不是JSON，则当作普通对话回复 final_response = response_text except Exception as e: final_response = f"处理技能调用时发生错误: {str(e)}" # 将Agent的回复加入历史 self.conversation_history.append({"role": "assistant", "content": final_response}) return final_response

记忆管理模块 `core/memory.py` 可以更复杂，但为了简化，我们暂时使用Agent内部的列表。在实际项目中，你可能需要集成 `ConversationBufferMemory` 或向量数据库。 ### 4.5 主程序入口与运行测试 最后，创建 `main.py` 来组装一切并提供一个简单的交互界面。 ```python # main.py import asyncio from core.agent import Agent from skills.web_search import WebSearchSkill from skills.calculator import CalculatorSkill from skills.deep_research import DeepResearchSkill from utils.config import config async def main(): print("初始化 Agent Skills 系统...") # 1. 实例化所有技能 skills = [ WebSearchSkill(config={"max_results": 3}), CalculatorSkill(), DeepResearchSkill(config), # 传入配置 ] # 2. 创建智能体 agent = Agent(skills=skills, model_name="gpt-3.5-turbo") print("系统初始化完成！") print("可用技能:", list(agent.skills.keys())) print("输入 'quit' 或 'exit' 退出程序。") print("-" * 50) # 3. 交互循环 while True: try: user_input = input("\n您: ").strip() if user_input.lower() in ['quit', 'exit', 'q']: print("再见！") break if not user_input: continue print("Agent 正在思考...") response = await agent.process_query(user_input) print(f"\n助手: {response}") except KeyboardInterrupt: print("\n\n程序被中断。") break except Exception as e: print(f"\n发生未预期错误: {e}") if __name__ == "__main__": asyncio.run(main())

运行与测试：

1. 确保你的.env文件已正确配置OPENAI_API_KEY。
2. 在终端运行：

   python main.py

3. 尝试以下交互：
   • “计算一下 (15 + 27) * 3 等于多少？”-> 应触发计算器技能。
   • “搜索一下今天 OpenAI 有什么新闻。”-> 应触发网络搜索技能。
   • “请深度研究一下‘人工智能在医疗诊断中的最新应用’。”-> 应触发深度研究技能。你会看到它先规划子问题，然后并行搜索，最后生成报告。

5. 常见问题与排查思路

在构建和运行此类Agent系统时，你可能会遇到以下典型问题：

6. 最佳实践与工程建议

将原型转化为可投入生产环境的系统，需要考虑更多工程化细节。

6.1 技能设计规范

• 单一职责：每个技能只做一件事，并做好。避免创建“万能技能”。
• 明确接口：输入输出使用Pydantic模型严格定义，便于验证和文档化。
• 错误处理：技能内部必须进行细致的异常处理，并返回统一的错误格式，避免整个Agent崩溃。
• 可配置化：像max_results这样的参数应从外部传入，提高灵活性。

6.2 Agent核心优化

• 技能路由优化：当前的基于LLM的“思考-决策”循环成本较高且可能不稳定。对于简单、明确的意图，可以优先使用意图识别分类器（例如微调一个小模型或使用规则）来路由，降低成本和提高速度。
• 流式输出：对于耗时的技能（如DeepResearch），实现流式输出（Streaming），让用户能看到“思考中...”、“正在搜索...”、“正在分析...”等中间状态，提升体验。
• 持久化记忆：集成向量数据库（如Chroma, Pinecone）来存储和检索长期记忆，使Agent能记住跨会话的关键信息。

6.3 系统架构扩展

• 技能市场与动态加载：设计一个技能注册中心，支持从配置文件或数据库动态加载技能，无需重启服务。
• 技能编排与工作流：引入工作流引擎（如Prefect、Airflow）来管理复杂技能的执行顺序和条件分支，实现更强大的自动化流程。
• 监控与可观测性：为每个技能调用和Agent决策记录日志、指标（如耗时、成功率），便于监控和调试。使用像LangSmith这样的工具可以极大提升开发效率。

6.4 安全与合规

• 输入净化：对所有用户输入和技能输入进行严格的验证和净化，防止注入攻击。
• 工具权限控制：为技能分级（如“读取网络信息”、“执行计算”、“写入数据库”），并根据用户或会话上下文进行权限校验。
• 内容审核：在技能输出最终给用户前，可增加一层内容安全审核，过滤不当信息。
• 数据隐私：如果技能处理用户个人数据，需确保符合数据隐私法规，避免在日志或记忆中泄露敏感信息。

通过本教程，你不仅实现了一个具备基础技能和深度研究能力的AI智能体，更重要的是掌握了一套构建可扩展、模块化Agent系统的设计方法论和实战代码。从技能抽象、Agent决策循环到复合技能编排，这些模式是开发更复杂AI应用的基础。你可以在此基础上，继续集成更多技能（如数据库查询、发送邮件、生成图表），优化决策逻辑，或将其封装为API服务，从而打造出真正解决实际业务问题的智能助手。