,Python智能体开发从入门到精通,收藏这一篇就够了!)
用一篇文章带你走完AI Agent开发的全流程从大模型调用到业务闭环代码能跑原理能懂。你有没有想过让AI不只是聊天而是真正帮你干活比如自动查询数据库、调用API发邮件、读取文件处理数据……这些事ChatGPT做不到但AI Agent可以。说白了ChatGPT像个会聊天的秘书而Agent是个能跑腿的助理。一个只会说另一个能行动。今天我们就来搭一个能干活的Agent。不玩虚的直接上代码。01 一张图看懂AI Agent是什么先搞清楚概念后面才不会晕。┌─────────────────────────────────────────────────────────────┐│ 传统 ChatGPT ││ 用户输入 → LLM处理 → 返回文字 ││ 只会说话 │└─────────────────────────────────────────────────────────────┘┌─────────────────────────────────────────────────────────────┐│ AI Agent ││ 用户输入 → LLM决策 → 调用工具 → 执行任务 → 返回结果 ││ ↓ ↓ ││ 理解意图 FunctionCalling / MCP Server ││ 能干活 │└─────────────────────────────────────────────────────────────┘看到区别了吗Agent多了一层工具调用能力。这个能力主要通过两个技术实现Function Calling让LLM能调用你定义的函数MCP Server把工具封装成标准服务Agent随时调用02 第一步搞定大模型API调用先从基础开始。我们用Python调用Claude API你也可以用GPT逻辑类似。2.1 安装依赖pip install anthropic2.2 最简单的调用# simple_chat.pyimport anthropicclient anthropic.Anthropic(api_key你的API密钥)message client.messages.create(modelclaude-3-5-sonnet-20241022,max_tokens1024, messages[{role: user,content: 你好你是谁 }])print(message.content[0].text)跑一下能正常对话就OK了。2.3 成本控制小技巧新手上路最容易忽略的就是成本。算笔账模型输入价格输出价格适用场景GPT-4o~$2.5/百万token~$10/百万token复杂推理Claude 3.5 Sonnet~$3/百万token~$15/百万token长文本GPT-4o-mini~$0.15/百万token~$0.6/百万token简单任务实用建议开发用便宜模型mini/haiku生产才上旗舰模型设置max_tokens限制输出长度对话历史太长就做摘要压缩03 核心能力Function Calling 让AI能干活Function Calling是Agent的武器库。你告诉LLM有哪些函数它自己决定什么时候调。3.1 基本原理用户: 帮我查一下明天北京的天气┌─────────────────────────────────────────────────────────────┐│ LLM判断需要调用天气查询函数 ││ ││ 返回: { ││ function_calls: [{ ││ name: get_weather, ││ arguments: {city: 北京, date: 明天} ││ }] ││ } │└─────────────────────────────────────────────────────────────┘ ↓┌─────────────────────────────────────────────────────────────┐│ 你的代码执行函数拿到真实天气数据 │└─────────────────────────────────────────────────────────────┘ ↓┌─────────────────────────────────────────────────────────────┐│ 把结果塞回给LLM让它生成回答 ││ ││ LLM输出: 明天北京晴气温15-25度 │└─────────────────────────────────────────────────────────────┘3.2 完整代码示例我们写一个能查天气、能计算、能搜索的Agent# agent_with_functions.pyimport anthropicimport jsonimport requestsfrom datetime import datetime, timedeltaclient anthropic.Anthropic(api_key你的API密钥)# 工具函数定义 defget_weather(city: str, date: str None) - dict:查询天气模拟实现真实项目可对接天气API# 这里用假数据演示 weather_data {北京: {temp: 15-25°C, condition: 晴},上海: {temp: 18-28°C, condition: 多云},深圳: {temp: 22-30°C, condition: 阵雨} } result weather_data.get(city, {temp: 未知, condition: 未知})return {city: city,date: date or今天,temperature: result[temp],condition: result[condition] }defcalculate(expression: str) - float:计算数学表达式try: result eval(expression) # 生产环境注意安全return {expression: expression, result: result}except Exception as e:return {error: str(e)}defsearch_web(query: str, num_results: int 3) - list:网络搜索模拟# 真实项目可以对接Google Search API或Serper mock_results [ {title: f{query}的搜索结果1, url: https://example.com/1}, {title: f{query}的搜索结果2, url: https://example.com/2}, ]return mock_results[:num_results]# Function Calling 主逻辑 defrun_agent(user_message: str):# 定义可用的函数 tools [ {name: get_weather,description: 查询指定城市的天气情况,input_schema: {type: object,properties: {city: {type: string,description: 城市名称如北京、上海 },date: {type: string,description: 日期如今天、明天、2024-03-15 } },required: [city] } }, {name: calculate,description: 计算数学表达式,input_schema: {type: object,properties: {expression: {type: string,description: 要计算的数学表达式如 23*4 } },required: [expression] } }, {name: search_web,description: 在网络上搜索信息,input_schema: {type: object,properties: {query: {type: string, description: 搜索关键词},num_results: {type: integer, description: 返回结果数量} },required: [query] } } ]# 函数映射表 function_map {get_weather: get_weather,calculate: calculate,search_web: search_web }# 第一次调用LLM response client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens4096, messages[{role: user, content: user_message}], toolstools )# 处理响应 assistant_message []for block in response.content:if block.type text: assistant_message.append({type: text, text: block.text})elif block.type tool_use:# 执行函数调用 tool_name block.name tool_args block.inputprint(f\n Agent调用函数: {tool_name})print(f 参数: {tool_args})# 执行实际函数 result function_map[tool_name](**tool_args)print(f 结果: {result})# 记录工具调用和结果 assistant_message.append({type: tool_use,id: block.id,name: tool_name,input: tool_args,content: json.dumps(result, ensure_asciiFalse) })# 如果有函数调用需要把结果塞回给LLMifany(block.get(type) tool_usefor block in assistant_message):# 构建完整的消息历史 messages [ {role: user, content: user_message}, {role: assistant, content: assistant_message} ]# 添加tool_result消息 tool_results []for block in assistant_message:if block[type] tool_use: tool_results.append({type: tool_result,tool_use_id: block[id],content: block[content] })# 最后一次调用让LLM生成最终回答 final_response client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens4096, messagesmessages [ {role: user, content: tool_results} ] )return final_response.content[0].text# 没有函数调用直接返回return response.content[0].text# 测试 if __name__ __main__:print( AI Agent Function Calling 演示 \n)# 测试1: 天气查询print(【测试1】用户: 北京明天天气怎么样) result run_agent(北京明天天气怎么样)print(fAgent: {result}\n)# 测试2: 数学计算print(【测试2】用户: 帮我算一下 (100 50) * 2 等于多少) result run_agent(帮我算一下 (100 50) * 2 等于多少)print(fAgent: {result}\n)# 测试3: 复杂任务print(【测试3】用户: 查一下上海天气然后算一下如果气温是20度华氏度是多少) result run_agent(查一下上海天气然后算一下如果气温是20度华氏度是多少)print(fAgent: {result}\n)跑起来你会看到类似这样的输出 AI Agent Function Calling 演示 【测试1】用户: 北京明天天气怎么样 Agent调用函数: get_weather 参数: {city: 北京, date: 明天} 结果: {city: 北京, date: 明天, temperature: 15-25°C, condition: 晴}Agent: 根据查询结果明天北京的天气是晴天气温在15到25度之间。天气不错适合出门【测试2】用户: 帮我算一下 (100 50) * 2 等于多少 Agent调用函数: calculate 参数: {expression: (100 50) * 2} 结果: {expression: (100 50) * 2, result: 300.0}Agent: (100 50) * 2 3003.3 常见坑点问题现象解决方案参数类型错误LLM传了字符串但函数要数字在input_schema严格定义类型函数调用死循环LLM反复调用同一个函数设置最大调用次数限制JSON解析失败LLM返回的不是合法JSON用try-catch包裹返回错误提示工具描述不清LLM不知道什么时候用哪个函数优化description给使用场景示例04 进阶MCP Server 让工具标准化Function Calling有个问题工具函数都写死在代码里。每次加新功能都要改代码、重启服务。MCP (Model Context Protocol)就是来解决这个问题的。它让工具变成可插拔的服务。4.1 MCP是什么传统方式:Agent代码 → 直接调用 get_weather() 函数 想加新功能改代码、重启MCP方式:Agent代码 → 通过MCP协议 → MCP Server → 调用工具 ↓ 可以热插拔、独立部署MCP Server可以提供三种资源Resources静态或动态数据如数据库查询结果Tools可执行的操作如发送邮件Prompts预定义的提示词模板4.2 写一个简单的MCP Server我们用TypeScript写一个待办事项的MCP Server// todo-mcp-server/src/index.tsimport { Server } from modelcontextprotocol/sdk/server/index.js;import { StdioServerTransport } from modelcontextprotocol/sdk/server/stdio.js;import { CallToolRequestSchema, ListToolsRequestSchema,} from modelcontextprotocol/sdk/types.js;// 内存存储待办事项生产环境用真实数据库lettodos: Array{ id: number; text: string; done: boolean } [ { id:1, text:学习AI Agent开发, done:false }, { id:2, text:完成MCP Server, done:true },];const server new Server( {name:todo-server,version:1.0.0, }, {capabilities: {tools: {}, }, });// 注册工具列表server.setRequestHandler(ListToolsRequestSchema, async () { return {tools: [ {name:add_todo,description:添加一个新的待办事项,inputSchema: {type:object,properties: {text: {type:string,description:待办事项的内容, }, },required: [text], }, }, {name:list_todos,description:列出所有待办事项,inputSchema: {type:object,properties: {}, }, }, {name:complete_todo,description:标记一个待办事项为完成,inputSchema: {type:object,properties: {id: {type:number,description:待办事项的ID, }, },required: [id], }, }, ], };});// 处理工具调用server.setRequestHandler(CallToolRequestSchema, async (request) { const { name, arguments: args } request.params; switch (name) { case add_todo: const newTodo {id: todos.length 1,text: args.text,done:false, }; todos.push(newTodo); return {content: [ {type:text,text: ✅ 已添加待办事项: ${args.text}, }, ], }; case list_todos: const todoList todos .map((t) ${t.done ?✓ : ○} ${t.id}. ${t.text}) .join(\n); return {content: [ {type:text,text: 待办事项列表:\n${todoList ||空}, }, ], }; case complete_todo: const todo todos.find((t) t.id args.id);if (todo) {todo.donetrue; return {content: [ {type:text,text: ✓ 已标记完成: ${todo.text}, }, ], }; } return {content: [ {type:text,text: ❌ 找不到ID为${args.id}的待办事项, }, ], };default:throw new Error(未知工具: ${name}); }});// 启动服务async function main() { const transport new StdioServerTransport(); await server.connect(transport); console.error(Todo MCP Server running...);}main().catch(console.error);配套的package.json:{name: todo-mcp-server,version: 1.0.0,type: module,dependencies: {modelcontextprotocol/sdk: ^1.0.4 },scripts: {start: tsx src/index.ts }}4.3 在Agent中集成MCPClaude Code原生支持MCP你只需要在配置中声明//.claude/settings.json{mcpServers: {todo-server: {command: node,args: [/path/to/todo-mcp-server/dist/index.js] } }}然后在Python代码中可以通过MCP客户端调用# agent_with_mcp.pyfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientasyncdefconnect_to_mcp_server():连接到MCP Server server_params StdioServerParameters( commandnode, args[/path/to/todo-mcp-server/dist/index.js] )asyncwith stdio_client(server_params) as (read, write):asyncwith ClientSession(read, write) as session:# 初始化连接await session.initialize()# 列出可用工具 tools await session.list_tools()print(f可用工具: {[t.name for t in tools.tools]})# 调用工具 result await session.call_tool(add_todo, {text: 学习MCP})print(result) result await session.call_tool(list_todos, {})print(result)# 运行import asyncioasyncio.run(connect_to_mcp_server())05 完整闭环一个能用的项目现在把所有技术串起来。我们做一个智能待办助手用户说: 帮我记录下周一要交周报然后列出本周待办 ↓Agent理解: 两个任务 → 添加待办 列表查询 ↓通过MCP调用todo-server ↓返回自然语言回复完整架构# smart_todo_agent.pyimport anthropicimport asynciofrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom datetime import datetime, timedeltaclassSmartTodoAgent:def__init__(self, api_key: str):self.client anthropic.Anthropic(api_keyapi_key)self.mcp_session Noneasyncdefconnect_mcp(self):连接MCP Server server_params StdioServerParameters( commandnode, args[./todo-mcp-server/dist/index.js] )self.stdio_context stdio_client(server_params)self.read, self.write awaitself.stdio_context.__aenter__()self.session ClientSession(self.read, self.write)awaitself.session.initialize()awaitself.session.__aenter__()asyncdefclose_mcp(self):关闭MCP连接ifself.session:awaitself.session.__aexit__(None, None, None)ifself.stdio_context:awaitself.stdio_context.__aexit__(None, None, None)asyncdefget_mcp_tools(self):获取MCP工具列表 tools awaitself.session.list_tools()return [ {name: t.name,description: t.description,input_schema: t.inputSchema, }for t in tools.tools ]asyncdefcall_mcp_tool(self, tool_name: str, arguments: dict):调用MCP工具 result awaitself.session.call_tool(tool_name, arguments)return result.content[0].textasyncdefchat(self, user_message: str) - str:主聊天逻辑# 获取MCP工具 mcp_tools awaitself.get_mcp_tools()# 调用LLM response self.client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens4096, messages[{role: user, content: user_message}], toolsmcp_tools )# 处理响应 assistant_message []for block in response.content:if block.type text: assistant_message.append({type: text, text: block.text})elif block.type tool_use:# 调用MCP工具 result awaitself.call_mcp_tool(block.name, block.input)print(f 调用MCP工具: {block.name})print(f 参数: {block.input})print(f 结果: {result}) assistant_message.append({type: tool_use,id: block.id,name: block.name,input: block.input,content: result })# 如果有工具调用让LLM生成最终回复ifany(b.get(type) tool_usefor b in assistant_message): tool_results [ {type: tool_result,tool_use_id: b[id],content: b[content] }for b in assistant_messageif b[type] tool_use ] final_response self.client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens4096, messages[ {role: user, content: user_message}, {role: assistant, content: assistant_message} ] [ {role: user, content: tool_results} ] )return final_response.content[0].textreturn response.content[0].text# 使用示例asyncdefmain(): agent SmartTodoAgent(api_key你的API密钥)await agent.connect_mcp()try:# 测试对话 questions [帮我添加一个待办下周一交周报,列出所有待办事项,把第一个待办标记为完成 ]for q in questions:print(f\n 用户: {q}) response await agent.chat(q)print(f Agent: {response})finally:await agent.close_mcp()if __name__ __main__: asyncio.run(main())06 生产实战你需要知道的事代码能跑只是第一步。真要上线还有不少坑。6.1 错误处理要到位# 不要这样result some_tool_call()# 要这样from tenacity import retry, stop_after_attempt, wait_exponentialretry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10))asyncdefsafe_tool_call(tool_name, args):try: result await call_tool(tool_name, args)return resultexcept Exception as e:print(f工具调用失败: {tool_name}, 错误: {e})# 返回一个错误信息给LLM让它尝试其他方案returnf错误: {str(e)}6.2 成本控制策略说明节省比例对话历史压缩超过10轮就做摘要~30%分级模型选择简单任务用mini~50%Token限制设置max_tokens上限~20%缓存常见问答相同问题走缓存~40%6.3 安全注意事项# ⚠️ 危险直接用eval执行用户输入defcalculate(expression: str):returneval(expression) # 用户可能输入 __import__(os).system(rm -rf /)# ✅ 安全限制执行环境defcalculate(expression: str):import astimport operator# 只允许安全的运算 allowed_operators { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, }# 解析AST检查安全性...其他安全要点API密钥用环境变量别硬编码用户数据要脱敏敏感操作要二次确认6.4 监控和日志import logginglogging.basicConfig( levellogging.INFO,format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(agent.log), logging.StreamHandler() ])logger logging.getLogger(agent)deflog_tool_call(tool_name, args, result, duration_ms): logger.info(fTool: {tool_name}, Args: {args}, Duration: {duration_ms}ms)# 可以发送到监控系统如Prometheus、DataDog07 总结与下一步我们今天走完了完整的Agent开发链路1️⃣ LLM API调用 → 基础对话能力2️⃣ FunctionCalling → 让AI能调用工具3️⃣ MCP Server → 工具标准化、可插拔4️⃣ 完整闭环 → 从用户输入到任务执行核心要点Agent LLM 工具调用能力Function Calling是让LLM调用函数的协议MCP让工具变成可插拔的服务生产环境要处理好错误、成本、安全下一步可以尝试用LangChain/LangGraph简化开发添加向量数据库做长期记忆用流式输出提升体验部署成API服务供前端调用学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】