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