基于Python編寫一個簡化版AI Agent智能體

更新時間：2025年08月20日 09:35:33 作者：MichaelIp

如果你覺得人工智能 (AI) 很有意思但又有點(diǎn)模糊,那么你一定會對 AI agent 感興趣,本文將使用Python編寫一個簡化版的AI Agent智能體,感興趣的小伙伴可以了解下

1. AI Agents的概念，它和LLM有什么不同？

如果你覺得人工智能 (AI) 很有意思但又有點(diǎn)模糊，那么你一定會對 AI agent 感興趣。沒錯，我們說的就是大語言模型 (LLM) 公司已經(jīng)投資了數(shù)十億來開發(fā)的 AI agent。AI agent 是軟件實體，可接收任務(wù)、檢查環(huán)境、根據(jù)角色執(zhí)行操作并根據(jù)經(jīng)驗進(jìn)行調(diào)整。

AI Agents與LLM的區(qū)別，可以簡單粗暴看看下面知乎大神畫的圖。

總體來說，Agent是在LLM的基礎(chǔ)上加上增加三個主要功能：Tools(工具使用), Memory (記憶), Planning (規(guī)劃)。下面就以這三個功能，用python簡單實現(xiàn)一下，幫助大家更好理解這三個功能。

2.工具使用 (Tool Use)

假設(shè)我有兩個工具，一個是“獲取當(dāng)前的日期和時間信息”，一個是“簡單計算器工具”，我將會問三個問題，

“現(xiàn)在幾點(diǎn)了？”,
“幫我計算 15 * 8 + 20”,
“今天是什么日期？”

下面是python代碼實現(xiàn)：

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.tools import Tool
import datetime

# 加載環(huán)境變量
load_dotenv()
os.environ["OPENAI_API_KEY"] = "sk-某寶35塊apikey"
api_key = os.getenv('OPENAI_API_KEY')

# ========== 1. 初始化 LLM（大語言模型） ==========
# LangChain 特點(diǎn)：統(tǒng)一的 LLM 接口，支持多種模型提供商
llm = ChatOpenAI(
    base_url="https://api.openai.com/v1",
    api_key=api_key,
    model="gpt-4o",
    temperature=0.7  # LangChain 特點(diǎn)：統(tǒng)一的參數(shù)配置
)

# ========== 2. LLMChain：Prompt → LLM → 輸出鏈的基本流程封裝 ==========
def demo_llm_chain():
    """
    演示 LLMChain：支持變量注入與模板復(fù)用的核心組件
    LangChain 特點(diǎn)：模板化提示詞管理，支持變量替換
    """
    print("=" * 50)
    print("?? LLMChain 演示：Prompt → LLM → 輸出鏈")
    print("=" * 50)
    
    # 創(chuàng)建提示詞模板 - LangChain 特點(diǎn)：模板復(fù)用
    prompt_template = PromptTemplate(
        input_variables=["topic", "style"],
        template="""
        請以{style}的風(fēng)格，寫一段關(guān)于{topic}的介紹。
        要求：簡潔明了，不超過100字。
        """
    )
    
    # LangChain 0.3 推薦使用 LCEL (LangChain Expression Language)
    # 這是新的鏈?zhǔn)浇M合方式：prompt | llm
    chain = prompt_template | llm
    
    # 執(zhí)行鏈 - 變量注入
    result = chain.invoke({"topic": "人工智能", "style": "科普"})
    print(f"?? LLMChain 輸出：\n{result.content}\n")
    
    return result.content

# ========== 3. Tools：工具系統(tǒng) ==========
def get_current_time(query: str) -> str:
    """獲取當(dāng)前時間的工具函數(shù)"""
    return f"當(dāng)前時間是：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

def calculate_simple(expression: str) -> str:
    """簡單計算器工具"""
    try:
        # 安全的數(shù)學(xué)表達(dá)式計算
        allowed_chars = set('0123456789+-*/.() ')
        if all(c in allowed_chars for c in expression):
            result = eval(expression)
            return f"計算結(jié)果：{expression} = {result}"
        else:
            return "錯誤：包含不允許的字符"
    except Exception as e:
        return f"計算錯誤：{str(e)}"

# LangChain 特點(diǎn)：統(tǒng)一的工具接口定義
tools = [
    Tool(
        name="get_time",
        func=get_current_time,
        description="獲取當(dāng)前的日期和時間信息"
    ),
    Tool(
        name="calculator",
        func=calculate_simple,
        description="執(zhí)行簡單的數(shù)學(xué)計算，如加減乘除運(yùn)算"
    )
]

def demo_tools():
    """演示 Tools 工具系統(tǒng)"""
    print("=" * 50)
    print("??? Tools 演示：工具系統(tǒng)")
    print("=" * 50)
    
    for tool in tools:
        print(f"工具名稱：{tool.name}")
        print(f"工具描述：{tool.description}")
        
        # 測試工具
        if tool.name == "get_time":
            result = tool.run("現(xiàn)在幾點(diǎn)了？")
        else:
            result = tool.run("10 + 5 * 2")
        
        print(f"工具輸出：{result}\n")

# ========== 4. 簡化版 Agents：手動工具選擇演示 ==========
def demo_simple_agents():
    """
    演示簡化版 Agents：手動工具選擇和執(zhí)行
    LangChain 特點(diǎn)：工具集成和智能選擇（這里用簡化版演示概念）
    """
    print("=" * 50)
    print("?? 簡化版 Agents 演示：工具選擇與執(zhí)行")
    print("=" * 50)
    
    # 創(chuàng)建工具選擇提示詞
    tool_selection_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一個智能助手，可以使用以下工具：
        1. get_time - 獲取當(dāng)前時間
        2. calculator - 執(zhí)行數(shù)學(xué)計算
        
        請分析用戶問題，選擇合適的工具并說明原因。
        只回答工具名稱和原因，格式：工具名稱|原因"""),
        ("human", "{question}")
    ])
    
    tool_chain = tool_selection_prompt | llm
    
    test_questions = [
        "現(xiàn)在幾點(diǎn)了？",
        "幫我計算 15 * 8 + 20",
        "今天是什么日期？"
    ]
    
    for question in test_questions:
        print(f"?? 用戶問題：{question}")
        
        # 1. 工具選擇
        selection_result = tool_chain.invoke({"question": question})
        print(f"?? 工具選擇：{selection_result.content}")
        
        # 2. 執(zhí)行工具（簡化版手動執(zhí)行）
        if "get_time" in selection_result.content.lower():
            result = get_current_time(question)
        elif "calculator" in selection_result.content.lower():
            # 提取數(shù)學(xué)表達(dá)式（簡化處理）
            if "15 * 8 + 20" in question:
                result = calculate_simple("15 * 8 + 20")
            else:
                result = "需要具體的數(shù)學(xué)表達(dá)式"
        else:
            result = "未找到合適的工具"
        
        print(f"??? 工具執(zhí)行結(jié)果：{result}\n")

def main():
    """主函數(shù)：依次演示各個核心組件"""
    print("?? LangChain 0.3 核心組件實戰(zhàn)演示")
    print("基于 OpenAI API 的完整示例（兼容版本）\n")
    
    try:
        # 1. LLMChain 演示（使用 LCEL）
        demo_llm_chain()
        
        # 2. Tools 演示
        demo_tools()
        
        # 3. 簡化版 Agents tools演示
        demo_simple_agents()
        
        print("? 所有演示完成！")
        
    except Exception as e:
        print(f"? 演示過程中出現(xiàn)錯誤：{str(e)}")
        print("請檢查 API 密鑰和網(wǎng)絡(luò)連接")

if __name__ == "__main__":
    main()

上面代碼大致分兩步問大模型：

Step1：“請分析用戶問題，選擇合適的工具并說明原因，只回答工具名稱和原因，格式：工具名稱|原因”。

Step2：提出模型的結(jié)果，再結(jié)合問題傳給對應(yīng)的方法，最后得出結(jié)論。

這是運(yùn)行結(jié)果：

3.記憶 (Memory)

假設(shè)我問大模型三個問題，如下所示，

conversations = [
        "我叫Michael，今年85歲",
        "我喜歡編程和閱讀",
        "你還記得我的名字嗎？",
        "我的愛好是什么？"
    ]

下面是完整代碼：

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.tools import Tool
import datetime

# 加載環(huán)境變量
load_dotenv()
os.environ["OPENAI_API_KEY"] = "sk-某寶35塊apikey"
api_key = os.getenv('OPENAI_API_KEY')

# ========== 1. 初始化 LLM（大語言模型） ==========
# LangChain 特點(diǎn)：統(tǒng)一的 LLM 接口，支持多種模型提供商
llm = ChatOpenAI(
    base_url="https://api.openai.com/v1",
    api_key=api_key,
    model="gpt-4o",
    temperature=0.7  # LangChain 特點(diǎn)：統(tǒng)一的參數(shù)配置
)


# ========== 5. Memory：記憶系統(tǒng) ==========
def demo_memory():
    """
    演示 Memory：對話記憶管理
    LangChain 特點(diǎn)：自動管理對話歷史
    """
    print("=" * 50)
    print("?? Memory 演示：記憶系統(tǒng)")
    print("=" * 50)
    
    # 使用簡化的記憶管理方式
    conversation_history = []
    
    # 創(chuàng)建帶記憶的對話提示詞
    memory_prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一個友好的助手，能夠記住之前的對話內(nèi)容。以下是對話歷史：{history}"),
        ("human", "{input}")
    ])
    
    memory_chain = memory_prompt | llm
    
    # 模擬多輪對話
    conversations = [
        "我叫Michael，今年85歲",
        "我喜歡編程和閱讀",
        "你還記得我的名字嗎？",
        "我的愛好是什么？"
    ]
    
    for i, user_input in enumerate(conversations, 1):
        print(f"?? 第{i}輪對話：{user_input}")
        
        # 構(gòu)建歷史記錄字符串
        history_str = "\n".join([f"用戶: {h['user']}\n助手: {h['assistant']}" for h in conversation_history])
        
        # 獲取回復(fù)
        response = memory_chain.invoke({
            "history": history_str,
            "input": user_input
        })
        
        print(f"?? 助手回復(fù)：{response.content}\n")
        
        # 更新對話歷史
        conversation_history.append({
            "user": user_input,
            "assistant": response.content
        })
        
        # 顯示當(dāng)前記憶內(nèi)容
        print(f"?? 當(dāng)前記憶：{len(conversation_history)} 輪對話")
        print(f"?? 記憶內(nèi)容：{conversation_history}")
        print("-" * 30)

def main():
    try:

        # Memory 演示
        demo_memory()

        print("? 所有演示完成！")
        
    except Exception as e:
        print(f"? 演示過程中出現(xiàn)錯誤：{str(e)}")
        print("請檢查 API 密鑰和網(wǎng)絡(luò)連接")

if __name__ == "__main__":
    main()

通過上面代碼，我們可以看出，最關(guān)鍵的部分是conversation_history.append，每次將對話信息保留，再傳給大模型。

下面是運(yùn)行結(jié)果：

隨著對話內(nèi)容越來越多，會影響大模型的回復(fù)效率，甚至?xí)鰐oken范圍，所以每個一段時間或?qū)υ挻螖?shù)超過某個閾值，我們需要大模型對所以歷史對話做一個總結(jié)，來壓縮歷史對話的內(nèi)容。

4.規(guī)劃 (Planning)

假設(shè)我有幾個工具，我問大模型’我要去日本東京，幫我安排行程。'，下面是可以用到的提示詞

你是一個專業(yè)任務(wù)規(guī)劃AI。請將以下任務(wù)分解為5-7個有序的關(guān)鍵步驟。
輸出要求:
1. 只輸出JSON格式，而且JSON格式要正確
2. JSON內(nèi)容包含一個"steps"數(shù)組，每個步驟是一個對象，包含"id"(數(shù)字序號)和"description"(用到的步驟，如research_destination)
3. 確保步驟間有邏輯順序和依賴關(guān)系
可以用到的步驟：
1. research_destination - 進(jìn)行目的地研究 - 收集當(dāng)?shù)匚幕?、天氣和景點(diǎn)信息
2. book_transportation - 預(yù)訂交通工具 - 選擇航班和當(dāng)?shù)亟煌ǚ绞?br /> 3. reserve_accommodation - 安排住宿 - 根據(jù)預(yù)算選擇酒店或民宿
4. plan_activities - 規(guī)劃每日活動 - 創(chuàng)建景點(diǎn)游覽和餐飲計劃
5. prepare_documents - 準(zhǔn)備旅行文件 - 檢查簽證、打印預(yù)訂確認(rèn)單
6. pack_luggage - 整理行李 - 根據(jù)天氣準(zhǔn)備衣物和必需品
7. calculator - 執(zhí)行數(shù)學(xué)計算
8. confirm_arrival - 確認(rèn)抵達(dá)安排 - 聯(lián)系接機(jī)服務(wù)和酒店

可以看到，我給大模型挖了個坑，多了一個‘calculator - 執(zhí)行數(shù)學(xué)計算’的步驟，看大模型怎么給我們安排。

下面是完整代碼

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.tools import Tool
import json

# 加載環(huán)境變量
load_dotenv()
os.environ["OPENAI_API_KEY"] = "sk-淘寶key"
api_key = os.getenv('OPENAI_API_KEY')

# ========== 1. 初始化 LLM（大語言模型） ==========
# LangChain 特點(diǎn)：統(tǒng)一的 LLM 接口，支持多種模型提供商
llm = ChatOpenAI(
    base_url="https://api.openai.com/v1",
    api_key=api_key,
    model="gpt-4o",
    temperature=0.7  # LangChain 特點(diǎn)：統(tǒng)一的參數(shù)配置
)



# 以下是具體的子任務(wù)方法
def research_destination():
    print("進(jìn)行目的地研究 - 收集當(dāng)?shù)匚幕?、天氣和景點(diǎn)信息")

def book_transportation():
    print("預(yù)訂交通工具 - 選擇航班和當(dāng)?shù)亟煌ǚ绞?)

def reserve_accommodation():
    print("安排住宿 - 根據(jù)預(yù)算選擇酒店或民宿")

def plan_activities():
    print("規(guī)劃每日活動 - 創(chuàng)建景點(diǎn)游覽和餐飲計劃")

def prepare_documents():
    print("準(zhǔn)備旅行文件 - 檢查簽證、打印預(yù)訂確認(rèn)單")

def pack_luggage():
    print("整理行李 - 根據(jù)天氣準(zhǔn)備衣物和必需品")

def confirm_arrival():
    print("確認(rèn)抵達(dá)安排 - 聯(lián)系接機(jī)服務(wù)和酒店")

def calculator():
    print(12*25)

# LangChain 特點(diǎn)：統(tǒng)一的工具接口定義
tools = [
    Tool(
        name="research_destination",
        func=research_destination,
        description="進(jìn)行目的地研究 - 收集當(dāng)?shù)匚幕?、天氣和景點(diǎn)信息"
    ),
    Tool(
        name="book_transportation",
        func=book_transportation,
        description="預(yù)訂交通工具 - 選擇航班和當(dāng)?shù)亟煌ǚ绞?
    ),
    Tool(
        name="reserve_accommodation",
        func=reserve_accommodation,
        description="安排住宿 - 根據(jù)預(yù)算選擇酒店或民宿"
    ),
    Tool(
        name="plan_activities",
        func=book_transportation,
        description="規(guī)劃每日活動 - 創(chuàng)建景點(diǎn)游覽和餐飲計劃"
    ),
    Tool(
        name="prepare_documents",
        func=prepare_documents,
        description="準(zhǔn)備旅行文件 - 檢查簽證、打印預(yù)訂確認(rèn)單"
    ),
    Tool(
        name="pack_luggage",
        func=pack_luggage,
        description="整理行李 - 根據(jù)天氣準(zhǔn)備衣物和必需品"
    ),
    Tool(
        name="calculator",
        func=calculator,
        description="執(zhí)行簡單的數(shù)學(xué)計算，如加減乘除運(yùn)算"
    ),
    Tool(
        name="confirm_arrival",
        func=confirm_arrival,
        description="確認(rèn)抵達(dá)安排 - 聯(lián)系接機(jī)服務(wù)和酒店"
    )
]
def explain_and_process(json_data):
    """
    解釋并處理旅行計劃JSON數(shù)據(jù)

    參數(shù):
    json_data -- 包含旅行計劃步驟的JSON字符串

    返回:
    處理后的計劃步驟列表
    """
    # 1. 解析JSON數(shù)據(jù)
    try:
        plan_data = json.loads(json_data)
        print("? JSON解析成功！數(shù)據(jù)結(jié)構(gòu)驗證通過")
    except json.JSONDecodeError as e:
        print(f"? JSON解析錯誤: {str(e)}")
        return None

    # 2. 解釋數(shù)據(jù)結(jié)構(gòu)
    print("\n?? JSON結(jié)構(gòu)分析:")
    print(f"- 頂層對象包含 {len(plan_data)} 個鍵: {list(plan_data.keys())}")
    print(f"- 'steps' 鍵包含 {len(plan_data['steps'])} 個步驟")

    # 3. 提取并處理步驟數(shù)據(jù)
    steps = []
    print("\n?? 計劃步驟詳情:")
    for step in plan_data["steps"]:
        # 驗證必要字段
        if "id" not in step or "description" not in step:
            print(f"?? 警告: 步驟缺少必要字段 - {step}")
            continue

        # 創(chuàng)建步驟對象
        step_obj = {
            "id": step["id"],
            "description": step["description"],
            "status": "pending"  # 添加執(zhí)行狀態(tài)
        }
        steps.append(step_obj)

        # 打印步驟信息
        print(f"步驟 {step['id']}: {step['description']}")

        if "research_destination" in step['description'].lower():
            research_destination()
        elif "book_transportation" in step['description'].lower():
            book_transportation()
        elif "reserve_accommodation" in step['description'].lower():
            reserve_accommodation()
        elif "plan_activities" in step['description'].lower():
            plan_activities()
        elif "prepare_documents" in step['description'].lower():
            prepare_documents()
        elif "pack_luggage" in step['description'].lower():
            pack_luggage()
        elif "calculator" in step['description'].lower():
            calculator()
        elif "confirm_arrival" in step['description'].lower():
            confirm_arrival()

    # 4. 添加額外信息
    print("\n? 處理后的數(shù)據(jù)結(jié)構(gòu):")
    print(f"- 總步驟數(shù): {len(steps)}")
    print(f"- 首個步驟: {steps[0]['description']}")
    print(f"- 最后步驟: {steps[-1]['description']}")

    return steps

def plan_agents():
    """
    演示簡化版 Agents：手動工具選擇和執(zhí)行
    LangChain 特點(diǎn)：工具集成和智能選擇（這里用簡化版演示概念）
    """
    print("=" * 50)
    print("?? 簡化版 Agents 演示：計劃選擇與執(zhí)行")
    print("=" * 50)

    # 創(chuàng)建工具選擇提示詞
    tool_selection_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一個專業(yè)任務(wù)規(guī)劃AI。請將以下任務(wù)分解為5-7個有序的關(guān)鍵步驟。
        輸出要求:
        1. 只輸出JSON格式，而且JSON格式要正確
        2. JSON內(nèi)容包含一個"steps"數(shù)組，每個步驟是一個對象，包含"id"(數(shù)字序號)和"description"(用到的步驟，如research_destination)
        3. 確保步驟間有邏輯順序和依賴關(guān)系
        可以用到的步驟：
        1. research_destination - 進(jìn)行目的地研究 - 收集當(dāng)?shù)匚幕⑻鞖夂途包c(diǎn)信息
        2. book_transportation - 預(yù)訂交通工具 - 選擇航班和當(dāng)?shù)亟煌ǚ绞?
        3. reserve_accommodation - 安排住宿 - 根據(jù)預(yù)算選擇酒店或民宿
        4. plan_activities - 規(guī)劃每日活動 - 創(chuàng)建景點(diǎn)游覽和餐飲計劃
        5. prepare_documents - 準(zhǔn)備旅行文件 - 檢查簽證、打印預(yù)訂確認(rèn)單
        6. pack_luggage - 整理行李 - 根據(jù)天氣準(zhǔn)備衣物和必需品
        7. calculator - 執(zhí)行數(shù)學(xué)計算
        8. confirm_arrival - 確認(rèn)抵達(dá)安排 - 聯(lián)系接機(jī)服務(wù)和酒店
        """),
        ("human", "{question}")
    ])

    tool_chain = tool_selection_prompt | llm

    question = '我要去日本東京，幫我安排行程。'
    print(f"?? 用戶問題：{question}")

    # 1. 工具選擇
    selection_result = tool_chain.invoke({"question": question})
    response_json = selection_result.content.replace('json','').replace('```','')
    print(f"?? 返回json：{response_json}")
    explain_and_process(response_json)

def main():
    try:

        print("=== AI旅行規(guī)劃Agent演示 ===")

        plan_agents()

        print("? 所有演示完成！")
        
    except Exception as e:
        print(f"? 演示過程中出現(xiàn)錯誤：{str(e)}")
        print("請檢查 API 密鑰和網(wǎng)絡(luò)連接")

if __name__ == "__main__":
    main()

執(zhí)行結(jié)果：

可以看出，大模型會先安排所需要的計劃，再去執(zhí)行，這就是Agent的‘計劃制定’能力，而Agent的計劃能力還包括‘計劃反思’。這里不做展示，后續(xù)可以試圖添加這個功能。

計劃反思

制定計劃后，必須對其優(yōu)點(diǎn)進(jìn)行反思和評估。Agent需要利用反饋機(jī)制（通常是從已有模型中汲取靈感）來完善和改進(jìn)其戰(zhàn)略和規(guī)劃方法。為了更好地與人類的價值觀和偏好保持一致，Agent會主動與人類接觸，從而糾正一些誤解，并將這些有針對性的反饋吸收到其規(guī)劃方法中。此外，它們還可以從有形或虛擬環(huán)境中獲得反饋，如任務(wù)完成情況的提示或行動后的觀察，幫助它們修改和完善計劃。