最近在嘗試將AI能力集成到業務系統中時發現市面上的智能體平臺雖然功能強大但要么是黑盒要么定制成本極高要么就是難以與現有開發流程和工具鏈深度集成。對于希望將智能體能力“工程化”落地的團隊來說從零理解其核心并搭建一套可控、可擴展、可集成的開發工具鏈是必經之路。本文將從零開始手把手帶你搭建一套專屬于你自己的智能體Agent開發工具鏈。我們將不依賴任何大型商業平臺而是基于開源組件和標準協議構建一個從環境配置、核心框架、工具集成到工程化部署的完整閉環。無論你是想深入理解Agent的內部機制還是希望為團隊打造一套標準化的AI開發基礎設施這篇文章都將提供一套可直接復用的實戰方案。1. 智能體Agent開發的核心概念與工程化挑戰在開始動手之前我們必須明確幾個核心概念并理解為什么需要一套工具鏈而不是簡單地調用一個API。1.1 什么是智能體Agent在AI語境下一個智能體Agent通常指一個能夠感知環境、進行決策并執行行動以實現特定目標的軟件實體。與傳統的“聊天機器人”或“問答系統”不同一個真正的Agent具備幾個關鍵特征自主性Autonomy能在沒有人類直接干預的情況下運行。反應性Reactivity能感知環境如用戶輸入、API返回、數據庫變化并做出及時響應。主動性Pro-activeness不僅被動響應還能主動發起目標導向的行為。社交能力Social Ability能與其他Agent或人類進行交互和協作。當前基于大語言模型LLM的Agent是其最流行的實現形式。LLM作為其“大腦”負責理解、規劃和決策而外部的“工具”Tools則成為其“手腳”用于執行具體的操作如查詢數據庫、調用API、運行代碼等。1.2 為什么需要“工具鏈”而非“單點方案”很多開發者初涉Agent開發時會從一個簡單的腳本開始接收用戶輸入調用LLM API解析返回結果然后執行某個操作。但隨著需求復雜化這種模式會迅速陷入困境工具管理混亂工具函數散落在各處缺乏統一的注冊、描述和調用機制。狀態管理困難Agent與用戶的多次對話多輪對話狀態如何保存和恢復流程編排缺失復雜的任務需要多個Agent協作或按特定工作流執行代碼會變得極其臃腫。可觀測性差Agent內部如何思考、為什么選擇某個工具、執行結果如何這些過程如同黑盒難以調試和優化。工程化部署難如何將開發好的Agent打包、部署、監控、擴縮容并與現有CI/CD流程集成因此一套完整的Agent開發工具鏈旨在系統性地解決上述問題將Agent開發從“腳本編寫”升級為“軟件工程”。1.3 工具鏈的核心組件我們計劃構建的工具鏈將包含以下核心層這也是本文的實踐路線圖環境與基礎層Python環境、虛擬環境管理、依賴管理。核心框架層選擇或自建一個輕量級Agent核心框架負責大腦LLM的調用、工具的管理與調度、記憶對話歷史的維護。工具集成層標準化工具的封裝、注冊與調用接口。編排與工作流層實現多個Agent的協作和復雜任務的流程控制。工程化與部署層日志、監控、配置管理、容器化部署。2. 環境準備與項目初始化我們選擇Python作為主要開發語言因其在AI生態中擁有最豐富的庫支持。2.1 基礎環境配置首先確保你的系統已安裝Python推薦3.9或以上版本和pip。然后為項目創建一個獨立的虛擬環境這是管理依賴的最佳實踐。# 創建項目目錄 mkdir my_agent_toolchain cd my_agent_toolchain # 創建Python虛擬環境使用venv python -m venv venv # 激活虛擬環境 # 在Windows上 venv\Scripts\activate # 在Linux/Mac上 source venv/bin/activate激活后你的命令行提示符前會出現(venv)標識。2.2 初始化項目結構與依賴管理我們使用pyproject.toml現代Python項目標準來管理依賴和項目元數據。# 創建基礎項目結構 mkdir -p src/my_agent tools configs tests touch src/my_agent/__init__.py touch pyproject.toml README.md .gitignore編輯pyproject.toml文件定義項目依賴。我們將從最核心的依賴開始。# pyproject.toml [project] name my-agent-toolchain version 0.1.0 description A custom agent development toolchain from scratch. authors [{name Your Name, email your.emailexample.com}] readme README.md requires-python 3.9 dependencies [ openai1.0.0, # 用于調用OpenAI API或其他兼容API langchain-core0.1.0, # 使用LangChain的核心抽象但不一定用其全量框架 pydantic2.0.0, # 用于數據驗證和設置管理 httpx0.25.0, # 異步HTTP客戶端用于工具調用 python-dotenv1.0.0, # 從.env文件加載環境變量 ] [project.optional-dependencies] dev [ pytest7.0.0, black23.0.0, isort5.12.0, ] web [ fastapi0.104.0, uvicorn[standard]0.24.0, ] [build-system] requires [setuptools61.0, wheel] build-back setuptools.build_meta然后安裝基礎依賴pip install -e . # 以可編輯模式安裝當前項目創建.env文件來存儲敏感信息如API密鑰切記不要將其提交到版本控制系統。# .env OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果使用其他兼容服務可修改此處編輯.gitignore文件忽略虛擬環境、緩存文件和.env。# .gitignore venv/ __pycache__/ *.py[cod] .env .pytest_cache/ .coverage3. 構建核心Agent框架我們不直接使用龐大的全功能框架而是基于清晰的概念自建核心這有助于深刻理解Agent的運行機制。3.1 定義核心抽象Agent、Tool、Memory在src/my_agent/core目錄下創建基礎抽象類。# src/my_agent/core/agent.py from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class Tool(BaseModel): 工具基類每個工具都必須繼承此類。 name: str Field(description工具的唯一名稱) description: str Field(description工具功能的自然語言描述用于讓LLM理解何時使用此工具) args_schema: Optional[type[BaseModel]] Field(defaultNone, description工具參數的Pydantic模型) abstractmethod async def run(self, **kwargs) - str: 執行工具的核心方法。 pass class Memory(BaseModel): 記憶基類負責存儲和檢索對話歷史。 messages: List[Dict[str, Any]] Field(default_factorylist) def add_message(self, role: str, content: str): 添加一條消息到歷史記錄。 self.messages.append({role: role, content: content}) def get_context(self, max_tokens: int 2000) - List[Dict[str, Any]]: 獲取最近的對話上下文用于發送給LLM。 # 簡單的實現返回全部消息生產環境需實現Token計數和截斷 return self.messages[-10:] # 示例返回最近10條 class BaseAgent(ABC): Agent基類。 def __init__(self, llm_client, memory: Optional[Memory] None): self.llm llm_client self.memory memory or Memory() self.tools: Dict[str, Tool] {} def register_tool(self, tool: Tool): 向Agent注冊一個工具。 self.tools[tool.name] tool abstractmethod async def think(self, user_input: str) - str: 核心思考循環處理用戶輸入可能調用工具并生成最終回復。 pass3.2 實現一個簡單的ReAct模式AgentReActReasoning Acting是一種經典的Agent推理模式。我們實現一個簡化版本。# src/my_agent/core/react_agent.py import json import re from typing import Dict, Any from .agent import BaseAgent, Tool, Memory from pydantic import BaseModel class ReasoningStep(BaseModel): thought: str action: Optional[str] None # 工具名 action_input: Optional[Dict[str, Any]] None observation: Optional[str] None final_answer: Optional[str] None class ReActAgent(BaseAgent): 一個實現ReAct推理模式的簡單Agent。 async def think(self, user_input: str) - str: # 將用戶輸入加入記憶 self.memory.add_message(user, user_input) # 構建系統提示包含工具描述 tools_description \n.join([f- {name}: {tool.description} for name, tool in self.tools.items()]) system_prompt f你是一個有幫助的AI助手可以調用工具來解決問題。 你可以使用的工具如下 {tools_description} 請遵循以下格式進行思考 Thought: 你需要思考當前情況決定是否需要使用工具以及使用哪個工具。 Action: 需要調用的工具名稱如果沒有工具可用或不需要就填 None。 Action Input: 調用工具所需的輸入參數必須是JSON格式。如果Action是None這里也填 null。 Observation: 工具執行后的結果。 ... (這個 Thought/Action/Action Input/Observation 循環可以重復多次) Thought: 我現在有足夠的信息來回答用戶了。 Final Answer: 給用戶的最終回答。 # 獲取對話上下文 context_messages self.memory.get_context() # 準備發送給LLM的消息 messages [ {role: system, content: system_prompt}, *context_messages, {role: user, content: user_input}, ] max_iterations 5 for i in range(max_iterations): # 調用LLM獲取下一步推理 llm_response await self.llm.chat.completions.create( modelgpt-3.5-turbo, # 或 gpt-4 messagesmessages, temperature0, ) response_text llm_response.choices[0].message.content # 解析LLM的響應提取 Thought, Action 等部分這里簡化實際應用需要更魯棒的解析 # 假設LLM嚴格按照格式回復 thought_match re.search(rThought:\s*(.), response_text, re.DOTALL) action_match re.search(rAction:\s*(.), response_text) action_input_match re.search(rAction Input:\s*(.), response_text, re.DOTALL) thought thought_match.group(1).strip() if thought_match else action action_match.group(1).strip() if action_match else None action_input_str action_input_match.group(1).strip() if action_input_match else null print(f[Agent Iteration {i1}] Thought: {thought}) print(f[Agent Iteration {i1}] Action: {action}) if action and action ! None: # 執行工具調用 try: action_input json.loads(action_input_str) if action_input_str ! null else {} tool self.tools.get(action) if tool: observation await tool.run(**action_input) print(f[Agent Iteration {i1}] Observation: {observation}) # 將本次行動和觀察加入消息歷史供下一輪參考 messages.append({role: assistant, content: fAction: {action}\nAction Input: {action_input_str}}) messages.append({role: user, content: fObservation: {observation}}) else: observation fError: Tool {action} not found. messages.append({role: user, content: fObservation: {observation}}) except json.JSONDecodeError: observation fError: Invalid JSON in Action Input: {action_input_str} messages.append({role: user, content: fObservation: {observation}}) except Exception as e: observation fError executing tool {action}: {str(e)} messages.append({role: user, content: fObservation: {observation}}) else: # 沒有更多行動嘗試提取最終答案 final_answer_match re.search(rFinal Answer:\s*(.), response_text, re.DOTALL) if final_answer_match: final_answer final_answer_match.group(1).strip() self.memory.add_message(assistant, final_answer) return final_answer else: # 如果沒有明確Final Answer可能LLM格式有誤直接返回其回復 self.memory.add_message(assistant, response_text) return response_text return 抱歉經過多輪推理仍未得到最終答案。3.3 集成LLM客戶端我們使用OpenAI官方Python SDK并對其進行簡單封裝以適配我們的Agent接口。# src/my_agent/llm/openai_client.py import os from openai import AsyncOpenAI from dotenv import load_dotenv load_dotenv() # 加載.env文件中的環境變量 class OpenAIClient: def __init__(self): api_key os.getenv(OPENAI_API_KEY) base_url os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) if not api_key: raise ValueError(OPENAI_API_KEY environment variable is not set.) self.client AsyncOpenAI(api_keyapi_key, base_urlbase_url) property def chat(self): # 提供一個與我們的Agent期望的接口兼容的屬性 return self.client.chat4. 開發與集成自定義工具Tools工具是Agent能力的延伸。我們來創建幾個常用工具。4.1 天氣查詢工具# src/my_agent/tools/weather_tool.py import httpx from pydantic import BaseModel, Field from ..core.agent import Tool from dotenv import load_dotenv import os load_dotenv() class WeatherInput(BaseModel): city: str Field(description城市名稱例如北京、Shanghai) class WeatherTool(Tool): def __init__(self): super().__init__( nameget_weather, description根據城市名稱查詢當前天氣情況。, args_schemaWeatherInput ) self.api_key os.getenv(WEATHER_API_KEY) # 假設你有一個天氣API的Key # 這里使用一個模擬的免費API示例實際使用時請替換為真實API self.base_url http://wttr.in/ async def run(self, city: str) - str: 調用天氣API。 try: async with httpx.AsyncClient() as client: # 注意wttr.in 是一個免費服務格式可能變化僅作示例 url f{self.base_url}{city}?format3 # 格式3返回簡短文本 response await client.get(url, timeout10.0) response.raise_for_status() weather_info response.text.strip() return f{city}的天氣是{weather_info} except httpx.RequestError as e: return f請求天氣API時出錯{str(e)} except Exception as e: return f處理天氣信息時發生未知錯誤{str(e)}4.2 計算器工具# src/my_agent/tools/calculator_tool.py from pydantic import BaseModel, Field from ..core.agent import Tool import ast import operator as op class CalculatorInput(BaseModel): expression: str Field(description一個有效的數學表達式例如(3 5) * 2) class CalculatorTool(Tool): def __init__(self): super().__init__( namecalculator, description計算一個數學表達式的結果。支持加減乘除和括號。, args_schemaCalculatorInput ) # 定義安全的運算符 self._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, } async def run(self, expression: str) - str: 安全地計算數學表達式。 try: # 使用ast.literal_eval進行安全評估 # 注意這里我們實現一個更安全的自定義評估器避免直接使用eval result self._safe_eval(expression) return f表達式 {expression} 的計算結果是{result} except (SyntaxError, ValueError, TypeError, ZeroDivisionError) as e: return f計算表達式 {expression} 時出錯{str(e)}。請確保表達式格式正確。 def _safe_eval(self, node): 遞歸安全地評估AST節點。 if isinstance(node, ast.Num): # number return node.n elif isinstance(node, ast.BinOp): # left operator right left_val self._safe_eval(node.left) right_val self._safe_eval(node.right) op_func self._allowed_operators.get(type(node.op)) if op_func is None: raise TypeError(f不允許的操作符{type(node.op)}) return op_func(left_val, right_val) elif isinstance(node, ast.UnaryOp): # operator operand e.g., -1 operand_val self._safe_eval(node.operand) op_func self._allowed_operators.get(type(node.op)) if op_func is None: raise TypeError(f不允許的一元操作符{type(node.op)}) return op_func(operand_val) elif isinstance(node, ast.Constant): # Python 3.8 常量 return node.value else: raise TypeError(f不支持的AST節點類型{type(node)}) def _safe_eval(self, expr: str): 入口函數將字符串表達式解析為AST并安全評估。 tree ast.parse(expr, modeeval) return self._safe_eval(tree.body) # 注意這里遞歸調用的是上面的方法需要重命名避免歧義。實際代碼中應調整。修正上面的遞歸問題將內部方法重命名def _eval_node(self, node): if isinstance(node, ast.Num): return node.n elif isinstance(node, ast.Constant): return node.value elif isinstance(node, ast.BinOp): left_val self._eval_node(node.left) right_val self._eval_node(node.right) op_func self._allowed_operators.get(type(node.op)) if op_func is None: raise TypeError(f不允許的操作符{type(node.op)}) return op_func(left_val, right_val) elif isinstance(node, ast.UnaryOp): operand_val self._eval_node(node.operand) op_func self._allowed_operators.get(type(node.op)) if op_func is None: raise TypeError(f不允許的一元操作符{type(node.op)}) return op_func(operand_val) else: raise TypeError(f不支持的AST節點類型{type(node)}) async def run(self, expression: str) - str: try: tree ast.parse(expression, modeeval) result self._eval_node(tree.body) return f表達式 {expression} 的計算結果是{result} except (SyntaxError, ValueError, TypeError, ZeroDivisionError, AttributeError) as e: return f計算表達式 {expression} 時出錯{str(e)}。請確保表達式格式正確且僅包含基本算術運算。5. 組裝并運行你的第一個Agent現在讓我們將各個部分組裝起來創建一個可以對話的Agent。5.1 創建主運行腳本# run_agent.py import asyncio import sys from src.my_agent.llm.openai_client import OpenAIClient from src.my_agent.core.react_agent import ReActAgent from src.my_agent.tools.weather_tool import WeatherTool from src.my_agent.tools.calculator_tool import CalculatorTool async def main(): # 1. 初始化LLM客戶端 llm_client OpenAIClient() # 2. 創建Agent實例 agent ReActAgent(llm_clientllm_client) # 3. 注冊工具 agent.register_tool(WeatherTool()) agent.register_tool(CalculatorTool()) print(智能體已啟動輸入 quit 或 exit 退出。) print(- * 40) while True: try: user_input input(\nYou: ).strip() if user_input.lower() in [quit, exit]: print(再見) break if not user_input: continue # 4. 讓Agent思考并回復 response await agent.think(user_input) print(f\nAgent: {response}) except KeyboardInterrupt: print(\n\n程序被中斷。) break except Exception as e: print(f\n發生錯誤{e}) if __name__ __main__: asyncio.run(main())5.2 運行與測試在項目根目錄下運行python run_agent.py你應該會看到提示符。嘗試輸入“北京今天天氣怎么樣”Agent會調用天氣工具“計算一下 (12 34) * 2 等于多少”Agent會調用計算器工具“你是誰”Agent會直接利用LLM知識回答觀察控制臺輸出的Thought、Action、Observation日志理解ReAct模式的運行過程。6. 工程化進階構建工具鏈的其他關鍵環節一個基礎的Agent跑起來了但要將其工程化我們還需要完善以下環節。6.1 工具的動態加載與發現手動注冊工具在工具數量多時會很麻煩。我們可以實現一個工具發現機制。# src/my_agent/core/tool_registry.py import importlib import pkgutil from pathlib import Path from typing import Dict, Type from .agent import Tool class ToolRegistry: _tools: Dict[str, Type[Tool]] {} classmethod def register(cls, tool_class: Type[Tool]): 類裝飾器用于注冊工具類。 instance tool_class() cls._tools[instance.name] tool_class return tool_class classmethod def discover_tools(cls, package_path: str): 自動發現指定包路徑下所有繼承了Tool的類并注冊。 package importlib.import_module(package_path) for _, module_name, is_pkg in pkgutil.iter_modules(package.__path__, package.__name__ .): if not is_pkg: module importlib.import_module(module_name) for attr_name in dir(module): attr getattr(module, attr_name) if (isinstance(attr, type) and issubclass(attr, Tool) and attr ! Tool): # 排除基類本身 cls.register(attr) classmethod def get_tool_instance(cls, tool_name: str) - Tool: 根據工具名獲取工具實例。 tool_class cls._tools.get(tool_name) if tool_class: return tool_class() raise KeyError(fTool {tool_name} not found in registry.) classmethod def get_all_tool_descriptions(cls) - Dict[str, str]: 獲取所有已注冊工具的描述。 return {name: cls.get_tool_instance(name).description for name in cls._tools.keys()}然后我們可以用裝飾器來聲明工具# src/my_agent/tools/weather_tool.py from src.my_agent.core.tool_registry import ToolRegistry ToolRegistry.register class WeatherTool(Tool): # ... 其余代碼不變 ...在主程序中可以自動加載所有工具# run_agent_auto.py from src.my_agent.core.tool_registry import ToolRegistry # ... 其他導入 ... async def main(): # 自動發現并注冊 src.my_agent.tools 包下的所有工具 ToolRegistry.discover_tools(src.my_agent.tools) llm_client OpenAIClient() agent ReActAgent(llm_clientllm_client) # 從注冊表獲取所有工具實例并注冊到Agent for tool_name in ToolRegistry._tools.keys(): agent.register_tool(ToolRegistry.get_tool_instance(tool_name)) # ... 其余代碼 ...6.2 記憶Memory的持久化當前的Memory類只在內存中保存對話。生產環境需要持久化到數據庫如Redis、SQLite或向量數據庫用于長上下文摘要。# src/my_agent/core/persistent_memory.py import json from typing import List, Dict, Any from pydantic import BaseModel import sqlite3 from datetime import datetime class PersistentMemory(BaseModel): session_id: str db_path: str agent_memory.db class Config: arbitrary_types_allowed True def __init__(self, session_id: str, **data): super().__init__(session_idsession_id, **data) self._init_db() def _init_db(self): conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS message_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def add_message(self, role: str, content: str): conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( INSERT INTO message_history (session_id, role, content) VALUES (?, ?, ?), (self.session_id, role, content) ) conn.commit() conn.close() def get_context(self, max_messages: int 10) - List[Dict[str, Any]]: conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( SELECT role, content FROM message_history WHERE session_id ? ORDER BY timestamp DESC LIMIT ?, (self.session_id, max_messages) ) rows cursor.fetchall() conn.close() # 返回時按時間順序從舊到新 messages [{role: row[0], content: row[1]} for row in reversed(rows)] return messages def clear_session(self): conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute(DELETE FROM message_history WHERE session_id ?, (self.session_id,)) conn.commit() conn.close()6.3 添加API服務層FastAPI要集成到現有系統需要提供HTTP API。# src/my_agent/api/server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from contextlib import asynccontextmanager from ..core.react_agent import ReActAgent from ..llm.openai_client import OpenAIClient from ..core.tool_registry import ToolRegistry import uuid # 全局Agent實例簡單示例生產環境需考慮并發和狀態隔離 _agent None asynccontextmanager async def lifespan(app: FastAPI): # 啟動時初始化 global _agent ToolRegistry.discover_tools(src.my_agent.tools) llm_client OpenAIClient() _agent ReActAgent(llm_clientllm_client) for tool_name in ToolRegistry._tools.keys(): _agent.register_tool(ToolRegistry.get_tool_instance(tool_name)) print(Agent initialized.) yield # 關閉時清理 print(Shutting down.) app FastAPI(lifespanlifespan) class ChatRequest(BaseModel): session_id: str None # 為空則創建新會話 message: str class ChatResponse(BaseModel): session_id: str reply: str app.post(/chat, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): if _agent is None: raise HTTPException(status_code503, detailAgent not initialized) # 這里簡化處理實際應將Memory與session_id綁定 session_id request.session_id or str(uuid.uuid4()) # TODO: 根據session_id從數據庫加載或創建PersistentMemory reply await _agent.think(request.message) return ChatResponse(session_idsession_id, replyreply) app.get(/health) async def health_check(): return {status: healthy}使用Uvicorn運行pip install fastapi uvicorn[standard] uvicorn src.my_agent.api.server:app --host 0.0.0.0 --port 8000 --reload6.4 配置管理Pydantic Settings使用Pydantic Settings管理所有配置。# src/my_agent/config/settings.py from pydantic_settings import BaseSettings from pydantic import Field class Settings(BaseSettings): openai_api_key: str Field(..., envOPENAI_API_KEY) openai_base_url: str Field(https://api.openai.com/v1, envOPENAI_BASE_URL) weather_api_key: str Field(, envWEATHER_API_KEY) database_url: str Field(sqlite:///./agent.db, envDATABASE_URL) log_level: str Field(INFO, envLOG_LEVEL) class Config: env_file .env extra ignore # 忽略.env中未定義的變量 settings Settings()7. 常見問題與排查思路在搭建和運行過程中你可能會遇到以下問題問題現象可能原因排查步驟與解決方案導入錯誤ModuleNotFoundError1. 虛擬環境未激活。2. 項目未以可編輯模式安裝。3.PYTHONPATH未包含項目根目錄。1. 確認命令行前有(venv)。2. 在項目根目錄執行pip install -e .。3. 在IDE中設置正確的項目根目錄和解釋器。OpenAI API 調用失敗1. API Key 未設置或錯誤。2. 網絡問題或代理配置。3. 余額不足或速率限制。1. 檢查.env文件中的OPENAI_API_KEY。2. 檢查網絡連接如需代理在代碼中配置http_client。3. 查看OpenAI控制臺賬單和用量。Agent 不調用工具直接回答1. 系統提示詞Prompt中工具描述不清晰。2. LLM 溫度temperature設置過高導致輸出不穩定。3. 工具名稱或描述與用戶問題匹配度低。1. 優化系統提示詞明確指令格式。2. 將temperature設為0確保確定性輸出。3. 檢查工具描述是否準確嘗試用更直接的問題測試。工具調用參數解析錯誤1. LLM 生成的Action Input不是合法JSON。2. JSON中的參數名與工具定義的args_schema不匹配。1. 在Agent代碼中增加更健壯的JSON解析和錯誤處理。2. 在工具描述中明確參數名稱和類型。可以使用Pydantic的schema_json()為LLM提供更精確的格式。多輪對話狀態丟失1.Memory類未正確集成到Agent中。2. 每次請求創建了新的Agent實例。1. 確保agent.think()方法中正確讀取和更新了self.memory。2. 對于Web服務需要將會話ID與Memory實例綁定并持久化存儲。性能問題響應慢1. 工具調用是同步的阻塞了主線程。2. LLM API調用耗時過長。3. 未實現流式輸出。1. 確保所有工具方法都是async并使用await調用。2. 考慮設置合理的超時時間或使用更快的模型。3. 對于Web API可以研究SSEServer-Sent Events實現流式響應。8. 最佳實踐與工程化建議將Agent投入生產環境需要遵循以下工程化準則提示詞工程化將系統提示詞、用戶提示詞模板等抽取到配置文件或數據庫中便于管理和A/B測試。對提示詞進行版本控制。使用Jinja2等模板引擎動態生成提示詞。工具開發的標準化為所有工具編寫清晰的文檔包括輸入/輸出格式、錯誤碼。工具函數內部必須有完善的錯誤處理和日志記錄。為工具編寫單元測試和集成測試。可觀測性與監控在Agent的每個關鍵步驟接收輸入、調用LLM、調用工具、返回輸出記錄結構化日志。記錄每次LLM調用的輸入Token、輸出Token數量及成本。使用像Prometheus和Grafana監控工具調用成功率、延遲和Agent整體響應時間。安全與權限工具權限控制不是所有用戶都能調用所有工具。實現一個權限層根據用戶身份或會話上下文決定可用的工具集。輸入輸出過濾對用戶輸入和工具返回的內容進行安全檢查防止Prompt注入、敏感信息泄露。沙箱環境對于執行代碼、訪問文件系統等高危工具必須在安全的沙箱環境中運行。測試策略單元測試測試每個工具函數的邏輯。集成測試測試Agent與LLM、工具的集成流程可以使用LLM的Mock來避免真實API調用。端到端測試模擬真實用戶場景測試完整的對話流。部署與運維容器化使用Docker將Agent及其依賴打包確保環境一致性。配置分離所有密鑰、端點URL等配置必須通過環境變量或配置中心管理絕不能硬編碼。健康檢查與就緒探針為Web服務添加/health端點便于K8s等編排系統管理。版本回滾Agent的代碼、模型版本、提示詞版本都應有明確的版本號支持快速回滾。通過以上步驟你不僅搭建了一個可運行的智能體更構建了一套支撐其持續迭代和穩定運行的工程化工具鏈雛形。這套工具鏈的核心思想是模塊化、可觀測、可測試、可部署。你可以在此基礎上繼續擴展工作流引擎、可視化編排界面、更復雜的記憶模塊如向量數據庫逐步將其打造成團隊內部強大的AI能力中臺。 從零搭建智能體開發工具鏈：開源組件構建可控AI工程化方案
最近在嘗試將AI能力集成到業務系統中時發現市面上的智能體平臺雖然功能強大但要么是黑盒要么定制成本極高要么就是難以與現有開發流程和工具鏈深度集成。對于希望將智能體能力“工程化”落地的團隊來說從零理解其核心并搭建一套可控、可擴展、可集成的開發工具鏈是必經之路。本文將從零開始手把手帶你搭建一套專屬于你自己的智能體Agent開發工具鏈。我們將不依賴任何大型商業平臺而是基于開源組件和標準協議構建一個從環境配置、核心框架、工具集成到工程化部署的完整閉環。無論你是想深入理解Agent的內部機制還是希望為團隊打造一套標準化的AI開發基礎設施這篇文章都將提供一套可直接復用的實戰方案。1. 智能體Agent開發的核心概念與工程化挑戰在開始動手之前我們必須明確幾個核心概念并理解為什么需要一套工具鏈而不是簡單地調用一個API。1.1 什么是智能體Agent在AI語境下一個智能體Agent通常指一個能夠感知環境、進行決策并執行行動以實現特定目標的軟件實體。與傳統的“聊天機器人”或“問答系統”不同一個真正的Agent具備幾個關鍵特征自主性Autonomy能在沒有人類直接干預的情況下運行。反應性Reactivity能感知環境如用戶輸入、API返回、數據庫變化并做出及時響應。主動性Pro-activeness不僅被動響應還能主動發起目標導向的行為。社交能力Social Ability能與其他Agent或人類進行交互和協作。當前基于大語言模型LLM的Agent是其最流行的實現形式。LLM作為其“大腦”負責理解、規劃和決策而外部的“工具”Tools則成為其“手腳”用于執行具體的操作如查詢數據庫、調用API、運行代碼等。1.2 為什么需要“工具鏈”而非“單點方案”很多開發者初涉Agent開發時會從一個簡單的腳本開始接收用戶輸入調用LLM API解析返回結果然后執行某個操作。但隨著需求復雜化這種模式會迅速陷入困境工具管理混亂工具函數散落在各處缺乏統一的注冊、描述和調用機制。狀態管理困難Agent與用戶的多次對話多輪對話狀態如何保存和恢復流程編排缺失復雜的任務需要多個Agent協作或按特定工作流執行代碼會變得極其臃腫。可觀測性差Agent內部如何思考、為什么選擇某個工具、執行結果如何這些過程如同黑盒難以調試和優化。工程化部署難如何將開發好的Agent打包、部署、監控、擴縮容并與現有CI/CD流程集成因此一套完整的Agent開發工具鏈旨在系統性地解決上述問題將Agent開發從“腳本編寫”升級為“軟件工程”。1.3 工具鏈的核心組件我們計劃構建的工具鏈將包含以下核心層這也是本文的實踐路線圖環境與基礎層Python環境、虛擬環境管理、依賴管理。核心框架層選擇或自建一個輕量級Agent核心框架負責大腦LLM的調用、工具的管理與調度、記憶對話歷史的維護。工具集成層標準化工具的封裝、注冊與調用接口。編排與工作流層實現多個Agent的協作和復雜任務的流程控制。工程化與部署層日志、監控、配置管理、容器化部署。2. 環境準備與項目初始化我們選擇Python作為主要開發語言因其在AI生態中擁有最豐富的庫支持。2.1 基礎環境配置首先確保你的系統已安裝Python推薦3.9或以上版本和pip。然后為項目創建一個獨立的虛擬環境這是管理依賴的最佳實踐。# 創建項目目錄 mkdir my_agent_toolchain cd my_agent_toolchain # 創建Python虛擬環境使用venv python -m venv venv # 激活虛擬環境 # 在Windows上 venv\Scripts\activate # 在Linux/Mac上 source venv/bin/activate激活后你的命令行提示符前會出現(venv)標識。2.2 初始化項目結構與依賴管理我們使用pyproject.toml現代Python項目標準來管理依賴和項目元數據。# 創建基礎項目結構 mkdir -p src/my_agent tools configs tests touch src/my_agent/__init__.py touch pyproject.toml README.md .gitignore編輯pyproject.toml文件定義項目依賴。我們將從最核心的依賴開始。# pyproject.toml [project] name my-agent-toolchain version 0.1.0 description A custom agent development toolchain from scratch. authors [{name Your Name, email your.emailexample.com}] readme README.md requires-python 3.9 dependencies [ openai1.0.0, # 用于調用OpenAI API或其他兼容API langchain-core0.1.0, # 使用LangChain的核心抽象但不一定用其全量框架 pydantic2.0.0, # 用于數據驗證和設置管理 httpx0.25.0, # 異步HTTP客戶端用于工具調用 python-dotenv1.0.0, # 從.env文件加載環境變量 ] [project.optional-dependencies] dev [ pytest7.0.0, black23.0.0, isort5.12.0, ] web [ fastapi0.104.0, uvicorn[standard]0.24.0, ] [build-system] requires [setuptools61.0, wheel] build-back setuptools.build_meta然后安裝基礎依賴pip install -e . # 以可編輯模式安裝當前項目創建.env文件來存儲敏感信息如API密鑰切記不要將其提交到版本控制系統。# .env OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果使用其他兼容服務可修改此處編輯.gitignore文件忽略虛擬環境、緩存文件和.env。# .gitignore venv/ __pycache__/ *.py[cod] .env .pytest_cache/ .coverage3. 構建核心Agent框架我們不直接使用龐大的全功能框架而是基于清晰的概念自建核心這有助于深刻理解Agent的運行機制。3.1 定義核心抽象Agent、Tool、Memory在src/my_agent/core目錄下創建基礎抽象類。# src/my_agent/core/agent.py from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class Tool(BaseModel): 工具基類每個工具都必須繼承此類。 name: str Field(description工具的唯一名稱) description: str Field(description工具功能的自然語言描述用于讓LLM理解何時使用此工具) args_schema: Optional[type[BaseModel]] Field(defaultNone, description工具參數的Pydantic模型) abstractmethod async def run(self, **kwargs) - str: 執行工具的核心方法。 pass class Memory(BaseModel): 記憶基類負責存儲和檢索對話歷史。 messages: List[Dict[str, Any]] Field(default_factorylist) def add_message(self, role: str, content: str): 添加一條消息到歷史記錄。 self.messages.append({role: role, content: content}) def get_context(self, max_tokens: int 2000) - List[Dict[str, Any]]: 獲取最近的對話上下文用于發送給LLM。 # 簡單的實現返回全部消息生產環境需實現Token計數和截斷 return self.messages[-10:] # 示例返回最近10條 class BaseAgent(ABC): Agent基類。 def __init__(self, llm_client, memory: Optional[Memory] None): self.llm llm_client self.memory memory or Memory() self.tools: Dict[str, Tool] {} def register_tool(self, tool: Tool): 向Agent注冊一個工具。 self.tools[tool.name] tool abstractmethod async def think(self, user_input: str) - str: 核心思考循環處理用戶輸入可能調用工具并生成最終回復。 pass3.2 實現一個簡單的ReAct模式AgentReActReasoning Acting是一種經典的Agent推理模式。我們實現一個簡化版本。# src/my_agent/core/react_agent.py import json import re from typing import Dict, Any from .agent import BaseAgent, Tool, Memory from pydantic import BaseModel class ReasoningStep(BaseModel): thought: str action: Optional[str] None # 工具名 action_input: Optional[Dict[str, Any]] None observation: Optional[str] None final_answer: Optional[str] None class ReActAgent(BaseAgent): 一個實現ReAct推理模式的簡單Agent。 async def think(self, user_input: str) - str: # 將用戶輸入加入記憶 self.memory.add_message(user, user_input) # 構建系統提示包含工具描述 tools_description \n.join([f- {name}: {tool.description} for name, tool in self.tools.items()]) system_prompt f你是一個有幫助的AI助手可以調用工具來解決問題。 你可以使用的工具如下 {tools_description} 請遵循以下格式進行思考 Thought: 你需要思考當前情況決定是否需要使用工具以及使用哪個工具。 Action: 需要調用的工具名稱如果沒有工具可用或不需要就填 None。 Action Input: 調用工具所需的輸入參數必須是JSON格式。如果Action是None這里也填 null。 Observation: 工具執行后的結果。 ... (這個 Thought/Action/Action Input/Observation 循環可以重復多次) Thought: 我現在有足夠的信息來回答用戶了。 Final Answer: 給用戶的最終回答。 # 獲取對話上下文 context_messages self.memory.get_context() # 準備發送給LLM的消息 messages [ {role: system, content: system_prompt}, *context_messages, {role: user, content: user_input}, ] max_iterations 5 for i in range(max_iterations): # 調用LLM獲取下一步推理 llm_response await self.llm.chat.completions.create( modelgpt-3.5-turbo, # 或 gpt-4 messagesmessages, temperature0, ) response_text llm_response.choices[0].message.content # 解析LLM的響應提取 Thought, Action 等部分這里簡化實際應用需要更魯棒的解析 # 假設LLM嚴格按照格式回復 thought_match re.search(rThought:\s*(.), response_text, re.DOTALL) action_match re.search(rAction:\s*(.), response_text) action_input_match re.search(rAction Input:\s*(.), response_text, re.DOTALL) thought thought_match.group(1).strip() if thought_match else action action_match.group(1).strip() if action_match else None action_input_str action_input_match.group(1).strip() if action_input_match else null print(f[Agent Iteration {i1}] Thought: {thought}) print(f[Agent Iteration {i1}] Action: {action}) if action and action ! None: # 執行工具調用 try: action_input json.loads(action_input_str) if action_input_str ! null else {} tool self.tools.get(action) if tool: observation await tool.run(**action_input) print(f[Agent Iteration {i1}] Observation: {observation}) # 將本次行動和觀察加入消息歷史供下一輪參考 messages.append({role: assistant, content: fAction: {action}\nAction Input: {action_input_str}}) messages.append({role: user, content: fObservation: {observation}}) else: observation fError: Tool {action} not found. messages.append({role: user, content: fObservation: {observation}}) except json.JSONDecodeError: observation fError: Invalid JSON in Action Input: {action_input_str} messages.append({role: user, content: fObservation: {observation}}) except Exception as e: observation fError executing tool {action}: {str(e)} messages.append({role: user, content: fObservation: {observation}}) else: # 沒有更多行動嘗試提取最終答案 final_answer_match re.search(rFinal Answer:\s*(.), response_text, re.DOTALL) if final_answer_match: final_answer final_answer_match.group(1).strip() self.memory.add_message(assistant, final_answer) return final_answer else: # 如果沒有明確Final Answer可能LLM格式有誤直接返回其回復 self.memory.add_message(assistant, response_text) return response_text return 抱歉經過多輪推理仍未得到最終答案。3.3 集成LLM客戶端我們使用OpenAI官方Python SDK并對其進行簡單封裝以適配我們的Agent接口。# src/my_agent/llm/openai_client.py import os from openai import AsyncOpenAI from dotenv import load_dotenv load_dotenv() # 加載.env文件中的環境變量 class OpenAIClient: def __init__(self): api_key os.getenv(OPENAI_API_KEY) base_url os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) if not api_key: raise ValueError(OPENAI_API_KEY environment variable is not set.) self.client AsyncOpenAI(api_keyapi_key, base_urlbase_url) property def chat(self): # 提供一個與我們的Agent期望的接口兼容的屬性 return self.client.chat4. 開發與集成自定義工具Tools工具是Agent能力的延伸。我們來創建幾個常用工具。4.1 天氣查詢工具# src/my_agent/tools/weather_tool.py import httpx from pydantic import BaseModel, Field from ..core.agent import Tool from dotenv import load_dotenv import os load_dotenv() class WeatherInput(BaseModel): city: str Field(description城市名稱例如北京、Shanghai) class WeatherTool(Tool): def __init__(self): super().__init__( nameget_weather, description根據城市名稱查詢當前天氣情況。, args_schemaWeatherInput ) self.api_key os.getenv(WEATHER_API_KEY) # 假設你有一個天氣API的Key # 這里使用一個模擬的免費API示例實際使用時請替換為真實API self.base_url http://wttr.in/ async def run(self, city: str) - str: 調用天氣API。 try: async with httpx.AsyncClient() as client: # 注意wttr.in 是一個免費服務格式可能變化僅作示例 url f{self.base_url}{city}?format3 # 格式3返回簡短文本 response await client.get(url, timeout10.0) response.raise_for_status() weather_info response.text.strip() return f{city}的天氣是{weather_info} except httpx.RequestError as e: return f請求天氣API時出錯{str(e)} except Exception as e: return f處理天氣信息時發生未知錯誤{str(e)}4.2 計算器工具# src/my_agent/tools/calculator_tool.py from pydantic import BaseModel, Field from ..core.agent import Tool import ast import operator as op class CalculatorInput(BaseModel): expression: str Field(description一個有效的數學表達式例如(3 5) * 2) class CalculatorTool(Tool): def __init__(self): super().__init__( namecalculator, description計算一個數學表達式的結果。支持加減乘除和括號。, args_schemaCalculatorInput ) # 定義安全的運算符 self._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, } async def run(self, expression: str) - str: 安全地計算數學表達式。 try: # 使用ast.literal_eval進行安全評估 # 注意這里我們實現一個更安全的自定義評估器避免直接使用eval result self._safe_eval(expression) return f表達式 {expression} 的計算結果是{result} except (SyntaxError, ValueError, TypeError, ZeroDivisionError) as e: return f計算表達式 {expression} 時出錯{str(e)}。請確保表達式格式正確。 def _safe_eval(self, node): 遞歸安全地評估AST節點。 if isinstance(node, ast.Num): # number return node.n elif isinstance(node, ast.BinOp): # left operator right left_val self._safe_eval(node.left) right_val self._safe_eval(node.right) op_func self._allowed_operators.get(type(node.op)) if op_func is None: raise TypeError(f不允許的操作符{type(node.op)}) return op_func(left_val, right_val) elif isinstance(node, ast.UnaryOp): # operator operand e.g., -1 operand_val self._safe_eval(node.operand) op_func self._allowed_operators.get(type(node.op)) if op_func is None: raise TypeError(f不允許的一元操作符{type(node.op)}) return op_func(operand_val) elif isinstance(node, ast.Constant): # Python 3.8 常量 return node.value else: raise TypeError(f不支持的AST節點類型{type(node)}) def _safe_eval(self, expr: str): 入口函數將字符串表達式解析為AST并安全評估。 tree ast.parse(expr, modeeval) return self._safe_eval(tree.body) # 注意這里遞歸調用的是上面的方法需要重命名避免歧義。實際代碼中應調整。修正上面的遞歸問題將內部方法重命名def _eval_node(self, node): if isinstance(node, ast.Num): return node.n elif isinstance(node, ast.Constant): return node.value elif isinstance(node, ast.BinOp): left_val self._eval_node(node.left) right_val self._eval_node(node.right) op_func self._allowed_operators.get(type(node.op)) if op_func is None: raise TypeError(f不允許的操作符{type(node.op)}) return op_func(left_val, right_val) elif isinstance(node, ast.UnaryOp): operand_val self._eval_node(node.operand) op_func self._allowed_operators.get(type(node.op)) if op_func is None: raise TypeError(f不允許的一元操作符{type(node.op)}) return op_func(operand_val) else: raise TypeError(f不支持的AST節點類型{type(node)}) async def run(self, expression: str) - str: try: tree ast.parse(expression, modeeval) result self._eval_node(tree.body) return f表達式 {expression} 的計算結果是{result} except (SyntaxError, ValueError, TypeError, ZeroDivisionError, AttributeError) as e: return f計算表達式 {expression} 時出錯{str(e)}。請確保表達式格式正確且僅包含基本算術運算。5. 組裝并運行你的第一個Agent現在讓我們將各個部分組裝起來創建一個可以對話的Agent。5.1 創建主運行腳本# run_agent.py import asyncio import sys from src.my_agent.llm.openai_client import OpenAIClient from src.my_agent.core.react_agent import ReActAgent from src.my_agent.tools.weather_tool import WeatherTool from src.my_agent.tools.calculator_tool import CalculatorTool async def main(): # 1. 初始化LLM客戶端 llm_client OpenAIClient() # 2. 創建Agent實例 agent ReActAgent(llm_clientllm_client) # 3. 注冊工具 agent.register_tool(WeatherTool()) agent.register_tool(CalculatorTool()) print(智能體已啟動輸入 quit 或 exit 退出。) print(- * 40) while True: try: user_input input(\nYou: ).strip() if user_input.lower() in [quit, exit]: print(再見) break if not user_input: continue # 4. 讓Agent思考并回復 response await agent.think(user_input) print(f\nAgent: {response}) except KeyboardInterrupt: print(\n\n程序被中斷。) break except Exception as e: print(f\n發生錯誤{e}) if __name__ __main__: asyncio.run(main())5.2 運行與測試在項目根目錄下運行python run_agent.py你應該會看到提示符。嘗試輸入“北京今天天氣怎么樣”Agent會調用天氣工具“計算一下 (12 34) * 2 等于多少”Agent會調用計算器工具“你是誰”Agent會直接利用LLM知識回答觀察控制臺輸出的Thought、Action、Observation日志理解ReAct模式的運行過程。6. 工程化進階構建工具鏈的其他關鍵環節一個基礎的Agent跑起來了但要將其工程化我們還需要完善以下環節。6.1 工具的動態加載與發現手動注冊工具在工具數量多時會很麻煩。我們可以實現一個工具發現機制。# src/my_agent/core/tool_registry.py import importlib import pkgutil from pathlib import Path from typing import Dict, Type from .agent import Tool class ToolRegistry: _tools: Dict[str, Type[Tool]] {} classmethod def register(cls, tool_class: Type[Tool]): 類裝飾器用于注冊工具類。 instance tool_class() cls._tools[instance.name] tool_class return tool_class classmethod def discover_tools(cls, package_path: str): 自動發現指定包路徑下所有繼承了Tool的類并注冊。 package importlib.import_module(package_path) for _, module_name, is_pkg in pkgutil.iter_modules(package.__path__, package.__name__ .): if not is_pkg: module importlib.import_module(module_name) for attr_name in dir(module): attr getattr(module, attr_name) if (isinstance(attr, type) and issubclass(attr, Tool) and attr ! Tool): # 排除基類本身 cls.register(attr) classmethod def get_tool_instance(cls, tool_name: str) - Tool: 根據工具名獲取工具實例。 tool_class cls._tools.get(tool_name) if tool_class: return tool_class() raise KeyError(fTool {tool_name} not found in registry.) classmethod def get_all_tool_descriptions(cls) - Dict[str, str]: 獲取所有已注冊工具的描述。 return {name: cls.get_tool_instance(name).description for name in cls._tools.keys()}然后我們可以用裝飾器來聲明工具# src/my_agent/tools/weather_tool.py from src.my_agent.core.tool_registry import ToolRegistry ToolRegistry.register class WeatherTool(Tool): # ... 其余代碼不變 ...在主程序中可以自動加載所有工具# run_agent_auto.py from src.my_agent.core.tool_registry import ToolRegistry # ... 其他導入 ... async def main(): # 自動發現并注冊 src.my_agent.tools 包下的所有工具 ToolRegistry.discover_tools(src.my_agent.tools) llm_client OpenAIClient() agent ReActAgent(llm_clientllm_client) # 從注冊表獲取所有工具實例并注冊到Agent for tool_name in ToolRegistry._tools.keys(): agent.register_tool(ToolRegistry.get_tool_instance(tool_name)) # ... 其余代碼 ...6.2 記憶Memory的持久化當前的Memory類只在內存中保存對話。生產環境需要持久化到數據庫如Redis、SQLite或向量數據庫用于長上下文摘要。# src/my_agent/core/persistent_memory.py import json from typing import List, Dict, Any from pydantic import BaseModel import sqlite3 from datetime import datetime class PersistentMemory(BaseModel): session_id: str db_path: str agent_memory.db class Config: arbitrary_types_allowed True def __init__(self, session_id: str, **data): super().__init__(session_idsession_id, **data) self._init_db() def _init_db(self): conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS message_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def add_message(self, role: str, content: str): conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( INSERT INTO message_history (session_id, role, content) VALUES (?, ?, ?), (self.session_id, role, content) ) conn.commit() conn.close() def get_context(self, max_messages: int 10) - List[Dict[str, Any]]: conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( SELECT role, content FROM message_history WHERE session_id ? ORDER BY timestamp DESC LIMIT ?, (self.session_id, max_messages) ) rows cursor.fetchall() conn.close() # 返回時按時間順序從舊到新 messages [{role: row[0], content: row[1]} for row in reversed(rows)] return messages def clear_session(self): conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute(DELETE FROM message_history WHERE session_id ?, (self.session_id,)) conn.commit() conn.close()6.3 添加API服務層FastAPI要集成到現有系統需要提供HTTP API。# src/my_agent/api/server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from contextlib import asynccontextmanager from ..core.react_agent import ReActAgent from ..llm.openai_client import OpenAIClient from ..core.tool_registry import ToolRegistry import uuid # 全局Agent實例簡單示例生產環境需考慮并發和狀態隔離 _agent None asynccontextmanager async def lifespan(app: FastAPI): # 啟動時初始化 global _agent ToolRegistry.discover_tools(src.my_agent.tools) llm_client OpenAIClient() _agent ReActAgent(llm_clientllm_client) for tool_name in ToolRegistry._tools.keys(): _agent.register_tool(ToolRegistry.get_tool_instance(tool_name)) print(Agent initialized.) yield # 關閉時清理 print(Shutting down.) app FastAPI(lifespanlifespan) class ChatRequest(BaseModel): session_id: str None # 為空則創建新會話 message: str class ChatResponse(BaseModel): session_id: str reply: str app.post(/chat, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): if _agent is None: raise HTTPException(status_code503, detailAgent not initialized) # 這里簡化處理實際應將Memory與session_id綁定 session_id request.session_id or str(uuid.uuid4()) # TODO: 根據session_id從數據庫加載或創建PersistentMemory reply await _agent.think(request.message) return ChatResponse(session_idsession_id, replyreply) app.get(/health) async def health_check(): return {status: healthy}使用Uvicorn運行pip install fastapi uvicorn[standard] uvicorn src.my_agent.api.server:app --host 0.0.0.0 --port 8000 --reload6.4 配置管理Pydantic Settings使用Pydantic Settings管理所有配置。# src/my_agent/config/settings.py from pydantic_settings import BaseSettings from pydantic import Field class Settings(BaseSettings): openai_api_key: str Field(..., envOPENAI_API_KEY) openai_base_url: str Field(https://api.openai.com/v1, envOPENAI_BASE_URL) weather_api_key: str Field(, envWEATHER_API_KEY) database_url: str Field(sqlite:///./agent.db, envDATABASE_URL) log_level: str Field(INFO, envLOG_LEVEL) class Config: env_file .env extra ignore # 忽略.env中未定義的變量 settings Settings()7. 常見問題與排查思路在搭建和運行過程中你可能會遇到以下問題問題現象可能原因排查步驟與解決方案導入錯誤ModuleNotFoundError1. 虛擬環境未激活。2. 項目未以可編輯模式安裝。3.PYTHONPATH未包含項目根目錄。1. 確認命令行前有(venv)。2. 在項目根目錄執行pip install -e .。3. 在IDE中設置正確的項目根目錄和解釋器。OpenAI API 調用失敗1. API Key 未設置或錯誤。2. 網絡問題或代理配置。3. 余額不足或速率限制。1. 檢查.env文件中的OPENAI_API_KEY。2. 檢查網絡連接如需代理在代碼中配置http_client。3. 查看OpenAI控制臺賬單和用量。Agent 不調用工具直接回答1. 系統提示詞Prompt中工具描述不清晰。2. LLM 溫度temperature設置過高導致輸出不穩定。3. 工具名稱或描述與用戶問題匹配度低。1. 優化系統提示詞明確指令格式。2. 將temperature設為0確保確定性輸出。3. 檢查工具描述是否準確嘗試用更直接的問題測試。工具調用參數解析錯誤1. LLM 生成的Action Input不是合法JSON。2. JSON中的參數名與工具定義的args_schema不匹配。1. 在Agent代碼中增加更健壯的JSON解析和錯誤處理。2. 在工具描述中明確參數名稱和類型。可以使用Pydantic的schema_json()為LLM提供更精確的格式。多輪對話狀態丟失1.Memory類未正確集成到Agent中。2. 每次請求創建了新的Agent實例。1. 確保agent.think()方法中正確讀取和更新了self.memory。2. 對于Web服務需要將會話ID與Memory實例綁定并持久化存儲。性能問題響應慢1. 工具調用是同步的阻塞了主線程。2. LLM API調用耗時過長。3. 未實現流式輸出。1. 確保所有工具方法都是async并使用await調用。2. 考慮設置合理的超時時間或使用更快的模型。3. 對于Web API可以研究SSEServer-Sent Events實現流式響應。8. 最佳實踐與工程化建議將Agent投入生產環境需要遵循以下工程化準則提示詞工程化將系統提示詞、用戶提示詞模板等抽取到配置文件或數據庫中便于管理和A/B測試。對提示詞進行版本控制。使用Jinja2等模板引擎動態生成提示詞。工具開發的標準化為所有工具編寫清晰的文檔包括輸入/輸出格式、錯誤碼。工具函數內部必須有完善的錯誤處理和日志記錄。為工具編寫單元測試和集成測試。可觀測性與監控在Agent的每個關鍵步驟接收輸入、調用LLM、調用工具、返回輸出記錄結構化日志。記錄每次LLM調用的輸入Token、輸出Token數量及成本。使用像Prometheus和Grafana監控工具調用成功率、延遲和Agent整體響應時間。安全與權限工具權限控制不是所有用戶都能調用所有工具。實現一個權限層根據用戶身份或會話上下文決定可用的工具集。輸入輸出過濾對用戶輸入和工具返回的內容進行安全檢查防止Prompt注入、敏感信息泄露。沙箱環境對于執行代碼、訪問文件系統等高危工具必須在安全的沙箱環境中運行。測試策略單元測試測試每個工具函數的邏輯。集成測試測試Agent與LLM、工具的集成流程可以使用LLM的Mock來避免真實API調用。端到端測試模擬真實用戶場景測試完整的對話流。部署與運維容器化使用Docker將Agent及其依賴打包確保環境一致性。配置分離所有密鑰、端點URL等配置必須通過環境變量或配置中心管理絕不能硬編碼。健康檢查與就緒探針為Web服務添加/health端點便于K8s等編排系統管理。版本回滾Agent的代碼、模型版本、提示詞版本都應有明確的版本號支持快速回滾。通過以上步驟你不僅搭建了一個可運行的智能體更構建了一套支撐其持續迭代和穩定運行的工程化工具鏈雛形。這套工具鏈的核心思想是模塊化、可觀測、可測試、可部署。你可以在此基礎上繼續擴展工作流引擎、可視化編排界面、更復雜的記憶模塊如向量數據庫逐步將其打造成團隊內部強大的AI能力中臺。