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(f"PRAGMA 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 type='table';") 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(model="llama3.2:3b", temperature=0) 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(content=str(result), tool_call_id=tool_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‘，你就回答‘上个月的总销售额是10，000元。‘ """ response = llm.invoke([HumanMessage(content=final_prompt)]) # 将最终回答添加到消息历史 state["messages"].append(AIMessage(content=response.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"{i+1}. {step}") print("\n=== 生成的SQL ===") print(final_state.get("sql_generated"))一个成功的运行流程可能是：
- Think: LLM看到问题，意识到需要查询数据库，但不确定表结构，决定调用
get_table_schema。 - Act: 执行
get_table_schema，返回表结构信息。 - Evaluate: 评估节点收到表结构，判断需要“继续思考”(continue_think)。
- Think: LLM根据表结构，生成SQL：
SELECT 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根据结果生成最终答案：“今年第一季度销售额最高的产品是‘智能手机’，总销售额为125，000元。”
- 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(model="llama3.2:7b") # 定义智能体角色 data_analyst = Agent( role=‘资深数据分析师‘, goal=‘从给定的数据源中准确提取、清洗和分析数据，并提供核心洞察‘, backstory=‘你拥有多年的数据分析经验，擅长从杂乱数据中发现规律。‘, tools=[query_database, get_table_schema], # 可以赋予不同的工具集 llm=llm, verbose=True ) report_writer = Agent( role=‘商业报告撰写专家‘, goal=‘将数据分析结果转化为结构清晰、观点明确、语言精炼的商业报告‘, backstory=‘你是一名前财经记者，擅长将复杂数据转化为易懂的故事。‘, llm=llm, verbose=True ) # 定义任务链 task1 = Task( description=“分析数据库‘sales.db‘中过去一年的产品销售趋势，找出增长最快的产品和下滑最严重的区域。”, agent=data_analyst, expected_output=“一份包含关键数据表格和初步洞察点的分析摘要。” ) task2 = Task( description=“基于数据分析师提供的摘要，撰写一份给管理层的半页纸商业简报，突出核心发现和建议。”, agent=report_writer, context=[task1], # 任务2依赖于任务1的输出 expected_output=“一份格式规范、语言专业的商业简报（Markdown格式）。” ) # 组建团队并运行 crew = Crew( agents=[data_analyst, report_writer], tasks=[task1, task2], process=Process.sequential, # 顺序执行 verbose=2 ) result = crew.kickoff() print(result)在这个配置下，data_analyst智能体会自主完成数据查询和分析，然后将结果传递给report_writer智能体去撰写报告。CrewAI自动管理它们之间的协作和上下文传递。
构建一个稳定、可靠的AI智能体是一个迭代的过程。从最简单的单循环智能体开始，逐步增加工具、完善状态管理、加入评估和护栏，最终扩展到多智能体协作。核心在于理解“状态流”和“决策逻辑”。本地模型（Ollama）和开源框架（LangGraph）的组合，为我们提供了低成本、高隐私、可深度定制的实践平台。
