基础介绍

1 什么是 AI Agent

AI Agent 是基于大语言模型（LLM）的智能体，能够自主感知→规划→行动→观察，从而完成复杂任务。
与单纯调用 LLM（输入-思考-输出）不同，Agent 更贴近人类“先拆任务、再动手、边干边改”的解决方式 。

开发框架：AI Agent的应用开发基于LangChain框架。
AI Agent与LM及人类的对比
LM的局限性：LM主要是接收输入，通过内部思考得到输出，缺乏记忆、工具使用和规划能力。
人类解决问题的特点：人类在接收外部环境输入、思考、输出之外，还会运用记忆、工具和规划来解决问题，例如老板下发任务后，人类会拆解任务、规划步骤、寻求帮助。
Agent对人类思维的模拟：Agent正是通过量化人类这种思维习惯来构建的，使其具备更复杂的任务处理能力。

2 AI Agent vs LLM vs 人类

例子：让纯 LLM 订机票，它只返回文字；Agent 会查天气→比价→下单→发日历邀请 。

3 核心 4 组件（2026 工业界共识）

1. 感知（Perception）
   接收多模态输入：文本、语音、传感器、API 回调 。

2. 记忆（Memory）

   • 短期：ConversationBufferWindow，随对话滑动
   • 长期：Chroma/Qdrant + Embedding，支持 Session 找回

3. 规划（Planning）

   • 单链：ReAct = Thought→Action→Observation
   • 多步：思维树 / CARS，可回溯
   • 状态图：LangGraph 支持循环、条件分支

4. 工具（Tools）&行动（Action）
   日历、搜索、代码解释器、REST API；通过 Function Calling 由 LLM 自行选择 。

记忆（Memory）：
相当于与模型对话的连贯上下文，用于分步解决问题时记住前面的步骤。分为短期记忆和长时记忆。短期记忆在编程练习时通常存于容器，再次调用接口可能清空；长时记忆在用户服务中会持久化存储到内存并给予Session ID，以便用户再次使用时读取之前的对话历史。
规划（Planning）：
指执行任务时的步骤安排，如第一步、第二步等，以及在执行过程中根据反馈是否需要重新计划。规划有很多框架，如思维链、思维树、ReAct框架、CARS框架等，后续会挑选Star数高或经测试好用的框架详细讲解。
工具（Tools）：
用于让LM连接外部世界，最典型的是Function Calling函数调用。例如，LM本身不具备联网获取天气的能力，但可通过外接API实现。工具还包括日历、计算、代码执行器、搜索等，后续会详细讲解工具的使用。
行动（Action）：
当整个任务循环想要执行时，开始当前的活动，是任务执行的具体动作。

4 决策循环流程（ReAct 经典 4 步）

感知输入 → 规划拆解 → 调用工具行动 → 观察反馈 →（循环直至目标）

📖 Part-A 整体知识速记

Agent 决策循环（ReAct 经典 4 步）
感知 → 思考 → 行动 → 观察 →（循环直到目标完成）

🛠 Part-B 10 行代码跑通「搜索+计算」Agent

# 依赖：首次运行请取消注释
# !pip install langchain==0.3.5 langchain-openai duckduckgo-search
import langchain; print('LangChain版本：', langchain.__version__)

LangChain版本： 1.2.0

#!pip install -U ddgs

# ========== LangChain 1.2.1 极简适配（无任何导入错误） ==========
from langchain_community.chat_models.ollama import ChatOllama
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.prompts import ChatPromptTemplate

# ========== 1. 初始化核心组件（保留你的 Ollama 地址） ==========
# 初始化 Ollama 模型（你的远程服务地址）
llm = ChatOllama(
    base_url="http://192.168.52.88:11434", model="qwen2:1.5b", temperature=0.1
)

# 初始化 DuckDuckGo 搜索工具
search_tool = DuckDuckGoSearchRun()


# ========== 2. 核心逻辑：先搜索，再让 LLM 生成 Markdown 表格 ==========
def get_ai_llm_rank_2026():
    # 步骤1：调用搜索工具获取原始信息
    try:
        print("🔍 正在搜索2026年AI大语言模型排名...")
        search_query = "2026年AI大语言模型排名 权威榜单"
        raw_search_result = search_tool.run(search_query)
        if not raw_search_result:
            return "❌ 未搜索到相关信息"
    except Exception as e:
        return f"❌ 搜索失败：{str(e)[:100]}"

    # 步骤2：定义 Prompt，让 LLM 将搜索结果转为 Markdown 表格
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                """你是数据格式化助手，严格按以下规则处理：
        1. 基于提供的搜索结果，提取2026年AI大语言模型排名信息；
        2. 仅输出 Markdown 表格格式，包含「框架名称」「排名」「核心特点」三列；
        3. 表格格式规范，无多余文字、无解释、无思考过程；
        4. 若信息不全，按行业常识补充核心特点（仅补充，不编造排名）。""",
            ),
            ("user", "搜索结果：{raw_result}\n请按要求输出Markdown表格"),
        ]
    )

    # 步骤3：调用 LLM 生成格式化结果
    chain = prompt | llm  # 1.2.1 支持的链式调用（核心特性）
    formatted_result = chain.invoke({"raw_result": raw_search_result})

    return formatted_result.content


# ========== 3. 执行并输出结果 ==========
if __name__ == "__main__":
    # 执行查询
    final_result = get_ai_llm_rank_2026()

    # 格式化输出
    print("\n" + "=" * 80)
    print("✅ 2026年AI大模型排名（Markdown表格）：")
    print("=" * 80)
    print(final_result)

🔍 正在搜索2026年AI大语言模型排名...

================================================================================
✅ 2026年AI大模型排名（Markdown表格）：
================================================================================
| 框架名称 | 排名 | 核心特点 |
| --- | --- | --- |
| Qwen2.5-1.5B-Instruct | 1 | 简单易用，适用于多种场景 |
| Llama3 | 2 | 高效文本生成能力，适合大量数据处理 |
| Gemma | 3 | 强大的语言模型，广泛应用于自然语言处理领域 |
| MaziyarPanahi/calme-3.2-instruct-78b | 4 | 强大的情绪识别和情感分析能力 |
| shuttleai/shuttle-3 | 5 | 高效的文本生成和翻译能力 |
| Hugging Face | 6 | 大量开源 模 型 和工具，支持多种语言 |
| Alibaba's Qwen | 7 | 与Qwen2.5-1.5B-Instruct相似，但具有更广泛的使用场景 |
| OpenAI's Codex | 8 | 强大的文本生成能力，适用于大量数据处理 |
| Google's BERT | 9 | 高效的自然语言处理模型，广泛应用于搜索引擎和问答系统 |
| Microsoft's T5 | 10 | 强大的文本生成和翻译能力，适合多种应用场景 |
| Alibaba's Qwen2.5-1.5B-Instruct | 11 | 具有更广泛的使用场景，适用于多种自然语言处理任务 |

运行后可见 Thought→Action→Observation 全流程日志

🖥 Part-C 本地免费方案：Ollama + LangChain + qdrantDB

# 核心导入
from langchain_community.chat_models import ChatOllama
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables import RunnableLambda
from langchain_core.messages import SystemMessage
from langchain_core.documents import Document
import traceback

# Qdrant + LangChain 检索相关导入
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Qdrant
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams

# ====================== 核心配置 ======================
# Ollama远程配置
OLLAMA_MODEL = "qwen2:1.5b"
OLLAMA_BASE_URL = "http://192.168.52.88:11434"

# Qdrant远程配置
QDRANT_URL = "http://192.168.52.88:6333"
QDRANT_COLLECTION = "qwen2_remote_memory"
QDRANT_VECTOR_SIZE = 1536  # qwen2:1.5b固定维度

# ====================== 初始化组件（关键：创建LangChain Qdrant VectorStore） ======================
try:
    # 1. 初始化Ollama LLM
    llm = ChatOllama(
        model=OLLAMA_MODEL,
        base_url=OLLAMA_BASE_URL,
        temperature=0,
        num_ctx=2048
    )
    print("✅ Ollama LLM初始化成功")

    # 2. 初始化Ollama Embeddings
    embeddings = OllamaEmbeddings(
        model=OLLAMA_MODEL,
        base_url=OLLAMA_BASE_URL,
        embed_instruction="为文本生成用于检索的向量："
    )
    # 验证向量维度
    test_emb = embeddings.embed_query("测试")
    assert len(test_emb) == QDRANT_VECTOR_SIZE, f"向量维度错误：{len(test_emb)}"
    print("✅ Embeddings初始化成功")

    # 3. 初始化远程Qdrant客户端
    qdrant_client = QdrantClient(
        url=QDRANT_URL,
        timeout=30,
        check_compatibility=False
    )
    qdrant_client.get_collections()
    print("✅ 远程Qdrant连接成功")

    # 4. 创建/确认Qdrant集合
    if not qdrant_client.collection_exists(QDRANT_COLLECTION):
        qdrant_client.create_collection(
            collection_name=QDRANT_COLLECTION,
            vectors_config=VectorParams(size=QDRANT_VECTOR_SIZE, distance=Distance.COSINE)
        )
        print(f"✅ Qdrant集合 {QDRANT_COLLECTION} 创建成功")

    # 5. 关键：创建LangChain Qdrant VectorStore（核心，用于as_retriever）
    vector_store = Qdrant(
        client=qdrant_client,
        collection_name=QDRANT_COLLECTION,
        embeddings=embeddings
    )
    print("✅ LangChain Qdrant VectorStore初始化成功")

    # 6. 创建MMR检索器（按你的参数配置）
    retriever = vector_store.as_retriever(
        search_type="mmr",  # 使用MMR检索策略
        search_kwargs={
            "k": 1,            # 返回前1个结果
            "fetch_k": 1,      # 先获取1个最相似文档
            "lambda_mult": 0.7 # 平衡相关性（值越大越注重相关）
        }
    )
    print("✅ MMR检索器初始化成功")

except Exception as e:
    print(f"❌ 组件初始化失败：{type(e).__name__}: {e}")
    traceback.print_exc()
    exit(1)

# ====================== 长时记忆函数（替换为MMR检索） ======================
def save_to_long_memory(input_text: str, output_text: str):
    """保存到Qdrant（用LangChain封装，避免底层API）"""
    try:
        doc = Document(page_content=f"用户：{input_text}\nAI：{output_text}")
        vector_store.add_documents([doc])  # 用vector_store保存，适配所有版本
        print("✅ 长时记忆保存成功")
    except Exception as e:
        print(f"❌ 长时记忆保存失败：{type(e).__name__}: {e}")
        traceback.print_exc()

def retrieve_long_memory(query: str) -> list:
    """用MMR检索器实现长时记忆检索（彻底规避底层API）"""
    try:
        # 方法1：先尝试用LangChain检索器（推荐）
        docs = retriever.invoke(query)
        
        # 解析检索结果
        retrieved_texts = []
        for doc in docs:
            retrieved_texts.append(doc.page_content)  # 获取文档内容
        
        # 转换为SystemMessage列表（适配MessagesPlaceholder）
        if retrieved_texts:
            long_history_text = "\n".join(retrieved_texts)
            return [SystemMessage(content=f"长时记忆：{long_history_text}")]
        else:
            return [SystemMessage(content="无相关长时记忆")]
    except AttributeError as e:
        # 方法2：如果LangChain调用失败，降级到直接使用Qdrant客户端API
        print(f"⚠️  LangChain检索失败，降级到直接Qdrant查询：{e}")
        try:
            # 生成查询向量
            query_vector = embeddings.embed_query(query)
            # 使用新版API query_points进行搜索
            search_results = qdrant_client.query_points(
                collection_name=QDRANT_COLLECTION,
                query=query_vector,  # 注意参数名从query_vector变为query
                limit=1
            )
            
            retrieved_texts = []
            for point in search_results.points:
                if point.payload and "page_content" in point.payload:
                    retrieved_texts.append(point.payload["page_content"])
            
            if retrieved_texts:
                long_history_text = "\n".join(retrieved_texts)
                return [SystemMessage(content=f"长时记忆：{long_history_text}")]
            else:
                return [SystemMessage(content="无相关长时记忆")]
        except Exception as inner_e:
            print(f"❌ 降级查询也失败：{type(inner_e).__name__}: {inner_e}")
            traceback.print_exc()
            return [SystemMessage(content="长时记忆检索出错")]
    except Exception as e:
        print(f"❌ 长时记忆检索失败：{type(e).__name__}: {e}")
        traceback.print_exc()
        return [SystemMessage(content="长时记忆检索出错")]
        

# ====================== 会话记忆链（保留核心逻辑） ======================
# Prompt模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是友好的中文助手，优先参考长时记忆回答。"),
    MessagesPlaceholder(variable_name="long_history"),
    MessagesPlaceholder(variable_name="short_history"),
    ("human", "{input}"),
])

# 整合记忆逻辑
def integrate_memories(inputs):
    try:
        return {
            "long_history": retrieve_long_memory(inputs["input"]),
            "short_history": inputs.get("short_history", []),
            "input": inputs["input"]
        }
    except Exception as e:
        print(f"❌ 记忆整合失败：{type(e).__name__}: {e}")
        traceback.print_exc()
        return {
            "long_history": [SystemMessage(content="记忆整合出错")],
            "short_history": [],
            "input": inputs["input"]
        }

# 完整链
chain = (
    RunnableLambda(integrate_memories)
    | prompt
    | llm
)

# 短期记忆存储
store = {}
def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 包装会话记忆
conversation_with_memory = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="short_history",
)

# ====================== 测试调用（验证MMR检索） ======================
# ====================== 测试调用（验证MMR检索） ======================
if __name__ == "__main__":
    input1 = "你好，我叫西米，今年5岁，喜欢贝乐虎识字，家住福州"
    try:
        # 第一轮调用（保存记忆）
        response1 = conversation_with_memory.invoke(
            {"input": input1},
            config={"configurable": {"session_id": "user123"}}
        )
        print(f"✅ 第一轮回复：{response1.content}")
        # 保存到长时记忆
        save_to_long_memory(input1, response1.content)

        # === 新增：显式调用并打印检索到的长时记忆 ===
        print("\n=== 显式验证长时记忆检索 ===")
        test_query = "西米"  # 检索关键词
        retrieved_memory_messages = retrieve_long_memory(test_query)
        # 打印检索结果
        for msg in retrieved_memory_messages:
            print(f"检索返回的消息: {msg.content}")
        print("=== 结束显式验证 ===\n")

        # 清空短期记忆后调用LLM
        store["user123"] = ChatMessageHistory()
        print("⚠️  已清空短期记忆 (ChatMessageHistory)")
        response2 = conversation_with_memory.invoke(
            {"input": "你还记得我叫什么？喜欢什么吗？"},
            config={"configurable": {"session_id": "user123"}}
        )
        print(f"\n✅ 第二轮回复：{response2.content}")

    except Exception as e:
        print(f"❌ 调用失败：{type(e).__name__}: {e}")
        traceback.print_exc()

✅ Ollama LLM初始化成功
✅ Embeddings初始化成功
✅ 远程Qdrant连接成功
✅ LangChain Qdrant VectorStore初始化成功
✅ MMR检索器初始化成功
⚠️  LangChain检索失败，降级到直接Qdrant查询：'QdrantClient' object has no attribute 'search'
✅ 第一轮回复：你好，西米！很高兴见到你。请问有什么可以帮助你的吗？
✅ 长时记忆保存成功

=== 显式验证长时记忆检索 ===
⚠️  LangChain检索失败，降级到直接Qdrant查询：'QdrantClient' object has no attribute 'search'
检索返回的消息: 长时记忆：用户：你好，我叫西米，今年5岁，喜欢贝乐虎识字，家住福州
AI：你好，小明！很高兴见到你。请问有什么可以帮助你的吗？
=== 结束显式验证 ===

⚠️  已清空短期记忆 (ChatMessageHistory)
⚠️  LangChain检索失败，降级到直接Qdrant查询：'QdrantClient' object has no attribute 'search'

✅ 第二轮回复：你好，西米！很高兴见到你。根据我的长时记忆，我记得你叫西米，今年5岁，来自福州，并且你喜欢贝乐虎识字。

#!pip show qdrant-client 

import qdrant_client
print(hasattr(qdrant_client.QdrantClient, "query_points"))  # 应输出True

True

Agent 决策流程应用场景举例

以「用户一句“我有点冷”」为例，完整复盘感知→规划→行动→观察→循环的智能家居温控实战

调节温度需先从外部环境获取输入，例如用户说“我感觉有点冷，能不能把温度调高一些”，这相当于给LM的prompt指令，智能系统会通过语音识别和情感分析技术感知用户需求，如判断房间温度低需提高温度，且情感分析能让AI回复带上人类情绪。
agent决策流程之规划阶段
规划的必要性与多样性：有了感知后需对需求进行规划，如确定温度调高多少，规划有多种可能性，类似数学题有多种解法，思路不同解法不同，若思路错误则需重新制定。
规划的具体步骤与依据：针对调高温度的需求，规划步骤包括：一是获取当前室内温度作为基准；二是根据用户偏好和当前温度决定升高几度，这里利用了memory能力，用户偏好有记录，如可能设置为20°或25°，但规划时还需综合评估其他信息，如用户体重变化（去年120斤怕冷，今年160斤怕热）；三是调整温度设置并告知用户。
agent决策流程之行动与反馈观察阶段
系统执行计划的行动：规划后开始行动，系统执行计划，先检查当前温度，如检测到室内温度20°C，根据用户偏好23°，则调到23°并通过语音助手反馈用户“已将温度调高到23摄氏度，请稍等温度将逐渐上升”。
结果的反馈观察与循环调整：调到23°不一定是对的结果，需对结果进行反馈观察，观察房间温度是否到23°及用户反馈是否合理。若用户说温度刚好，系统感知调节OK；若用户仍觉得冷，则replay计划，调整计划进一步调高温度，调高阈值可规划为1°C、2°C调或按用户明确指令调（如调到25°C）。若需调整则反过头重新制定计划，形成循环，直到task任务完成才终止循环。

1 回顾与场景设定

• 概念痛点：只讲抽象流程，听众对“task 到底长啥样”无感。
• 具体场景：家庭客厅，任务——把室内温度和灯光调到“家人觉得舒服” 。

2 需求感知：一句话如何变成结构化输入

1. 语音捕获 → “我感觉有点冷，能不能把温度调高一些？”
2. 语音识别 + 情感分析 → 紧急度 8/10，情绪偏“弱请求”。
3. 参数抽取（Agent 感知输出）：
   • intent: raise_temp
   • target_temp: null（未给数值）
   • urgency: 8
   • location: living_room

情感模块让回复带温度：“好的，我这就帮您升温～”。

3 规划阶段：把“冷”翻译成可执行步骤

Agent 在 300 ms 内并行检查 3 条分支，最终合成 Plan-A：

输出 Plan-A JSON：

[{"op":"set_temp","value":22},{"op":"set_light","brightness":70},
 {"op":"notify","text":"已调到 22 ℃，稍等"},{"op":"wait_feedback","timeout":120}]