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"))

一个成功的运行流程可能是：

  1. Think : LLM看到问题，意识到需要查询数据库，但不确定表结构，决定调用 get_table_schema 。
  2. Act : 执行 get_table_schema ，返回表结构信息。
  3. Evaluate : 评估节点收到表结构，判断需要“继续思考”(continue_think)。
  4. 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 。
  5. Act : 执行 query_database ，返回结果。
  6. Evaluate : 查询成功，判断需要“解释”(explain)。
  7. Explain : LLM根据结果生成最终答案：“今年第一季度销售额最高的产品是‘智能手机’，总销售额为125，000元。”
  8. 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）的组合，为我们提供了低成本、高隐私、可深度定制的实践平台。

# AI智能体 # Ollama
Logo
亚马逊云科技技术品牌专区

更多推荐

cover

百万Token不是数字游戏：Kimi K3完整吞下10万行代码库后，我测出了它的真实理解力

avatar 亚马逊云科技技术品牌专区
cover

【2026最新版】Codex换肤助手下载和使用教程：一键美化 Codex 客户端，支持自定义主题

avatar 亚马逊云科技技术品牌专区

FastAPI + OpenAI 兼容协议 + DeepSeek 实战：大模型 Function Calling 工具调用全拆解

模型本身不能直接查天气、查数据库。我们告诉模型"你有这些函数可用"，模型在觉得需要时返回"我想调用某个函数、参数是这样"，真正的函数由我们本地执行，再把结果交还给模型整理成话。这就是"模型决策 + 本地执行"的混合模式。

avatar 亚马逊云科技技术品牌专区