工具鏈：開源組件構(gòu)建可控AI工程化方案)
最近在嘗試將AI能力集成到業(yè)務(wù)系統(tǒng)中時發(fā)現(xiàn)市面上的智能體平臺雖然功能強大但要么是黑盒要么定制成本極高要么就是難以與現(xiàn)有開發(fā)流程和工具鏈深度集成。對于希望將智能體能力“工程化”落地的團隊來說從零理解其核心并搭建一套可控、可擴展、可集成的開發(fā)工具鏈?zhǔn)潜亟?jīng)之路。本文將從零開始手把手帶你搭建一套專屬于你自己的智能體Agent開發(fā)工具鏈。我們將不依賴任何大型商業(yè)平臺而是基于開源組件和標(biāo)準(zhǔn)協(xié)議構(gòu)建一個從環(huán)境配置、核心框架、工具集成到工程化部署的完整閉環(huán)。無論你是想深入理解Agent的內(nèi)部機制還是希望為團隊打造一套標(biāo)準(zhǔn)化的AI開發(fā)基礎(chǔ)設(shè)施這篇文章都將提供一套可直接復(fù)用的實戰(zhàn)方案。1. 智能體Agent開發(fā)的核心概念與工程化挑戰(zhàn)在開始動手之前我們必須明確幾個核心概念并理解為什么需要一套工具鏈而不是簡單地調(diào)用一個API。1.1 什么是智能體Agent在AI語境下一個智能體Agent通常指一個能夠感知環(huán)境、進行決策并執(zhí)行行動以實現(xiàn)特定目標(biāo)的軟件實體。與傳統(tǒng)的“聊天機器人”或“問答系統(tǒng)”不同一個真正的Agent具備幾個關(guān)鍵特征自主性Autonomy能在沒有人類直接干預(yù)的情況下運行。反應(yīng)性Reactivity能感知環(huán)境如用戶輸入、API返回、數(shù)據(jù)庫變化并做出及時響應(yīng)。主動性Pro-activeness不僅被動響應(yīng)還能主動發(fā)起目標(biāo)導(dǎo)向的行為。社交能力Social Ability能與其他Agent或人類進行交互和協(xié)作。當(dāng)前基于大語言模型LLM的Agent是其最流行的實現(xiàn)形式。LLM作為其“大腦”負責(zé)理解、規(guī)劃和決策而外部的“工具”Tools則成為其“手腳”用于執(zhí)行具體的操作如查詢數(shù)據(jù)庫、調(diào)用API、運行代碼等。1.2 為什么需要“工具鏈”而非“單點方案”很多開發(fā)者初涉Agent開發(fā)時會從一個簡單的腳本開始接收用戶輸入調(diào)用LLM API解析返回結(jié)果然后執(zhí)行某個操作。但隨著需求復(fù)雜化這種模式會迅速陷入困境工具管理混亂工具函數(shù)散落在各處缺乏統(tǒng)一的注冊、描述和調(diào)用機制。狀態(tài)管理困難Agent與用戶的多次對話多輪對話狀態(tài)如何保存和恢復(fù)流程編排缺失復(fù)雜的任務(wù)需要多個Agent協(xié)作或按特定工作流執(zhí)行代碼會變得極其臃腫?？捎^測性差A(yù)gent內(nèi)部如何思考、為什么選擇某個工具、執(zhí)行結(jié)果如何這些過程如同黑盒難以調(diào)試和優(yōu)化。工程化部署難如何將開發(fā)好的Agent打包、部署、監(jiān)控、擴縮容并與現(xiàn)有CI/CD流程集成因此一套完整的Agent開發(fā)工具鏈旨在系統(tǒng)性地解決上述問題將Agent開發(fā)從“腳本編寫”升級為“軟件工程”。1.3 工具鏈的核心組件我們計劃構(gòu)建的工具鏈將包含以下核心層這也是本文的實踐路線圖環(huán)境與基礎(chǔ)層Python環(huán)境、虛擬環(huán)境管理、依賴管理。核心框架層選擇或自建一個輕量級Agent核心框架負責(zé)大腦LLM的調(diào)用、工具的管理與調(diào)度、記憶對話歷史的維護。工具集成層標(biāo)準(zhǔn)化工具的封裝、注冊與調(diào)用接口。編排與工作流層實現(xiàn)多個Agent的協(xié)作和復(fù)雜任務(wù)的流程控制。工程化與部署層日志、監(jiān)控、配置管理、容器化部署。2. 環(huán)境準(zhǔn)備與項目初始化我們選擇Python作為主要開發(fā)語言因其在AI生態(tài)中擁有最豐富的庫支持。2.1 基礎(chǔ)環(huán)境配置首先確保你的系統(tǒng)已安裝Python推薦3.9或以上版本和pip。然后為項目創(chuàng)建一個獨立的虛擬環(huán)境這是管理依賴的最佳實踐。# 創(chuàng)建項目目錄 mkdir my_agent_toolchain cd my_agent_toolchain # 創(chuàng)建Python虛擬環(huán)境使用venv python -m venv venv # 激活虛擬環(huán)境 # 在Windows上 venv\Scripts\activate # 在Linux/Mac上 source venv/bin/activate激活后你的命令行提示符前會出現(xiàn)(venv)標(biāo)識。2.2 初始化項目結(jié)構(gòu)與依賴管理我們使用pyproject.toml現(xiàn)代Python項目標(biāo)準(zhǔn)來管理依賴和項目元數(shù)據(jù)。# 創(chuàng)建基礎(chǔ)項目結(jié)構(gòu) 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, # 用于調(diào)用OpenAI API或其他兼容API langchain-core0.1.0, # 使用LangChain的核心抽象但不一定用其全量框架 pydantic2.0.0, # 用于數(shù)據(jù)驗證和設(shè)置管理 httpx0.25.0, # 異步HTTP客戶端用于工具調(diào)用 python-dotenv1.0.0, # 從.env文件加載環(huán)境變量 ] [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然后安裝基礎(chǔ)依賴pip install -e . # 以可編輯模式安裝當(dāng)前項目創(chuàng)建.env文件來存儲敏感信息如API密鑰切記不要將其提交到版本控制系統(tǒng)。# .env OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果使用其他兼容服務(wù)可修改此處編輯.gitignore文件忽略虛擬環(huán)境、緩存文件和.env。# .gitignore venv/ __pycache__/ *.py[cod] .env .pytest_cache/ .coverage3. 構(gòu)建核心Agent框架我們不直接使用龐大的全功能框架而是基于清晰的概念自建核心這有助于深刻理解Agent的運行機制。3.1 定義核心抽象Agent、Tool、Memory在src/my_agent/core目錄下創(chuàng)建基礎(chǔ)抽象類。# 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工具參數(shù)的Pydantic模型) abstractmethod async def run(self, **kwargs) - str: 執(zhí)行工具的核心方法。 pass class Memory(BaseModel): 記憶基類負責(zé)存儲和檢索對話歷史。 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]]: 獲取最近的對話上下文用于發(fā)送給LLM。 # 簡單的實現(xiàn)返回全部消息生產(chǎn)環(huán)境需實現(xiàn)Token計數(shù)和截斷 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: 核心思考循環(huán)處理用戶輸入可能調(diào)用工具并生成最終回復(fù)。 pass3.2 實現(xiàn)一個簡單的ReAct模式AgentReActReasoning Acting是一種經(jīng)典的Agent推理模式。我們實現(xiàn)一個簡化版本。# 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): 一個實現(xiàn)ReAct推理模式的簡單Agent。 async def think(self, user_input: str) - str: # 將用戶輸入加入記憶 self.memory.add_message(user, user_input) # 構(gòu)建系統(tǒng)提示包含工具描述 tools_description \n.join([f- {name}: {tool.description} for name, tool in self.tools.items()]) system_prompt f你是一個有幫助的AI助手可以調(diào)用工具來解決問題。 你可以使用的工具如下 {tools_description} 請遵循以下格式進行思考 Thought: 你需要思考當(dāng)前情況決定是否需要使用工具以及使用哪個工具。 Action: 需要調(diào)用的工具名稱如果沒有工具可用或不需要就填 None。 Action Input: 調(diào)用工具所需的輸入?yún)?shù)必須是JSON格式。如果Action是None這里也填 null。 Observation: 工具執(zhí)行后的結(jié)果。 ... (這個 Thought/Action/Action Input/Observation 循環(huán)可以重復(fù)多次) Thought: 我現(xiàn)在有足夠的信息來回答用戶了。 Final Answer: 給用戶的最終回答。 # 獲取對話上下文 context_messages self.memory.get_context() # 準(zhǔn)備發(fā)送給LLM的消息 messages [ {role: system, content: system_prompt}, *context_messages, {role: user, content: user_input}, ] max_iterations 5 for i in range(max_iterations): # 調(diào)用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的響應(yīng)提取 Thought, Action 等部分這里簡化實際應(yīng)用需要更魯棒的解析 # 假設(shè)LLM嚴(yán)格按照格式回復(fù) 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: # 執(zhí)行工具調(diào)用 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格式有誤直接返回其回復(fù) self.memory.add_message(assistant, response_text) return response_text return 抱歉經(jīng)過多輪推理仍未得到最終答案。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文件中的環(huán)境變量 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. 開發(fā)與集成自定義工具Tools工具是Agent能力的延伸。我們來創(chuàng)建幾個常用工具。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根據(jù)城市名稱查詢當(dāng)前天氣情況。, args_schemaWeatherInput ) self.api_key os.getenv(WEATHER_API_KEY) # 假設(shè)你有一個天氣API的Key # 這里使用一個模擬的免費API示例實際使用時請?zhí)鎿Q為真實API self.base_url http://wttr.in/ async def run(self, city: str) - str: 調(diào)用天氣API。 try: async with httpx.AsyncClient() as client: # 注意wttr.in 是一個免費服務(wù)格式可能變化僅作示例 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處理天氣信息時發(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一個有效的數(shù)學(xué)表達式例如(3 5) * 2) class CalculatorTool(Tool): def __init__(self): super().__init__( namecalculator, description計算一個數(shù)學(xué)表達式的結(jié)果。支持加減乘除和括號。, 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: 安全地計算數(shù)學(xué)表達式。 try: # 使用ast.literal_eval進行安全評估 # 注意這里我們實現(xiàn)一個更安全的自定義評估器避免直接使用eval result self._safe_eval(expression) return f表達式 {expression} 的計算結(jié)果是{result} except (SyntaxError, ValueError, TypeError, ZeroDivisionError) as e: return f計算表達式 {expression} 時出錯{str(e)}。請確保表達式格式正確。 def _safe_eval(self, node): 遞歸安全地評估AST節(jié)點。 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節(jié)點類型{type(node)}) def _safe_eval(self, expr: str): 入口函數(shù)將字符串表達式解析為AST并安全評估。 tree ast.parse(expr, modeeval) return self._safe_eval(tree.body) # 注意這里遞歸調(diào)用的是上面的方法需要重命名避免歧義。實際代碼中應(yīng)調(diào)整。修正上面的遞歸問題將內(nèi)部方法重命名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節(jié)點類型{type(node)}) async def run(self, expression: str) - str: try: tree ast.parse(expression, modeeval) result self._eval_node(tree.body) return f表達式 {expression} 的計算結(jié)果是{result} except (SyntaxError, ValueError, TypeError, ZeroDivisionError, AttributeError) as e: return f計算表達式 {expression} 時出錯{str(e)}。請確保表達式格式正確且僅包含基本算術(shù)運算。5. 組裝并運行你的第一個Agent現(xiàn)在讓我們將各個部分組裝起來創(chuàng)建一個可以對話的Agent。5.1 創(chuàng)建主運行腳本# 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. 創(chuàng)建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思考并回復(fù) response await agent.think(user_input) print(f\nAgent: {response}) except KeyboardInterrupt: print(\n\n程序被中斷。) break except Exception as e: print(f\n發(fā)生錯誤{e}) if __name__ __main__: asyncio.run(main())5.2 運行與測試在項目根目錄下運行python run_agent.py你應(yīng)該會看到提示符。嘗試輸入“北京今天天氣怎么樣”Agent會調(diào)用天氣工具“計算一下 (12 34) * 2 等于多少”Agent會調(diào)用計算器工具“你是誰”Agent會直接利用LLM知識回答觀察控制臺輸出的Thought、Action、Observation日志理解ReAct模式的運行過程。6. 工程化進階構(gòu)建工具鏈的其他關(guān)鍵環(huán)節(jié)一個基礎(chǔ)的Agent跑起來了但要將其工程化我們還需要完善以下環(huán)節(jié)。6.1 工具的動態(tài)加載與發(fā)現(xiàn)手動注冊工具在工具數(shù)量多時會很麻煩。我們可以實現(xiàn)一個工具發(fā)現(xiàn)機制。# 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): 自動發(fā)現(xiàn)指定包路徑下所有繼承了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: 根據(jù)工具名獲取工具實例。 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 # ... 其他導(dǎo)入 ... async def main(): # 自動發(fā)現(xiàn)并注冊 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的持久化當(dāng)前的Memory類只在內(nèi)存中保存對話。生產(chǎn)環(huán)境需要持久化到數(shù)據(jù)庫如Redis、SQLite或向量數(shù)據(jù)庫用于長上下文摘要。# 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服務(wù)層FastAPI要集成到現(xiàn)有系統(tǒng)需要提供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實例簡單示例生產(chǎn)環(huán)境需考慮并發(fā)和狀態(tài)隔離 _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 # 關(guān)閉時清理 print(Shutting down.) app FastAPI(lifespanlifespan) class ChatRequest(BaseModel): session_id: str None # 為空則創(chuàng)建新會話 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) # 這里簡化處理實際應(yīng)將Memory與session_id綁定 session_id request.session_id or str(uuid.uuid4()) # TODO: 根據(jù)session_id從數(shù)據(jù)庫加載或創(chuàng)建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. 常見問題與排查思路在搭建和運行過程中你可能會遇到以下問題問題現(xiàn)象可能原因排查步驟與解決方案導(dǎo)入錯誤ModuleNotFoundError1. 虛擬環(huán)境未激活。2. 項目未以可編輯模式安裝。3.PYTHONPATH未包含項目根目錄。1. 確認命令行前有(venv)。2. 在項目根目錄執(zhí)行pip install -e .。3. 在IDE中設(shè)置正確的項目根目錄和解釋器。OpenAI API 調(diào)用失敗1. API Key 未設(shè)置或錯誤。2. 網(wǎng)絡(luò)問題或代理配置。3. 余額不足或速率限制。1. 檢查.env文件中的OPENAI_API_KEY。2. 檢查網(wǎng)絡(luò)連接如需代理在代碼中配置http_client。3. 查看OpenAI控制臺賬單和用量。Agent 不調(diào)用工具直接回答1. 系統(tǒng)提示詞Prompt中工具描述不清晰。2. LLM 溫度temperature設(shè)置過高導(dǎo)致輸出不穩(wěn)定。3. 工具名稱或描述與用戶問題匹配度低。1. 優(yōu)化系統(tǒng)提示詞明確指令格式。2. 將temperature設(shè)為0確保確定性輸出。3. 檢查工具描述是否準(zhǔn)確嘗試用更直接的問題測試。工具調(diào)用參數(shù)解析錯誤1. LLM 生成的Action Input不是合法JSON。2. JSON中的參數(shù)名與工具定義的args_schema不匹配。1. 在Agent代碼中增加更健壯的JSON解析和錯誤處理。2. 在工具描述中明確參數(shù)名稱和類型?？梢允褂肞ydantic的schema_json()為LLM提供更精確的格式。多輪對話狀態(tài)丟失1.Memory類未正確集成到Agent中。2. 每次請求創(chuàng)建了新的Agent實例。1. 確保agent.think()方法中正確讀取和更新了self.memory。2. 對于Web服務(wù)需要將會話ID與Memory實例綁定并持久化存儲。性能問題響應(yīng)慢1. 工具調(diào)用是同步的阻塞了主線程。2. LLM API調(diào)用耗時過長。3. 未實現(xiàn)流式輸出。1. 確保所有工具方法都是async并使用await調(diào)用。2. 考慮設(shè)置合理的超時時間或使用更快的模型。3. 對于Web API可以研究SSEServer-Sent Events實現(xiàn)流式響應(yīng)。8. 最佳實踐與工程化建議將Agent投入生產(chǎn)環(huán)境需要遵循以下工程化準(zhǔn)則提示詞工程化將系統(tǒng)提示詞、用戶提示詞模板等抽取到配置文件或數(shù)據(jù)庫中便于管理和A/B測試。對提示詞進行版本控制。使用Jinja2等模板引擎動態(tài)生成提示詞。工具開發(fā)的標(biāo)準(zhǔn)化為所有工具編寫清晰的文檔包括輸入/輸出格式、錯誤碼。工具函數(shù)內(nèi)部必須有完善的錯誤處理和日志記錄。為工具編寫單元測試和集成測試。可觀測性與監(jiān)控在Agent的每個關(guān)鍵步驟接收輸入、調(diào)用LLM、調(diào)用工具、返回輸出記錄結(jié)構(gòu)化日志。記錄每次LLM調(diào)用的輸入Token、輸出Token數(shù)量及成本。使用像Prometheus和Grafana監(jiān)控工具調(diào)用成功率、延遲和Agent整體響應(yīng)時間。安全與權(quán)限工具權(quán)限控制不是所有用戶都能調(diào)用所有工具。實現(xiàn)一個權(quán)限層根據(jù)用戶身份或會話上下文決定可用的工具集。輸入輸出過濾對用戶輸入和工具返回的內(nèi)容進行安全檢查防止Prompt注入、敏感信息泄露。沙箱環(huán)境對于執(zhí)行代碼、訪問文件系統(tǒng)等高危工具必須在安全的沙箱環(huán)境中運行。測試策略單元測試測試每個工具函數(shù)的邏輯。集成測試測試Agent與LLM、工具的集成流程可以使用LLM的Mock來避免真實API調(diào)用。端到端測試模擬真實用戶場景測試完整的對話流。部署與運維容器化使用Docker將Agent及其依賴打包確保環(huán)境一致性。配置分離所有密鑰、端點URL等配置必須通過環(huán)境變量或配置中心管理絕不能硬編碼。健康檢查與就緒探針為Web服務(wù)添加/health端點便于K8s等編排系統(tǒng)管理。版本回滾Agent的代碼、模型版本、提示詞版本都應(yīng)有明確的版本號支持快速回滾。通過以上步驟你不僅搭建了一個可運行的智能體更構(gòu)建了一套支撐其持續(xù)迭代和穩(wěn)定運行的工程化工具鏈雛形。這套工具鏈的核心思想是模塊化、可觀測、可測試、可部署。你可以在此基礎(chǔ)上繼續(xù)擴展工作流引擎、可視化編排界面、更復(fù)雜的記憶模塊如向量數(shù)據(jù)庫逐步將其打造成團隊內(nèi)部強大的AI能力中臺。 從零搭建智能體開發(fā)工具鏈：開源組件構(gòu)建可控AI工程化方案
最近在嘗試將AI能力集成到業(yè)務(wù)系統(tǒng)中時發(fā)現(xiàn)市面上的智能體平臺雖然功能強大但要么是黑盒要么定制成本極高要么就是難以與現(xiàn)有開發(fā)流程和工具鏈深度集成。對于希望將智能體能力“工程化”落地的團隊來說從零理解其核心并搭建一套可控、可擴展、可集成的開發(fā)工具鏈?zhǔn)潜亟?jīng)之路。本文將從零開始手把手帶你搭建一套專屬于你自己的智能體Agent開發(fā)工具鏈。我們將不依賴任何大型商業(yè)平臺而是基于開源組件和標(biāo)準(zhǔn)協(xié)議構(gòu)建一個從環(huán)境配置、核心框架、工具集成到工程化部署的完整閉環(huán)。無論你是想深入理解Agent的內(nèi)部機制還是希望為團隊打造一套標(biāo)準(zhǔn)化的AI開發(fā)基礎(chǔ)設(shè)施這篇文章都將提供一套可直接復(fù)用的實戰(zhàn)方案。1. 智能體Agent開發(fā)的核心概念與工程化挑戰(zhàn)在開始動手之前我們必須明確幾個核心概念并理解為什么需要一套工具鏈而不是簡單地調(diào)用一個API。1.1 什么是智能體Agent在AI語境下一個智能體Agent通常指一個能夠感知環(huán)境、進行決策并執(zhí)行行動以實現(xiàn)特定目標(biāo)的軟件實體。與傳統(tǒng)的“聊天機器人”或“問答系統(tǒng)”不同一個真正的Agent具備幾個關(guān)鍵特征自主性Autonomy能在沒有人類直接干預(yù)的情況下運行。反應(yīng)性Reactivity能感知環(huán)境如用戶輸入、API返回、數(shù)據(jù)庫變化并做出及時響應(yīng)。主動性Pro-activeness不僅被動響應(yīng)還能主動發(fā)起目標(biāo)導(dǎo)向的行為。社交能力Social Ability能與其他Agent或人類進行交互和協(xié)作。當(dāng)前基于大語言模型LLM的Agent是其最流行的實現(xiàn)形式。LLM作為其“大腦”負責(zé)理解、規(guī)劃和決策而外部的“工具”Tools則成為其“手腳”用于執(zhí)行具體的操作如查詢數(shù)據(jù)庫、調(diào)用API、運行代碼等。1.2 為什么需要“工具鏈”而非“單點方案”很多開發(fā)者初涉Agent開發(fā)時會從一個簡單的腳本開始接收用戶輸入調(diào)用LLM API解析返回結(jié)果然后執(zhí)行某個操作。但隨著需求復(fù)雜化這種模式會迅速陷入困境工具管理混亂工具函數(shù)散落在各處缺乏統(tǒng)一的注冊、描述和調(diào)用機制。狀態(tài)管理困難Agent與用戶的多次對話多輪對話狀態(tài)如何保存和恢復(fù)流程編排缺失復(fù)雜的任務(wù)需要多個Agent協(xié)作或按特定工作流執(zhí)行代碼會變得極其臃腫?？捎^測性差A(yù)gent內(nèi)部如何思考、為什么選擇某個工具、執(zhí)行結(jié)果如何這些過程如同黑盒難以調(diào)試和優(yōu)化。工程化部署難如何將開發(fā)好的Agent打包、部署、監(jiān)控、擴縮容并與現(xiàn)有CI/CD流程集成因此一套完整的Agent開發(fā)工具鏈旨在系統(tǒng)性地解決上述問題將Agent開發(fā)從“腳本編寫”升級為“軟件工程”。1.3 工具鏈的核心組件我們計劃構(gòu)建的工具鏈將包含以下核心層這也是本文的實踐路線圖環(huán)境與基礎(chǔ)層Python環(huán)境、虛擬環(huán)境管理、依賴管理。核心框架層選擇或自建一個輕量級Agent核心框架負責(zé)大腦LLM的調(diào)用、工具的管理與調(diào)度、記憶對話歷史的維護。工具集成層標(biāo)準(zhǔn)化工具的封裝、注冊與調(diào)用接口。編排與工作流層實現(xiàn)多個Agent的協(xié)作和復(fù)雜任務(wù)的流程控制。工程化與部署層日志、監(jiān)控、配置管理、容器化部署。2. 環(huán)境準(zhǔn)備與項目初始化我們選擇Python作為主要開發(fā)語言因其在AI生態(tài)中擁有最豐富的庫支持。2.1 基礎(chǔ)環(huán)境配置首先確保你的系統(tǒng)已安裝Python推薦3.9或以上版本和pip。然后為項目創(chuàng)建一個獨立的虛擬環(huán)境這是管理依賴的最佳實踐。# 創(chuàng)建項目目錄 mkdir my_agent_toolchain cd my_agent_toolchain # 創(chuàng)建Python虛擬環(huán)境使用venv python -m venv venv # 激活虛擬環(huán)境 # 在Windows上 venv\Scripts\activate # 在Linux/Mac上 source venv/bin/activate激活后你的命令行提示符前會出現(xiàn)(venv)標(biāo)識。2.2 初始化項目結(jié)構(gòu)與依賴管理我們使用pyproject.toml現(xiàn)代Python項目標(biāo)準(zhǔn)來管理依賴和項目元數(shù)據(jù)。# 創(chuàng)建基礎(chǔ)項目結(jié)構(gòu) 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, # 用于調(diào)用OpenAI API或其他兼容API langchain-core0.1.0, # 使用LangChain的核心抽象但不一定用其全量框架 pydantic2.0.0, # 用于數(shù)據(jù)驗證和設(shè)置管理 httpx0.25.0, # 異步HTTP客戶端用于工具調(diào)用 python-dotenv1.0.0, # 從.env文件加載環(huán)境變量 ] [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然后安裝基礎(chǔ)依賴pip install -e . # 以可編輯模式安裝當(dāng)前項目創(chuàng)建.env文件來存儲敏感信息如API密鑰切記不要將其提交到版本控制系統(tǒng)。# .env OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果使用其他兼容服務(wù)可修改此處編輯.gitignore文件忽略虛擬環(huán)境、緩存文件和.env。# .gitignore venv/ __pycache__/ *.py[cod] .env .pytest_cache/ .coverage3. 構(gòu)建核心Agent框架我們不直接使用龐大的全功能框架而是基于清晰的概念自建核心這有助于深刻理解Agent的運行機制。3.1 定義核心抽象Agent、Tool、Memory在src/my_agent/core目錄下創(chuàng)建基礎(chǔ)抽象類。# 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工具參數(shù)的Pydantic模型) abstractmethod async def run(self, **kwargs) - str: 執(zhí)行工具的核心方法。 pass class Memory(BaseModel): 記憶基類負責(zé)存儲和檢索對話歷史。 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]]: 獲取最近的對話上下文用于發(fā)送給LLM。 # 簡單的實現(xiàn)返回全部消息生產(chǎn)環(huán)境需實現(xiàn)Token計數(shù)和截斷 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: 核心思考循環(huán)處理用戶輸入可能調(diào)用工具并生成最終回復(fù)。 pass3.2 實現(xiàn)一個簡單的ReAct模式AgentReActReasoning Acting是一種經(jīng)典的Agent推理模式。我們實現(xiàn)一個簡化版本。# 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): 一個實現(xiàn)ReAct推理模式的簡單Agent。 async def think(self, user_input: str) - str: # 將用戶輸入加入記憶 self.memory.add_message(user, user_input) # 構(gòu)建系統(tǒng)提示包含工具描述 tools_description \n.join([f- {name}: {tool.description} for name, tool in self.tools.items()]) system_prompt f你是一個有幫助的AI助手可以調(diào)用工具來解決問題。 你可以使用的工具如下 {tools_description} 請遵循以下格式進行思考 Thought: 你需要思考當(dāng)前情況決定是否需要使用工具以及使用哪個工具。 Action: 需要調(diào)用的工具名稱如果沒有工具可用或不需要就填 None。 Action Input: 調(diào)用工具所需的輸入?yún)?shù)必須是JSON格式。如果Action是None這里也填 null。 Observation: 工具執(zhí)行后的結(jié)果。 ... (這個 Thought/Action/Action Input/Observation 循環(huán)可以重復(fù)多次) Thought: 我現(xiàn)在有足夠的信息來回答用戶了。 Final Answer: 給用戶的最終回答。 # 獲取對話上下文 context_messages self.memory.get_context() # 準(zhǔn)備發(fā)送給LLM的消息 messages [ {role: system, content: system_prompt}, *context_messages, {role: user, content: user_input}, ] max_iterations 5 for i in range(max_iterations): # 調(diào)用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的響應(yīng)提取 Thought, Action 等部分這里簡化實際應(yīng)用需要更魯棒的解析 # 假設(shè)LLM嚴(yán)格按照格式回復(fù) 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: # 執(zhí)行工具調(diào)用 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格式有誤直接返回其回復(fù) self.memory.add_message(assistant, response_text) return response_text return 抱歉經(jīng)過多輪推理仍未得到最終答案。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文件中的環(huán)境變量 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. 開發(fā)與集成自定義工具Tools工具是Agent能力的延伸。我們來創(chuàng)建幾個常用工具。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根據(jù)城市名稱查詢當(dāng)前天氣情況。, args_schemaWeatherInput ) self.api_key os.getenv(WEATHER_API_KEY) # 假設(shè)你有一個天氣API的Key # 這里使用一個模擬的免費API示例實際使用時請?zhí)鎿Q為真實API self.base_url http://wttr.in/ async def run(self, city: str) - str: 調(diào)用天氣API。 try: async with httpx.AsyncClient() as client: # 注意wttr.in 是一個免費服務(wù)格式可能變化僅作示例 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處理天氣信息時發(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一個有效的數(shù)學(xué)表達式例如(3 5) * 2) class CalculatorTool(Tool): def __init__(self): super().__init__( namecalculator, description計算一個數(shù)學(xué)表達式的結(jié)果。支持加減乘除和括號。, 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: 安全地計算數(shù)學(xué)表達式。 try: # 使用ast.literal_eval進行安全評估 # 注意這里我們實現(xiàn)一個更安全的自定義評估器避免直接使用eval result self._safe_eval(expression) return f表達式 {expression} 的計算結(jié)果是{result} except (SyntaxError, ValueError, TypeError, ZeroDivisionError) as e: return f計算表達式 {expression} 時出錯{str(e)}。請確保表達式格式正確。 def _safe_eval(self, node): 遞歸安全地評估AST節(jié)點。 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節(jié)點類型{type(node)}) def _safe_eval(self, expr: str): 入口函數(shù)將字符串表達式解析為AST并安全評估。 tree ast.parse(expr, modeeval) return self._safe_eval(tree.body) # 注意這里遞歸調(diào)用的是上面的方法需要重命名避免歧義。實際代碼中應(yīng)調(diào)整。修正上面的遞歸問題將內(nèi)部方法重命名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節(jié)點類型{type(node)}) async def run(self, expression: str) - str: try: tree ast.parse(expression, modeeval) result self._eval_node(tree.body) return f表達式 {expression} 的計算結(jié)果是{result} except (SyntaxError, ValueError, TypeError, ZeroDivisionError, AttributeError) as e: return f計算表達式 {expression} 時出錯{str(e)}。請確保表達式格式正確且僅包含基本算術(shù)運算。5. 組裝并運行你的第一個Agent現(xiàn)在讓我們將各個部分組裝起來創(chuàng)建一個可以對話的Agent。5.1 創(chuàng)建主運行腳本# 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. 創(chuàng)建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思考并回復(fù) response await agent.think(user_input) print(f\nAgent: {response}) except KeyboardInterrupt: print(\n\n程序被中斷。) break except Exception as e: print(f\n發(fā)生錯誤{e}) if __name__ __main__: asyncio.run(main())5.2 運行與測試在項目根目錄下運行python run_agent.py你應(yīng)該會看到提示符。嘗試輸入“北京今天天氣怎么樣”Agent會調(diào)用天氣工具“計算一下 (12 34) * 2 等于多少”Agent會調(diào)用計算器工具“你是誰”Agent會直接利用LLM知識回答觀察控制臺輸出的Thought、Action、Observation日志理解ReAct模式的運行過程。6. 工程化進階構(gòu)建工具鏈的其他關(guān)鍵環(huán)節(jié)一個基礎(chǔ)的Agent跑起來了但要將其工程化我們還需要完善以下環(huán)節(jié)。6.1 工具的動態(tài)加載與發(fā)現(xiàn)手動注冊工具在工具數(shù)量多時會很麻煩。我們可以實現(xiàn)一個工具發(fā)現(xiàn)機制。# 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): 自動發(fā)現(xiàn)指定包路徑下所有繼承了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: 根據(jù)工具名獲取工具實例。 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 # ... 其他導(dǎo)入 ... async def main(): # 自動發(fā)現(xiàn)并注冊 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的持久化當(dāng)前的Memory類只在內(nèi)存中保存對話。生產(chǎn)環(huán)境需要持久化到數(shù)據(jù)庫如Redis、SQLite或向量數(shù)據(jù)庫用于長上下文摘要。# 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服務(wù)層FastAPI要集成到現(xiàn)有系統(tǒng)需要提供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實例簡單示例生產(chǎn)環(huán)境需考慮并發(fā)和狀態(tài)隔離 _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 # 關(guān)閉時清理 print(Shutting down.) app FastAPI(lifespanlifespan) class ChatRequest(BaseModel): session_id: str None # 為空則創(chuàng)建新會話 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) # 這里簡化處理實際應(yīng)將Memory與session_id綁定 session_id request.session_id or str(uuid.uuid4()) # TODO: 根據(jù)session_id從數(shù)據(jù)庫加載或創(chuàng)建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. 常見問題與排查思路在搭建和運行過程中你可能會遇到以下問題問題現(xiàn)象可能原因排查步驟與解決方案導(dǎo)入錯誤ModuleNotFoundError1. 虛擬環(huán)境未激活。2. 項目未以可編輯模式安裝。3.PYTHONPATH未包含項目根目錄。1. 確認命令行前有(venv)。2. 在項目根目錄執(zhí)行pip install -e .。3. 在IDE中設(shè)置正確的項目根目錄和解釋器。OpenAI API 調(diào)用失敗1. API Key 未設(shè)置或錯誤。2. 網(wǎng)絡(luò)問題或代理配置。3. 余額不足或速率限制。1. 檢查.env文件中的OPENAI_API_KEY。2. 檢查網(wǎng)絡(luò)連接如需代理在代碼中配置http_client。3. 查看OpenAI控制臺賬單和用量。Agent 不調(diào)用工具直接回答1. 系統(tǒng)提示詞Prompt中工具描述不清晰。2. LLM 溫度temperature設(shè)置過高導(dǎo)致輸出不穩(wěn)定。3. 工具名稱或描述與用戶問題匹配度低。1. 優(yōu)化系統(tǒng)提示詞明確指令格式。2. 將temperature設(shè)為0確保確定性輸出。3. 檢查工具描述是否準(zhǔn)確嘗試用更直接的問題測試。工具調(diào)用參數(shù)解析錯誤1. LLM 生成的Action Input不是合法JSON。2. JSON中的參數(shù)名與工具定義的args_schema不匹配。1. 在Agent代碼中增加更健壯的JSON解析和錯誤處理。2. 在工具描述中明確參數(shù)名稱和類型?？梢允褂肞ydantic的schema_json()為LLM提供更精確的格式。多輪對話狀態(tài)丟失1.Memory類未正確集成到Agent中。2. 每次請求創(chuàng)建了新的Agent實例。1. 確保agent.think()方法中正確讀取和更新了self.memory。2. 對于Web服務(wù)需要將會話ID與Memory實例綁定并持久化存儲。性能問題響應(yīng)慢1. 工具調(diào)用是同步的阻塞了主線程。2. LLM API調(diào)用耗時過長。3. 未實現(xiàn)流式輸出。1. 確保所有工具方法都是async并使用await調(diào)用。2. 考慮設(shè)置合理的超時時間或使用更快的模型。3. 對于Web API可以研究SSEServer-Sent Events實現(xiàn)流式響應(yīng)。8. 最佳實踐與工程化建議將Agent投入生產(chǎn)環(huán)境需要遵循以下工程化準(zhǔn)則提示詞工程化將系統(tǒng)提示詞、用戶提示詞模板等抽取到配置文件或數(shù)據(jù)庫中便于管理和A/B測試。對提示詞進行版本控制。使用Jinja2等模板引擎動態(tài)生成提示詞。工具開發(fā)的標(biāo)準(zhǔn)化為所有工具編寫清晰的文檔包括輸入/輸出格式、錯誤碼。工具函數(shù)內(nèi)部必須有完善的錯誤處理和日志記錄。為工具編寫單元測試和集成測試。可觀測性與監(jiān)控在Agent的每個關(guān)鍵步驟接收輸入、調(diào)用LLM、調(diào)用工具、返回輸出記錄結(jié)構(gòu)化日志。記錄每次LLM調(diào)用的輸入Token、輸出Token數(shù)量及成本。使用像Prometheus和Grafana監(jiān)控工具調(diào)用成功率、延遲和Agent整體響應(yīng)時間。安全與權(quán)限工具權(quán)限控制不是所有用戶都能調(diào)用所有工具。實現(xiàn)一個權(quán)限層根據(jù)用戶身份或會話上下文決定可用的工具集。輸入輸出過濾對用戶輸入和工具返回的內(nèi)容進行安全檢查防止Prompt注入、敏感信息泄露。沙箱環(huán)境對于執(zhí)行代碼、訪問文件系統(tǒng)等高危工具必須在安全的沙箱環(huán)境中運行。測試策略單元測試測試每個工具函數(shù)的邏輯。集成測試測試Agent與LLM、工具的集成流程可以使用LLM的Mock來避免真實API調(diào)用。端到端測試模擬真實用戶場景測試完整的對話流。部署與運維容器化使用Docker將Agent及其依賴打包確保環(huán)境一致性。配置分離所有密鑰、端點URL等配置必須通過環(huán)境變量或配置中心管理絕不能硬編碼。健康檢查與就緒探針為Web服務(wù)添加/health端點便于K8s等編排系統(tǒng)管理。版本回滾Agent的代碼、模型版本、提示詞版本都應(yīng)有明確的版本號支持快速回滾。通過以上步驟你不僅搭建了一個可運行的智能體更構(gòu)建了一套支撐其持續(xù)迭代和穩(wěn)定運行的工程化工具鏈雛形。這套工具鏈的核心思想是模塊化、可觀測、可測試、可部署。你可以在此基礎(chǔ)上繼續(xù)擴展工作流引擎、可視化編排界面、更復(fù)雜的記憶模塊如向量數(shù)據(jù)庫逐步將其打造成團隊內(nèi)部強大的AI能力中臺。