1. 从零到一构建一个能“思考”和“行动”的AI智能体如果你已经玩过ChatGPT体验过RAG检索增强生成系统甚至用LangChain搭过一些简单的应用那你可能会觉得这些工具虽然强大但总感觉缺了点什么。它们更像是“一问一答”的智能文档库或者一个功能强大的脚本。真正的“智能”应该是什么是能够像人一样面对一个复杂目标主动去规划、去调用工具、去执行、去反思最终完成任务。这就是AI智能体AI Agent的魅力所在。简单来说智能体是一个具备自主性的AI系统。它不再是被动地等待你的完整指令而是被赋予一个目标比如“帮我分析一下上个月的销售数据并写一份报告”然后自己决定需要哪些步骤调用哪些工具如数据库查询、Python分析、文档生成并在执行过程中根据结果调整策略。这听起来很科幻但得益于大型语言模型LLM强大的规划与推理能力以及像LangChain、LangGraph、CrewAI这样的框架我们现在完全可以在自己的电脑上构建这样的智能体。我最近花了大量时间深入实践了curiousily的“AI-Bootcamp”中关于智能体的部分特别是基于LangGraph和本地模型如Ollama的构建。我发现从简单的“聊天机器人”升级到“智能体”不仅仅是技术栈的叠加更是一种思维模式的转变。你需要从“如何回答一个问题”转变为“如何设计一个能解决问题的自主系统”。接下来我将结合我的实操经验为你拆解构建一个实用AI智能体的完整路径、核心组件以及那些教程里不会写的“坑”。2. 智能体系统的核心架构与设计哲学在动手写代码之前我们必须理解智能体系统的核心架构。一个典型的智能体无论是简单的单智能体还是复杂的多智能体协作系统都离不开以下几个核心组件它们共同构成了智能体的“大脑”和“身体”。2.1 大脑规划与决策引擎LLM 框架大型语言模型是智能体的“大脑”负责理解目标、规划步骤、做出决策。但原始的LLM就像一个天马行空的思想家我们需要用框架来约束和引导它使其行为可控、可预测。LangChain LangGraph这是目前生态最成熟的选择。LangChain提供了构建链Chains的基础组件而LangGraph是其用于构建有状态、多步骤工作流的超集。它的核心概念是“图”Graph你可以将智能体的每个步骤如“思考”、“执行工具”、“评估”定义为一个节点Node用边Edge来定义流程逻辑。LangGraph内置了循环、分支等控制流非常适合构建需要反复“思考-行动-观察”的智能体。我的体会是一旦理解了“状态”State在节点间的传递构建复杂工作流就会变得非常直观。CrewAI这是一个更高层次的抽象框架专注于多智能体协作。它帮你处理了智能体角色定义、任务委派、协作流程等繁琐工作。如果你要构建的是一个“团队”比如一个分析师智能体、一个写手智能体、一个审查员智能体CrewAI能极大提升开发效率。但它的灵活性相对LangGraph稍低更适合标准化协作场景。AutoGen由微软推出同样专注于多智能体对话与协作其对话模式非常强大适合需要多个智能体通过“聊天”来协商解决问题的场景。选择建议对于初学者或需要高度定制化、复杂控制流的项目我强烈推荐从LangGraph开始。它提供了从简单到复杂最平滑的学习曲线。CrewAI则适合当你明确需要“经理-员工”式团队协作时快速上手。2.2 身体工具与执行层智能体不能只“想”还要能“做”。工具Tools就是智能体的手和脚。一个工具本质上是一个函数LLM可以决定在何时调用它。内置工具LangChain等框架提供了海量社区工具如网络搜索SerpAPI、维基百科查询、Python REPL执行代码、文件操作等。自定义工具这是智能体真正发挥价值的地方。你可以将任何API、数据库查询函数、内部业务系统接口封装成工具。例如query_database(sql): 执行SQL查询。send_email(to, subject, body): 发送邮件。analyze_sales_data(date_range): 调用内部数据分析服务。get_current_stock_price(symbol): 调用金融API。关键设计原则工具的描述description至关重要。LLM完全依赖你提供的自然语言描述来决定是否以及如何使用这个工具。描述必须清晰、准确说明工具的用途、输入参数和预期输出。例如“query_database: 执行一条SQL SELECT查询语句并返回结果。输入应为有效的SQL字符串。”这比简单的“查询数据库”要好得多。2.3 记忆与状态管理智能体需要在多轮交互中记住上下文这就是状态State。状态通常是一个字典dict随着工作流的执行在不同节点间流转和更新。短期记忆对话记忆存储当前会话的对话历史。通常使用ConversationBufferMemory或ConversationSummaryMemory后者可以压缩长历史。长期记忆向量数据库用于存储和检索智能体过去的经验或知识使其能从历史中学习。这通常通过RAG技术实现将历史决策和结果存入向量库供后续任务参考。工作流状态在LangGraph中这是核心。状态对象包含了当前任务的所有信息如用户目标、已执行步骤列表、工具执行结果、LLM的思考过程等。你需要精心设计状态的结构确保每个节点都能读取和写入所需的信息。2.4 评估与安全护栏让智能体自主运行是有风险的。它可能陷入无限循环、调用错误工具、或产生有害输出。因此必须建立评估与安全机制。超时与最大步骤限制强制设定智能体运行的最大步数或最长时间防止死循环。工具使用验证在关键工具如删除数据、发送邮件被调用前可以加入一个人工确认节点或设置更严格的权限。输出内容过滤对智能体的最终输出进行内容安全检查。评估节点在工作流中插入评估节点检查上一步的结果是否合理、是否已达成目标从而决定是继续、重试还是终止。理解了这些组件我们就有了设计蓝图。接下来我们进入实战环节看看如何用LangGraph和本地模型将这些部件组装起来。3. 实战用LangGraph和Ollama构建本地数据库查询智能体我们来实现一个经典且实用的智能体一个能理解自然语言问题、自动生成并执行SQL查询、然后对结果进行解释的“数据库分析师”智能体。全程使用本地模型通过Ollama运行保证数据隐私。3.1 环境准备与工具定义首先安装核心库并启动Ollama服务。我们选择llama3.2或qwen2.5这类较强的开源模型。# 安装依赖 pip install langgraph langchain langchain-community ollama chromadb sqlite3 # 启动Ollama并拉取模型 (假设已安装Ollama) ollama pull llama3.2:3b # 选择一个适合你硬件尺寸的版本接着定义两个核心工具一个用于查询SQLite数据库一个用于在查询失败时获取数据库表结构以供参考。import sqlite3 from langchain.tools import tool from typing import Optional # 假设我们有一个简单的销售数据库 ‘sales.db‘包含一个 ‘orders‘ 表 # 表结构: id (INT), product (TEXT), amount (REAL), date (TEXT), region (TEXT) tool def query_database(sql_query: str) - str: 执行一条SQL SELECT查询语句并返回结果。如果查询失败返回错误信息。请确保SQL语法正确。 try: conn sqlite3.connect(sales.db) cursor conn.cursor() cursor.execute(sql_query) results cursor.fetchall() conn.close() # 将结果格式化为易读的字符串 if results: columns [description[0] for description in cursor.description] return f查询成功。结果\n列名{columns}\n数据{results} else: return 查询成功但未返回任何数据。 except Exception as e: return f查询失败错误{str(e)} tool def get_table_schema(table_name: Optional[str] None) - str: 获取数据库的表结构信息。如果不指定表名则列出所有表。 conn sqlite3.connect(sales.db) cursor conn.cursor() if table_name: cursor.execute(fPRAGMA table_info({table_name})) schema cursor.fetchall() conn.close() if schema: return f表 ‘{table_name}‘ 的结构{schema} else: return f未找到名为 ‘{table_name}‘ 的表。 else: cursor.execute(SELECT name FROM sqlite_master WHERE typetable;) tables cursor.fetchall() conn.close() return f数据库中的所有表{tables}注意在真实生产环境中query_database工具必须进行严格的SQL注入检查和权限控制例如只允许SELECT操作或使用参数化查询。这里为演示简化了安全措施。3.2 构建LangGraph工作流这是最核心的部分。我们将设计一个包含“思考”、“行动”、“评估”三个主要节点的有状态图。from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_community.chat_models import ChatOllama from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langgraph.prebuilt import ToolExecutor, ToolInvocation # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[List, operator.add] # 消息历史 user_query: str # 用户的原始问题 sql_generated: str # 生成的SQL query_result: str # 查询结果 steps: Annotated[List[str], operator.add] # 记录已执行的步骤用于追踪 # 2. 初始化模型和工具执行器 llm ChatOllama(modelllama3.2:3b, temperature0) tools [query_database, get_table_schema] tool_executor ToolExecutor(tools) # 3. 绑定工具到LLM llm_with_tools llm.bind_tools(tools) # 4. 定义各个节点函数 def think_node(state: AgentState) - AgentState: 思考节点分析用户问题决定下一步行动。 system_prompt 你是一个专业的数据库分析师。你的目标是理解用户关于销售数据的问题并生成正确的SQL查询来获取答案。 你可以使用以下工具 1. get_table_schema: 如果你不确定数据库结构先用这个工具查看表信息。 2. query_database: 当你确定了SQL语句后用这个工具执行查询。 请一步步思考。如果你的思考中决定要使用工具请直接输出对应的工具调用ToolCall。 # 构建对话历史 prompt [{role: system, content: system_prompt}] state[messages] # 调用LLM response llm_with_tools.invoke(prompt) # 更新消息历史 new_messages [response] state[messages].extend(new_messages) state[steps].append(f思考节点分析了问题‘{state[‘user_query‘]}‘决定下一步行动。) # 如果LLM决定调用工具state中会包含ToolCall信息供路由函数判断 return state def action_node(state: AgentState) - AgentState: 行动节点执行LLM选择的工具。 # 从上一条消息即think_node的响应中获取工具调用信息 last_message state[messages][-1] tool_calls last_message.tool_calls if not tool_calls: # 如果没有工具调用可能是LLM直接回答了则跳转到结束 state[steps].append(行动节点未发现工具调用直接生成回答。) return state # 执行每一个工具调用 tool_messages [] for tool_call in tool_calls: # 执行工具 result tool_executor.invoke(tool_call) # 记录生成的SQL或结果 if tool_call[‘name‘] ‘query_database‘: state[sql_generated] tool_call[‘args‘][‘sql_query‘] state[query_result] result # 创建工具响应消息 tool_message ToolMessage(contentstr(result), tool_call_idtool_call[‘id‘]) tool_messages.append(tool_message) # 更新消息历史 state[messages].extend(tool_messages) state[steps].append(f行动节点执行了工具 {[tc[‘name‘] for tc in tool_calls]}。) return state def evaluate_node(state: AgentState) - AgentState: 评估节点检查工具执行结果判断是否完成任务或需要继续。 last_message state[messages][-1] # 检查查询结果是否包含错误 if query_result in state and 失败 in state[query_result]: # 查询失败需要重新思考可能要先获取表结构 state[steps].append(评估节点查询失败需要重新规划。) # 可以在这里添加逻辑比如如果失败次数过多则终止 return think_again # 返回边名称指向重新思考 # 检查是否已经得到了查询结果 if query_result in state and 查询成功 in state[query_result]: # 成功获取数据现在需要解释结果 state[steps].append(评估节点查询成功准备生成最终解释。) return explain # 返回边名称指向解释节点 # 其他情况比如刚获取了表结构继续思考生成SQL state[steps].append(评估节点获取了元信息继续生成SQL。) return continue_think def explain_node(state: AgentState) - AgentState: 解释节点基于查询结果生成面向用户的自然语言回答。 final_prompt f 用户最初的问题是{state[‘user_query‘]} 你生成的SQL查询是{state.get(‘sql_generated‘, ‘N/A‘)} 数据库返回的结果是{state.get(‘query_result‘, ‘N/A‘)} 请根据以上信息用简洁明了的语言直接回答用户的问题。不要提及SQL或技术细节只需给出基于数据的结论。 例如如果用户问‘上个月总销售额是多少‘而结果是‘10000‘你就回答‘上个月的总销售额是10000元。‘ response llm.invoke([HumanMessage(contentfinal_prompt)]) # 将最终回答添加到消息历史 state[messages].append(AIMessage(contentresponse.content)) state[steps].append(解释节点生成了最终的用户回答。) return state # 5. 构建图并设置路由逻辑 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(think, think_node) workflow.add_node(act, action_node) workflow.add_node(evaluate, evaluate_node) workflow.add_node(explain, explain_node) # 设置入口点 workflow.set_entry_point(think) # 定义边路由 workflow.add_edge(think, act) # 思考完总是去执行 workflow.add_edge(act, evaluate) # 执行完总是去评估 # 评估节点的条件边 workflow.add_conditional_edges( evaluate, # 路由函数就是 evaluate_node 的返回值 lambda x: x, # 这里x就是evaluate_node返回的边名称字符串 { think_again: think, continue_think: think, explain: explain, } ) workflow.add_edge(explain, END) # 解释完成后结束 # 编译图 app workflow.compile()3.3 运行与测试智能体现在我们可以运行这个智能体来处理用户查询了。# 初始化状态 initial_state { messages: [HumanMessage(content帮我找出今年第一季度销售额最高的产品是什么)], user_query: 帮我找出今年第一季度销售额最高的产品是什么, sql_generated: , query_result: , steps: [] } # 运行智能体 final_state app.invoke(initial_state) # 查看最终结果 print( 最终回答 ) print(final_state[messages][-1].content) print(\n 执行步骤追踪 ) for i, step in enumerate(final_state[steps]): print(f{i1}. {step}) print(\n 生成的SQL ) print(final_state.get(sql_generated))一个成功的运行流程可能是Think: LLM看到问题意识到需要查询数据库但不确定表结构决定调用get_table_schema。Act: 执行get_table_schema返回表结构信息。Evaluate: 评估节点收到表结构判断需要“继续思考”(continue_think)。Think: LLM根据表结构生成SQLSELECT product, SUM(amount) FROM orders WHERE date BETWEEN ‘2024-01-01‘ AND ‘2024-03-31‘ GROUP BY product ORDER BY SUM(amount) DESC LIMIT 1。Act: 执行query_database返回结果。Evaluate: 查询成功判断需要“解释”(explain)。Explain: LLM根据结果生成最终答案“今年第一季度销售额最高的产品是‘智能手机’总销售额为125000元。”END: 工作流结束。这个智能体展现出了自主性它自己决定先去查看表结构再生成查询最后解释结果。这就是智能体与简单链的本质区别。4. 避坑指南与进阶优化策略构建和运行智能体的过程绝非一帆风顺。以下是我在实践中总结的常见问题与解决方案很多是你在官方文档里找不到的“血泪经验”。4.1 常见问题与排查清单问题现象可能原因排查与解决思路智能体陷入死循环评估逻辑有缺陷或LLM在“思考”和“获取表结构”间来回跳转。1.添加步骤计数器在状态中增加step_count在evaluate_node中检查超过阈值则强制终止并报错。2.细化评估逻辑区分“首次需要表结构”和“反复失败”。可以记录已获取过的表名避免重复查询。3.使用更智能的模型较小的本地模型如3B参数逻辑能力有限升级到7B或更高参数模型能显著改善。LLM不调用工具直接胡编答案1. 工具描述不清晰。2. 系统提示词未强调必须使用工具。3. 模型能力不足。1.优化工具描述确保描述明确说明工具的用途、输入格式和输出示例。2.强化系统提示词使用类似“你必须使用提供的工具来获取信息严禁猜测或编造数据”的强硬指令。3.调整温度temperature设置为0或更低减少随机性。4.在思考节点使用bind_tools确保LLM的输出格式支持工具调用。生成的SQL语法错误1. LLM对特定数据库方言不熟。2. 问题描述模糊。1.在提示词中明确数据库类型如“你正在为SQLite数据库生成查询”。2.提供少量示例Few-Shot在系统提示词中给1-2个从自然语言到正确SQL的示例。3.增加验证层在action_node执行SQL前可以先用一个简单的语法检查工具如sqlparse预验证如果失败则返回一个要求LLM修正的指令。处理复杂、多步骤问题能力差智能体设计为单次“思考-行动”循环无法进行深层规划。1.引入规划节点在初始思考前增加一个“规划”节点让LLM先拆解任务为子步骤列表例如[“1. 确定时间范围”, “2. 查询相关表”, “3. 聚合计算”, “4. 排序”]并将此计划存入状态。后续节点参照计划执行。2.采用分层图结构将复杂任务拆分成子图Subgraph每个子图负责一个子目标主图负责协调。这是构建复杂智能体的高级模式。工具执行结果太长超出LLM上下文查询返回大量数据导致后续LLM调用时上下文窗口不足。1.结果摘要在action_node中如果结果很大先用一个简单的函数或另一个LLM调用对结果进行摘要再将摘要放入状态。2.选择性保留只将最关键的数据字段放入状态消息中。3.使用具有长上下文窗口的模型。4.2 性能与成本优化本地模型选择llama3.2:7b在逻辑和代码能力上是性价比很高的起点。如果硬件允许qwen2.5:14b或command-r表现更佳。务必在Ollama中启用GPU加速如果可用。缓存对频繁且结果不变的工具调用如get_table_schema实施缓存可以极大减少不必要的LLM调用和等待时间。LangChain支持多种缓存后端。异步执行如果智能体需要调用多个独立的工具可以考虑在action_node中使用异步并行执行缩短整体耗时。流式输出对于最终的解释回答可以配置流式输出提升用户体验。4.3 从单智能体到多智能体协作当你需要处理涉及多个专业领域的复杂任务时例如市场分析报告需要数据提取、趋势分析、文案撰写、视觉设计单智能体就力不从心了。这时需要引入多智能体系统。使用CrewAI可以快速搭建这样一个团队from crewai import Agent, Task, Crew, Process from langchain_community.chat_models import ChatOllama llm ChatOllama(modelllama3.2:7b) # 定义智能体角色 data_analyst Agent( role‘资深数据分析师‘, goal‘从给定的数据源中准确提取、清洗和分析数据并提供核心洞察‘, backstory‘你拥有多年的数据分析经验擅长从杂乱数据中发现规律。‘, tools[query_database, get_table_schema], # 可以赋予不同的工具集 llmllm, verboseTrue ) report_writer Agent( role‘商业报告撰写专家‘, goal‘将数据分析结果转化为结构清晰、观点明确、语言精炼的商业报告‘, backstory‘你是一名前财经记者擅长将复杂数据转化为易懂的故事。‘, llmllm, verboseTrue ) # 定义任务链 task1 Task( description“分析数据库‘sales.db‘中过去一年的产品销售趋势找出增长最快的产品和下滑最严重的区域。”, agentdata_analyst, expected_output“一份包含关键数据表格和初步洞察点的分析摘要。” ) task2 Task( description“基于数据分析师提供的摘要撰写一份给管理层的半页纸商业简报突出核心发现和建议。”, agentreport_writer, context[task1], # 任务2依赖于任务1的输出 expected_output“一份格式规范、语言专业的商业简报Markdown格式。” ) # 组建团队并运行 crew Crew( agents[data_analyst, report_writer], tasks[task1, task2], processProcess.sequential, # 顺序执行 verbose2 ) result crew.kickoff() print(result)在这个配置下data_analyst智能体会自主完成数据查询和分析然后将结果传递给report_writer智能体去撰写报告。CrewAI自动管理它们之间的协作和上下文传递。构建一个稳定、可靠的AI智能体是一个迭代的过程。从最简单的单循环智能体开始逐步增加工具、完善状态管理、加入评估和护栏最终扩展到多智能体协作。核心在于理解“状态流”和“决策逻辑”。本地模型Ollama和开源框架LangGraph的组合为我们提供了低成本、高隐私、可深度定制的实践平台。