經(jīng)濟模擬中的架構設計與實現(xiàn))
1. 項目概述當AI智能體走進動態(tài)經(jīng)濟沙盤最近在AI和模擬經(jīng)濟交叉的圈子里一個叫“EconAI”的概念討論度挺高。簡單來說它試圖用大語言模型驅動的智能體在一個持續(xù)變化的經(jīng)濟環(huán)境里模擬具有“成長性”和“記憶力”的決策者。這聽起來有點像把《模擬人生》里的角色丟進一個動態(tài)的宏觀經(jīng)濟模型里讓他們不僅要賺錢糊口還要學習、適應甚至預測經(jīng)濟周期的波動。傳統(tǒng)的經(jīng)濟模型或游戲AI角色的行為模式往往是靜態(tài)的或者基于一套預設的、有限的規(guī)則。比如一個模擬中的“消費者”智能體其消費傾向可能是固定的或者只對價格做出簡單反應。但現(xiàn)實中的經(jīng)濟參與者不是這樣的我們的決策深受過去經(jīng)驗記憶、對未來的預期預測以及自身身份認知人設的影響。一個經(jīng)歷了經(jīng)濟蕭條的人其儲蓄和投資行為會與一直生活在繁榮期的人截然不同。EconAI瞄準的正是用LLM賦予智能體這種動態(tài)演化的“人設”和持續(xù)積累的“記憶”能力讓它們在模擬中更像真實的、會學習的個體。這項目適合誰呢如果你是對AI智能體架構、多智能體模擬、LLM應用落地或者計算經(jīng)濟學感興趣的研究者、開發(fā)者或學生這里面的設計思路和實現(xiàn)挑戰(zhàn)會很有嚼頭。它不是一個開箱即用的產(chǎn)品更像一個前沿的實驗框架探討如何將LLM的認知能力與復雜系統(tǒng)仿真結合起來。接下來我會結合當前Agent開發(fā)的熱點拆解EconAI可能的核心設計、實操難點以及我們?nèi)绾蝿邮执罱ㄒ粋€簡化版本。2. 核心架構設計記憶、人格與環(huán)境的三角博弈EconAI的吸引力在于它試圖協(xié)調(diào)三個動態(tài)變化的復雜系統(tǒng)智能體的內(nèi)部狀態(tài)人格與記憶、智能體間的社會交互以及外部的經(jīng)濟環(huán)境。要讓這個系統(tǒng)運轉起來而不是陷入混沌或僵化架構設計上必須解決幾個核心問題。2.1 動態(tài)人格演化的實現(xiàn)機制“動態(tài)人格”是讓智能體擺脫腳本的關鍵。這里的“人格”不是一個標簽而是一套可量化、可調(diào)整的決策參數(shù)和偏好集合。例如可以包括風險厭惡系數(shù)、消費傾向、學習能力、社交活躍度等。一個可行的實現(xiàn)機制是基于經(jīng)驗反饋的參數(shù)微調(diào)。智能體初始人格由LLM根據(jù)一段簡短的背景描述生成例如“你是一個初入職場、謹慎的年輕程序員”。LLM會輸出一組初始參數(shù)。之后每一次重大經(jīng)濟決策如是否進行一筆高風險投資的結果都會成為一個反饋信號。系統(tǒng)需要設計一個人格更新函數(shù)。這個函數(shù)不直接讓LLM修改參數(shù)而是將決策結果盈利/虧損、環(huán)境狀態(tài)市場繁榮/衰退和智能體自身的感受由LLM總結作為輸入生成對人格參數(shù)的調(diào)整量。注意避免讓LLM直接輸出“我的風險厭惡系數(shù)從0.7調(diào)整為0.65”。這會導致不可控的漂移。更好的做法是讓LLM以文本形式總結本次經(jīng)驗教訓例如“這次冒險投資成功讓我信心大增未來可能愿意承擔稍多一點風險”然后由一個確定的、可控的規(guī)則模塊將這段文本映射到一個小的參數(shù)增量如風險厭惡系數(shù)-0.05。2.2 記憶系統(tǒng)的分層與檢索設計記憶是人格演化的燃料。EconAI中的記憶不能是簡單的聊天記錄堆砌必須是結構化的、可高效檢索的。一個分層記憶架構是必要的情景記憶存儲具體的交互事件和經(jīng)濟事件?！?023年Q3向智能體B借貸100元利率5%已按期歸還?！?這類記憶包含具體的時間、實體、動作和結果。語義記憶存儲從經(jīng)驗中歸納出的知識、信念和規(guī)則?！爸悄荏wB的信用良好。” “經(jīng)濟衰退時消費品價格會下跌但失業(yè)率會上升。” 這部分記憶通常由LLM對情景記憶進行定期總結、提煉而成。核心偏好這部分更接近人格是相對穩(wěn)定的長期特質(zhì)但也會緩慢變化。“我傾向于信任熟人?！?“我重視長期財務安全勝過短期消費享受?！睓z索機制至關重要。當智能體需要做決策時系統(tǒng)需要從海量記憶中快速找到最相關的信息。這通常結合向量檢索和元數(shù)據(jù)過濾。例如智能體面臨一個投資決策系統(tǒng)會用“投資”、“風險”、“科技股”等關鍵詞生成查詢向量從語義記憶中查找相關經(jīng)驗。用時間過濾器優(yōu)先檢索近期的記憶經(jīng)濟環(huán)境可能已變化。用情感元數(shù)據(jù)正/負面結果加權檢索結果負面記憶可能被賦予更高權重損失厭惡。2.3 經(jīng)濟環(huán)境引擎的構建要點經(jīng)濟環(huán)境是智能體活動的舞臺它需要為智能體提供感知輸入并對其行動給出反饋。一個簡化的經(jīng)濟引擎至少需要包含市場商品/勞動力/資產(chǎn)市場具備基本的供需模型和價格形成機制如拍賣、雙邊匹配。宏觀經(jīng)濟指標發(fā)布通貨膨脹率、基準利率、GDP增長率等這些指標應能根據(jù)所有智能體的集體行為總消費、總投資等動態(tài)計算或按一定規(guī)則演變。隨機事件引入可控的外部沖擊如技術突破、政策變化、自然災害等模擬經(jīng)濟環(huán)境的不確定性。環(huán)境引擎與智能體的接口需要標準化。每個仿真周期如一天、一周引擎向每個智能體廣播其可觀測的信息如市場價格、個人資產(chǎn)、宏觀經(jīng)濟新聞接收智能體的行動指令如購買X商品、提供Y服務、進行Z投資然后計算所有行動交互后的結果更新環(huán)境狀態(tài)并反饋給智能體。3. 智能體核心循環(huán)與決策過程拆解有了架構我們看單個智能體在一個周期內(nèi)如何“思考”和“行動”。這是一個典型的感知-思考-行動循環(huán)但注入了LLM和記憶。3.1 感知階段信息收集與上下文構建智能體并非全知全能。在每一步它只能從環(huán)境引擎獲取有限的信息例如個人狀態(tài)現(xiàn)金、庫存、負債、技能等級。市場信息有限幾種關注商品的價格、薪資水平。公共信息當前的宏觀經(jīng)濟指標頭條新聞。社交信息收到的交易請求、合作邀約等。系統(tǒng)需要將這些結構化數(shù)據(jù)與從記憶庫中檢索出的相關記憶見2.2組合成一個豐富的、文本化的決策上下文。這個上下文是一段精心設計的提示詞它告訴LLM“這是你當前的情況這是你過去的相關經(jīng)驗現(xiàn)在你需要做一個決定?！?.2 認知與決策階段LLM作為推理引擎這是最核心的環(huán)節(jié)。我們將構建好的上下文提交給LLM如GPT-4、Claude 3或開源模型Llama 3要求它扮演該智能體輸出決策。提示詞設計需要極高的技巧角色鎖定明確告知LLM“你是[智能體名稱]擁有以下人格特質(zhì)...你的長期目標是...”。任務明確清晰列出本周期可執(zhí)行的動作選項及其格式例如“動作BUY [商品] [數(shù)量]”、“動作PROPOSE_TRADE [對方] [出價]”。思維鏈要求強制要求LLM以“思考...”開頭先內(nèi)部推理引用相關記憶權衡利弊最后再輸出“動作...”。這能提升決策的合理性也便于我們調(diào)試。輸出規(guī)范化嚴格限制輸出格式方便后續(xù)系統(tǒng)解析。例如一個提示詞片段可能是你是Alex一個風險厭惡型但渴望提升社會地位的零售商。你當前的現(xiàn)金是1000元持有50單位谷物。過去三次經(jīng)濟衰退中谷物價格都先跌后漲來自你的語義記憶。昨天你剛和批發(fā)商Bob進行了一次不愉快的議價他顯得很急躁來自情景記憶。 當前市場谷物價格15元/單位木材價格30元/單位。宏觀經(jīng)濟新聞央行暗示可能加息。 請基于你的記憶、人格和目標決定本周期的行動。請先進行思考再輸出最終動作。 可選動作格式BUY [商品] [數(shù)量], SELL [商品] [數(shù)量], NEGOTIATE [對方] [商品] [目標價格], HOLD。 思考3.3 行動執(zhí)行與記憶更新階段LLM輸出規(guī)范化的動作后環(huán)境引擎會驗證并執(zhí)行該動作例如檢查現(xiàn)金是否足夠完成購買計算結果更新資產(chǎn)、可能觸發(fā)市場波動并將結果反饋給智能體。緊接著必須立即進行記憶更新。這是一個關鍵步驟決定了智能體能否學習。系統(tǒng)需要自動生成一條新的情景記憶“在時間T觀察到市場狀態(tài)S經(jīng)過思考T采取了行動A導致了結果R成功/失敗獲利XX?！?然后可以將這條新記憶與近期記憶一起定期如每10個周期觸發(fā)一次LLM進行記憶總結與提煉生成或更新語義記憶和核心偏好。例如如果智能體連續(xù)三次在價格低點買入谷物并在高點賣出成功LLM總結后可能會生成一條新的語義記憶“我似乎掌握了谷物價格季節(jié)性波動的規(guī)律并在實踐中驗證了低買高賣策略的有效性。” 同時其“風險厭惡”參數(shù)可能會略微下調(diào)“自信”參數(shù)可能會上升。4. 多智能體交互與社會網(wǎng)絡涌現(xiàn)單個智能體的學習固然有趣但EconAI的真正魅力在于多智能體互動中涌現(xiàn)出的復雜社會經(jīng)濟現(xiàn)象。這涉及到智能體間的通信、信任建立、合作與競爭。4.1 交互協(xié)議的設計智能體之間不能隨意“腦電波交流”。需要設計一套簡潔有效的交互協(xié)議定義交互的類型、格式和規(guī)則。常見的交互類型包括交易一方發(fā)起包含具體條款商品、數(shù)量、價格的要約另一方可以接受、拒絕或還盤。借貸涉及本金、利率、期限和抵押物需要記錄在案并影響信用記憶。信息交換智能體可以有選擇地分享自己的市場觀點或私有信息可能是真也可能是假這需要設計信譽機制。合作形成例如多個智能體可以組成“公司”共同完成一個需要多種技能的生產(chǎn)項目。所有這些交互在系統(tǒng)層面都體現(xiàn)為結構化消息的傳遞。LLM的角色是生成和理解這些消息的內(nèi)容。例如在談判中LLM需要根據(jù)當前上下文和人格是強硬還是隨和生成一段討價還價的自然語言文本同時系統(tǒng)會將其核心條款解析為結構化數(shù)據(jù)供引擎處理。4.2 信任與信譽模型的構建在動態(tài)環(huán)境中信任是降低交易成本的關鍵。一個簡單的信譽模型可以為每個智能體維護一個“信譽分”其更新規(guī)則基于歷史交互成功履行合約如按時還款、交貨達標增加雙方信譽分。違約嚴重降低違約方信譽分。信息真實性事后可驗證的信息如果為真則增加提供者信譽如果為假則降低。LLM在決策時可以將對方的信譽分作為重要輸入。一個信譽極低的智能體可能根本收不到交易邀約或者只能以極高利率借款這模擬了現(xiàn)實中的信用懲罰。4.3 宏觀現(xiàn)象的涌現(xiàn)觀察當數(shù)十上百個具備記憶和人格的智能體在一個動態(tài)經(jīng)濟環(huán)境中長期運行一些有趣的宏觀現(xiàn)象可能會自發(fā)涌現(xiàn)市場周期智能體的一致樂觀導致過度投資和資產(chǎn)泡沫隨后泡沫破裂進入衰退集體學習后又開始復蘇。社會分層由于初始條件、人格和運氣差異部分智能體積累更多財富和資源形成階層。行業(yè)興衰智能體根據(jù)利潤信號進入或退出某個行業(yè)導致該行業(yè)產(chǎn)能和價格的周期性波動。制度自發(fā)形成為了降低交易風險智能體們可能“約定俗成”某些規(guī)則或選舉出“仲裁者”這類似于原始制度的萌芽。觀察和量化這些涌現(xiàn)現(xiàn)象是EconAI項目最重要的研究價值所在。我們需要設計一系列觀測指標和可視化工具來跟蹤財富基尼系數(shù)、行業(yè)集中度、平均價格波動率等。5. 實操搭建從零構建一個簡化版EconAI理論聊了很多現(xiàn)在我們動手搭建一個極度簡化的版本以理解整個技術棧和流程。我們將使用Python作為主要語言。5.1 技術棧選型與環(huán)境準備編程語言Python 3.10。生態(tài)豐富是AI和科學計算的首選。LLM接口OpenAI API (GPT-4) 或開源模型本地部署如使用ollama運行Llama 3或vLLM部署Qwen系列。對于實驗API更方便對于大規(guī)模模擬需考慮本地模型的成本和可控性。向量數(shù)據(jù)庫用于存儲和檢索記憶。輕量級可選ChromaDB或FAISS。ChromaDB易于集成適合原型。開發(fā)框架雖然有很多新興的Agent框架如LangChain, AutoGen但為了深入理解底層機制我建議初期不用重型框架而是用基礎庫自建核心循環(huán)。這能避免“黑箱”讓你清楚每一個環(huán)節(jié)?？梢灾饕蕾噐equests(調(diào)用API)、numpy/pandas(數(shù)據(jù)處理)、logging(記錄) 等基礎庫。環(huán)境配置# 創(chuàng)建虛擬環(huán)境 python -m venv econai_env source econai_env/bin/activate # Linux/Mac # econai_env\Scripts\activate # Windows # 安裝核心庫 pip install openai chromadb numpy pandas # 如果使用本地LLM例如ollama # pip install ollama5.2 基礎數(shù)據(jù)結構定義我們首先定義幾個核心的Python類來表征系統(tǒng)的基本元素。import uuid from datetime import datetime from typing import Dict, List, Any, Optional from pydantic import BaseModel class MemoryItem(BaseModel): 單條記憶項 id: str type: str # episodic, semantic, preference content: str # 記憶的文本內(nèi)容 embedding: Optional[List[float]] None # 向量嵌入 metadata: Dict[str, Any] # 時間、關聯(lián)實體、情感標簽等 importance: float 1.0 # 記憶重要性權重 class Agent: 智能體類 def __init__(self, name: str, persona_desc: str): self.id str(uuid.uuid4()) self.name name self.persona_desc persona_desc self.persona_params self._initialize_persona(persona_desc) # 人格參數(shù)字典 self.memory_store: List[MemoryItem] [] # 簡化版實際應接向量庫 self.inventory: Dict[str, float] {cash: 1000.0} # 庫存/資產(chǎn) self.known_agents {} # 已知的其他智能體及信譽 def _initialize_persona(self, desc: str) - Dict[str, float]: 調(diào)用LLM根據(jù)描述初始化人格參數(shù) # 這里簡化處理實際需要調(diào)用LLM并解析輸出 # 示例返回一個參數(shù)字典 return { risk_aversion: 0.7, # 0-1, 越高越厭惡風險 consumption_propensity: 0.6, # 消費傾向 social_activity: 0.5, # 社交活躍度 learning_rate: 0.1, # 從經(jīng)驗中學習的速度 } class Market: 單一商品市場 def __init__(self, good_name: str): self.good_name good_name self.current_price 10.0 self.supply 100.0 self.demand 0.0 def update_price(self, net_transaction: float): 根據(jù)凈交易量更新價格非常簡單的線性模型 # net_transaction: 正表示凈買入負表示凈賣出 price_adjustment net_transaction * 0.01 self.current_price max(0.1, self.current_price price_adjustment)5.3 核心決策循環(huán)的實現(xiàn)這是智能體的“大腦”部分。我們實現(xiàn)一個make_decision方法。import openai # 假設已設置 openai.api_key class Agent(Agent): # 續(xù)接上面的Agent類 def retrieve_relevant_memories(self, query: str, top_k: int 5) - List[MemoryItem]: 從記憶中檢索最相關的幾條簡化版未實現(xiàn)真實向量檢索 # 在實際中這里應使用向量數(shù)據(jù)庫查詢 # 此處僅返回最近幾條記憶作為演示 return self.memory_store[-top_k:] if self.memory_store else [] def construct_context(self, market_info: Dict, other_agents_info: List) - str: 構建LLM決策上下文 # 1. 檢索相關記憶 query fmarket prices: {market_info}, my status: {self.inventory} relevant_mems self.retrieve_relevant_memories(query) memory_text \n.join([f- {m.content} for m in relevant_mems]) # 2. 構建上下文提示詞 context f 你是一個經(jīng)濟模擬中的智能體名叫{self.name}。 你的核心人格特征是{self.persona_desc}。 你當前的人格參數(shù)是{self.persona_params}。 【你當前的資產(chǎn)與狀態(tài)】 {self.inventory} 【你相關的過往記憶】 {memory_text} 【當前市場環(huán)境】 {market_info} 【可交互的其他智能體】 {other_agents_info} 【你可以執(zhí)行的動作】 1. BUY [商品名稱] [數(shù)量] - 以市場價格購買商品 2. SELL [商品名稱] [數(shù)量] - 以市場價格出售商品 3. HOLD - 什么也不做保持觀望 請仔細分析你的記憶、當前狀況和人格做出最符合你角色和利益的決策。 首先以“思考”開頭詳細寫出你的推理過程引用你的記憶和人格特質(zhì)。 最后以“動作”開頭輸出你的最終動作嚴格遵循上述動作格式。 return context def make_decision(self, market_info: Dict) - str: 調(diào)用LLM做出決策 context_prompt self.construct_context(market_info, []) try: response openai.ChatCompletion.create( modelgpt-3.5-turbo, # 實驗可用gpt-3.5-turbo研究建議用gpt-4 messages[{role: user, content: context_prompt}], temperature0.7, # 控制創(chuàng)造性決策時可稍低如0.3-0.5 max_tokens500 ) full_response response.choices[0].message.content # 解析出動作部分 lines full_response.split(\n) for line in lines: if line.startswith(動作): return line.strip() return 動作HOLD # 默認動作 except Exception as e: print(fLLM調(diào)用失敗: {e}) return 動作HOLD5.4 記憶的存儲、檢索與總結實現(xiàn)記憶系統(tǒng)是智能體學習的核心。我們需要實現(xiàn)一個與向量數(shù)據(jù)庫交互的MemorySystem類。import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer # 用于生成向量 class MemorySystem: def __init__(self, agent_id: str): self.agent_id agent_id self.client chromadb.Client(Settings(chroma_db_implduckdbparquet, persist_directory./chroma_db)) # 獲取或創(chuàng)建該智能體的記憶集合 self.collection self.client.get_or_create_collection(namefagent_memories_{agent_id}) self.embedder SentenceTransformer(all-MiniLM-L6-v2) # 輕量級嵌入模型 def add_memory(self, memory_item: MemoryItem): 添加一條記憶到向量數(shù)據(jù)庫 # 生成內(nèi)容向量 embedding self.embedder.encode(memory_item.content).tolist() memory_item.embedding embedding # 存儲到ChromaDB self.collection.add( documents[memory_item.content], metadatas[memory_item.metadata], embeddings[embedding], ids[memory_item.id] ) # 同時保留在本地列表供快速訪問可選 # self.memory_store.append(memory_item) def retrieve_memories(self, query_text: str, n_results: int 5, memory_type: str None) - List[MemoryItem]: 根據(jù)查詢文本檢索相關記憶 query_embedding self.embedder.encode(query_text).tolist() # 構建過濾條件 where_clause None if memory_type: where_clause {type: memory_type} results self.collection.query( query_embeddings[query_embedding], n_resultsn_results, wherewhere_clause ) retrieved_memories [] if results[documents]: for i in range(len(results[documents][0])): mem MemoryItem( idresults[ids][0][i], contentresults[documents][0][i], metadataresults[metadatas][0][i], typeresults[metadatas][0][i].get(type, episodic), embeddingresults[embeddings][0][i] if results[embeddings] else None ) retrieved_memories.append(mem) return retrieved_memories def summarize_memories(self, recent_memories: List[MemoryItem]) - Optional[str]: 定期調(diào)用LLM對近期記憶進行總結生成語義記憶 if not recent_memories: return None memory_text \n.join([f- {m.content} for m in recent_memories[-10:]]) # 總結最近10條 prompt f 以下是智能體{self.agent_id}在過去一段時間內(nèi)的經(jīng)歷 {memory_text} 請從這些具體經(jīng)歷中總結出1-3條普遍性的經(jīng)驗、教訓或更新你對世界的認知?？偨Y應簡潔、抽象適用于指導未來行動。 輸出格式為 總結1[你的總結] 總結2[你的總結] ... try: response openai.ChatCompletion.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.5, max_tokens300 ) summary response.choices[0].message.content # 將總結作為一條新的“語義記憶”存儲 semantic_memory MemoryItem( idstr(uuid.uuid4()), typesemantic, contentsummary, metadata{summary_time: datetime.now().isoformat(), source_memories: [m.id for m in recent_memories]} ) self.add_memory(semantic_memory) return summary except Exception as e: print(f記憶總結失敗: {e}) return None6. 系統(tǒng)集成、運行與觀測將各個模塊組合起來形成一個可以運行的模擬世界。6.1 主模擬循環(huán)搭建import time import random class Simulation: def __init__(self, num_agents5): self.agents: List[Agent] [] self.markets: Dict[str, Market] {grain: Market(grain), lumber: Market(lumber)} self.current_step 0 self.data_log [] # 用于記錄每一步的宏觀數(shù)據(jù) # 初始化智能體 personas [ 一個保守的農(nóng)民重視糧食安全, 一個激進的木材商人追求高利潤, 一個平衡的工匠注重穩(wěn)定收入, 一個年輕的投機者樂于冒險, 一個經(jīng)驗豐富的倉儲商善于把握時機 ] for i, p in enumerate(personas[:num_agents]): agent Agent(namefAgent_{i}, persona_descp) agent.memory_system MemorySystem(agent.id) # 為每個智能體掛載記憶系統(tǒng) self.agents.append(agent) def run_step(self): 運行一個模擬周期 self.current_step 1 print(f\n 模擬步數(shù) {self.current_step} ) # 1. 環(huán)境更新例如隨機事件 if random.random() 0.1: # 10%概率發(fā)生隨機事件 event random.choice([豐收, 干旱, 政策利好]) print(f隨機事件: {event}) # 這里可以調(diào)整市場參數(shù) if event 豐收: self.markets[grain].supply * 1.2 elif event 干旱: self.markets[grain].supply * 0.8 market_snapshot {name: market.current_price for name, market in self.markets.items()} # 2. 每個智能體決策并行動 total_buy_sell {good: 0.0 for good in self.markets} for agent in self.agents: # 獲取決策 action_str agent.make_decision(market_snapshot) print(f{agent.name}: {action_str}) # 解析并執(zhí)行動作這里簡化處理僅處理BUY/SELL try: if action_str.startswith(動作BUY): _, good, qty action_str.replace(動作BUY, ).strip().split() qty float(qty) good good.lower() cost self.markets[good].current_price * qty if agent.inventory[cash] cost: agent.inventory[cash] - cost agent.inventory[good] agent.inventory.get(good, 0) qty total_buy_sell[good] qty # 凈買入為正 # 記錄記憶 memory MemoryItem( idstr(uuid.uuid4()), typeepisodic, contentf在步驟{self.current_step}以{self.markets[good].current_price}的價格購買了{qty}單位{good}。, metadata{step: self.current_step, action: BUY, good: good, quantity: qty, price: self.markets[good].current_price} ) agent.memory_system.add_memory(memory) elif action_str.startswith(動作SELL): _, good, qty action_str.replace(動作SELL, ).strip().split() qty float(qty) good good.lower() if agent.inventory.get(good, 0) qty: revenue self.markets[good].current_price * qty agent.inventory[cash] revenue agent.inventory[good] - qty total_buy_sell[good] - qty # 凈賣出為負 # 記錄記憶... except Exception as e: print(f解析/執(zhí)行動作失敗: {action_str}, 錯誤: {e}) # 每隔一定步數(shù)觸發(fā)記憶總結 if self.current_step % 5 0: recent_mems agent.memory_system.retrieve_memories(query_textrecent, n_results10) summary agent.memory_system.summarize_memories(recent_mems) if summary: print(f{agent.name} 的記憶總結: {summary[:100]}...) # 3. 市場根據(jù)總交易量更新價格 for good, net in total_buy_sell.items(): self.markets[good].update_price(net) print(f{good}市場: 凈交易量{net:.2f}, 新價格{self.markets[good].current_price:.2f}) # 4. 記錄宏觀數(shù)據(jù) step_data { step: self.current_step, grain_price: self.markets[grain].current_price, lumber_price: self.markets[lumber].current_price, total_cash: sum([a.inventory[cash] for a in self.agents]), # ... 其他指標 } self.data_log.append(step_data) def run(self, steps20): 運行多步模擬 for _ in range(steps): self.run_step() time.sleep(0.5) # 方便觀察 # 運行模擬 if __name__ __main__: sim Simulation(num_agents3) sim.run(steps10)6.2 可視化與結果分析模擬運行后data_log里積累了時間序列數(shù)據(jù)。我們可以用matplotlib進行簡單的可視化觀察涌現(xiàn)現(xiàn)象。import pandas as pd import matplotlib.pyplot as plt # 假設simulation.data_log已經(jīng)存在 df pd.DataFrame(sim.data_log) fig, axes plt.subplots(2, 1, figsize(10, 8)) # 價格走勢 axes[0].plot(df[step], df[grain_price], labelGrain Price, markero) axes[0].plot(df[step], df[lumber_price], labelLumber Price, markers) axes[0].set_xlabel(Simulation Step) axes[0].set_ylabel(Price) axes[0].set_title(Market Price Dynamics) axes[0].legend() axes[0].grid(True, linestyle--, alpha0.7) # 總現(xiàn)金量簡單衡量經(jīng)濟活躍度 axes[1].plot(df[step], df[total_cash], labelTotal Cash in System, colorgreen, marker^) axes[1].set_xlabel(Simulation Step) axes[1].set_ylabel(Total Cash) axes[1].set_title(System Liquidity) axes[1].legend() axes[1].grid(True, linestyle--, alpha0.7) plt.tight_layout() plt.show()通過圖表你可以觀察價格是否出現(xiàn)波動周期總現(xiàn)金量如何變化如果設計了生產(chǎn)消費總現(xiàn)金可能不變但資產(chǎn)價值會變。更復雜的分析可以包括計算每個智能體的財富變化、交易網(wǎng)絡的演化等。7. 避坑指南與進階優(yōu)化方向在實踐EconAI這類項目時你會遇到許多預料之中和預料之外的挑戰(zhàn)。以下是一些關鍵的注意事項和進階思路。7.1 常見問題與調(diào)試技巧LLM調(diào)用成本與延遲這是最大的實踐瓶頸。頻繁調(diào)用GPT-4 API進行模擬成本會迅速攀升。技巧對于實驗大量使用gpt-3.5-turbo。將多個智能體的決策請求批量處理如果它們的環(huán)境上下文獨立。設置合理的max_tokens和temperature以控制輸出長度和穩(wěn)定性?？紤]使用開源模型本地部署雖然初期設置復雜但長期成本可控且隱私性好。智能體行為失控或趨同所有智能體可能做出相似的非理性決策或者行為完全脫離預設人格。調(diào)試首先詳細記錄每個智能體每個周期的完整提示詞和LLM響應。這是調(diào)試的黃金標準。檢查提示詞是否清晰鎖定了角色人格參數(shù)是否被有效嵌入到上下文中temperature參數(shù)是否設置得當太低導致死板太高導致隨機其次為決策引入少量隨機噪聲或個性化偏差防止趨同。記憶檢索不相關或效率低技巧優(yōu)化查詢生成。不要直接用原始觀察作為查詢而是讓LLM先根據(jù)當前決策任務生成一個更聚焦的查詢問題例如“根據(jù)當前市場我是否應該賣出谷物”再用這個問題去檢索記憶。調(diào)整向量模型對于經(jīng)濟文本可以嘗試用經(jīng)濟新聞或報告微調(diào)過的嵌入模型。模擬運行速度慢瓶頸分析主要瓶頸通常是LLM調(diào)用和向量檢索。對于向量檢索確保使用索引如FAISS的IVF索引。對于LLM如前所述考慮批量調(diào)用、緩存常見響應、或使用更小的模型。經(jīng)濟系統(tǒng)失衡市場價格飛漲或暴跌至不合理范圍或者某個資源被單一智能體壟斷。對策在環(huán)境引擎中引入穩(wěn)定機制如價格限幅、基礎貨幣注入/回收、反壟斷規(guī)則模擬稅收或補貼。這并非“作弊”而是模擬現(xiàn)實中的宏觀調(diào)控。7.2 性能優(yōu)化與擴展建議當基礎版本跑通后你可以從以下方向深化分層決策與快速路徑不是每個決策都需要勞駕LLM?？梢詾橹悄荏w設計一個快速決策系統(tǒng)對于常規(guī)、低風險決策如按習慣購買日用品使用基于規(guī)則的快速響應。只有遇到復雜、高風險或新情況時才觸發(fā)完整的LLM推理循環(huán)。這能大幅降低計算開銷。更復雜的環(huán)境與動作空間引入生產(chǎn)將原材料加工為商品、研發(fā)提升生產(chǎn)效率、金融借貸、債券、政府稅收、公共品等模塊。動作空間從簡單的買賣擴展到更復雜的合同簽訂、長期投資等。情感與社交關系的建模在記憶中增加情感標簽喜悅、憤怒、后悔并讓情感影響人格參數(shù)和決策。智能體之間可以建立“友誼”、“競爭”或“信任”關系這些關系會影響信息分享的真實性和交易條件。離線學習與策略改進讓智能體不僅從在線經(jīng)驗中學習還能進行“離線反思”。定期用一批歷史記憶訓練一個輕量級的策略網(wǎng)絡一個小型神經(jīng)網(wǎng)絡讓它學習在什么狀態(tài)下采取什么動作收益更高。這個策略網(wǎng)絡可以作為LLM決策的輔助或快速決策模塊。引入外部數(shù)據(jù)讓經(jīng)濟環(huán)境接收真實世界的新聞摘要或經(jīng)濟數(shù)據(jù)作為輸入使模擬環(huán)境與真實世界產(chǎn)生弱關聯(lián)觀察智能體如何應對“真實”的沖擊。EconAI是一個宏大的愿景我們構建的只是一個極其簡化的雛形。它的價值不在于預測真實經(jīng)濟而在于為我們提供一個可控的“數(shù)字實驗室”去檢驗關于人類經(jīng)濟行為、制度演化、復雜系統(tǒng)動力學的各種理論和猜想。每一步深入的實現(xiàn)都會讓你對智能體、LLM以及社會經(jīng)濟系統(tǒng)的復雜性有更深刻的理解。從這個玩具系統(tǒng)出發(fā)你可以選擇任何一個感興趣的方向深挖下去比如優(yōu)化記憶架構、設計更精巧的市場機制或是探索多智能體協(xié)作的涌現(xiàn)每一條路都充滿了挑戰(zhàn)和樂趣。 基于LLM的AI智能體在動態(tài)經(jīng)濟模擬中的架構設計與實現(xiàn)
1. 項目概述當AI智能體走進動態(tài)經(jīng)濟沙盤最近在AI和模擬經(jīng)濟交叉的圈子里一個叫“EconAI”的概念討論度挺高。簡單來說它試圖用大語言模型驅動的智能體在一個持續(xù)變化的經(jīng)濟環(huán)境里模擬具有“成長性”和“記憶力”的決策者。這聽起來有點像把《模擬人生》里的角色丟進一個動態(tài)的宏觀經(jīng)濟模型里讓他們不僅要賺錢糊口還要學習、適應甚至預測經(jīng)濟周期的波動。傳統(tǒng)的經(jīng)濟模型或游戲AI角色的行為模式往往是靜態(tài)的或者基于一套預設的、有限的規(guī)則。比如一個模擬中的“消費者”智能體其消費傾向可能是固定的或者只對價格做出簡單反應。但現(xiàn)實中的經(jīng)濟參與者不是這樣的我們的決策深受過去經(jīng)驗記憶、對未來的預期預測以及自身身份認知人設的影響。一個經(jīng)歷了經(jīng)濟蕭條的人其儲蓄和投資行為會與一直生活在繁榮期的人截然不同。EconAI瞄準的正是用LLM賦予智能體這種動態(tài)演化的“人設”和持續(xù)積累的“記憶”能力讓它們在模擬中更像真實的、會學習的個體。這項目適合誰呢如果你是對AI智能體架構、多智能體模擬、LLM應用落地或者計算經(jīng)濟學感興趣的研究者、開發(fā)者或學生這里面的設計思路和實現(xiàn)挑戰(zhàn)會很有嚼頭。它不是一個開箱即用的產(chǎn)品更像一個前沿的實驗框架探討如何將LLM的認知能力與復雜系統(tǒng)仿真結合起來。接下來我會結合當前Agent開發(fā)的熱點拆解EconAI可能的核心設計、實操難點以及我們?nèi)绾蝿邮执罱ㄒ粋€簡化版本。2. 核心架構設計記憶、人格與環(huán)境的三角博弈EconAI的吸引力在于它試圖協(xié)調(diào)三個動態(tài)變化的復雜系統(tǒng)智能體的內(nèi)部狀態(tài)人格與記憶、智能體間的社會交互以及外部的經(jīng)濟環(huán)境。要讓這個系統(tǒng)運轉起來而不是陷入混沌或僵化架構設計上必須解決幾個核心問題。2.1 動態(tài)人格演化的實現(xiàn)機制“動態(tài)人格”是讓智能體擺脫腳本的關鍵。這里的“人格”不是一個標簽而是一套可量化、可調(diào)整的決策參數(shù)和偏好集合。例如可以包括風險厭惡系數(shù)、消費傾向、學習能力、社交活躍度等。一個可行的實現(xiàn)機制是基于經(jīng)驗反饋的參數(shù)微調(diào)。智能體初始人格由LLM根據(jù)一段簡短的背景描述生成例如“你是一個初入職場、謹慎的年輕程序員”。LLM會輸出一組初始參數(shù)。之后每一次重大經(jīng)濟決策如是否進行一筆高風險投資的結果都會成為一個反饋信號。系統(tǒng)需要設計一個人格更新函數(shù)。這個函數(shù)不直接讓LLM修改參數(shù)而是將決策結果盈利/虧損、環(huán)境狀態(tài)市場繁榮/衰退和智能體自身的感受由LLM總結作為輸入生成對人格參數(shù)的調(diào)整量。注意避免讓LLM直接輸出“我的風險厭惡系數(shù)從0.7調(diào)整為0.65”。這會導致不可控的漂移。更好的做法是讓LLM以文本形式總結本次經(jīng)驗教訓例如“這次冒險投資成功讓我信心大增未來可能愿意承擔稍多一點風險”然后由一個確定的、可控的規(guī)則模塊將這段文本映射到一個小的參數(shù)增量如風險厭惡系數(shù)-0.05。2.2 記憶系統(tǒng)的分層與檢索設計記憶是人格演化的燃料。EconAI中的記憶不能是簡單的聊天記錄堆砌必須是結構化的、可高效檢索的。一個分層記憶架構是必要的情景記憶存儲具體的交互事件和經(jīng)濟事件?！?023年Q3向智能體B借貸100元利率5%已按期歸還?！?這類記憶包含具體的時間、實體、動作和結果。語義記憶存儲從經(jīng)驗中歸納出的知識、信念和規(guī)則?！爸悄荏wB的信用良好。” “經(jīng)濟衰退時消費品價格會下跌但失業(yè)率會上升。” 這部分記憶通常由LLM對情景記憶進行定期總結、提煉而成。核心偏好這部分更接近人格是相對穩(wěn)定的長期特質(zhì)但也會緩慢變化。“我傾向于信任熟人?！?“我重視長期財務安全勝過短期消費享受?！睓z索機制至關重要。當智能體需要做決策時系統(tǒng)需要從海量記憶中快速找到最相關的信息。這通常結合向量檢索和元數(shù)據(jù)過濾。例如智能體面臨一個投資決策系統(tǒng)會用“投資”、“風險”、“科技股”等關鍵詞生成查詢向量從語義記憶中查找相關經(jīng)驗。用時間過濾器優(yōu)先檢索近期的記憶經(jīng)濟環(huán)境可能已變化。用情感元數(shù)據(jù)正/負面結果加權檢索結果負面記憶可能被賦予更高權重損失厭惡。2.3 經(jīng)濟環(huán)境引擎的構建要點經(jīng)濟環(huán)境是智能體活動的舞臺它需要為智能體提供感知輸入并對其行動給出反饋。一個簡化的經(jīng)濟引擎至少需要包含市場商品/勞動力/資產(chǎn)市場具備基本的供需模型和價格形成機制如拍賣、雙邊匹配。宏觀經(jīng)濟指標發(fā)布通貨膨脹率、基準利率、GDP增長率等這些指標應能根據(jù)所有智能體的集體行為總消費、總投資等動態(tài)計算或按一定規(guī)則演變。隨機事件引入可控的外部沖擊如技術突破、政策變化、自然災害等模擬經(jīng)濟環(huán)境的不確定性。環(huán)境引擎與智能體的接口需要標準化。每個仿真周期如一天、一周引擎向每個智能體廣播其可觀測的信息如市場價格、個人資產(chǎn)、宏觀經(jīng)濟新聞接收智能體的行動指令如購買X商品、提供Y服務、進行Z投資然后計算所有行動交互后的結果更新環(huán)境狀態(tài)并反饋給智能體。3. 智能體核心循環(huán)與決策過程拆解有了架構我們看單個智能體在一個周期內(nèi)如何“思考”和“行動”。這是一個典型的感知-思考-行動循環(huán)但注入了LLM和記憶。3.1 感知階段信息收集與上下文構建智能體并非全知全能。在每一步它只能從環(huán)境引擎獲取有限的信息例如個人狀態(tài)現(xiàn)金、庫存、負債、技能等級。市場信息有限幾種關注商品的價格、薪資水平。公共信息當前的宏觀經(jīng)濟指標頭條新聞。社交信息收到的交易請求、合作邀約等。系統(tǒng)需要將這些結構化數(shù)據(jù)與從記憶庫中檢索出的相關記憶見2.2組合成一個豐富的、文本化的決策上下文。這個上下文是一段精心設計的提示詞它告訴LLM“這是你當前的情況這是你過去的相關經(jīng)驗現(xiàn)在你需要做一個決定?！?.2 認知與決策階段LLM作為推理引擎這是最核心的環(huán)節(jié)。我們將構建好的上下文提交給LLM如GPT-4、Claude 3或開源模型Llama 3要求它扮演該智能體輸出決策。提示詞設計需要極高的技巧角色鎖定明確告知LLM“你是[智能體名稱]擁有以下人格特質(zhì)...你的長期目標是...”。任務明確清晰列出本周期可執(zhí)行的動作選項及其格式例如“動作BUY [商品] [數(shù)量]”、“動作PROPOSE_TRADE [對方] [出價]”。思維鏈要求強制要求LLM以“思考...”開頭先內(nèi)部推理引用相關記憶權衡利弊最后再輸出“動作...”。這能提升決策的合理性也便于我們調(diào)試。輸出規(guī)范化嚴格限制輸出格式方便后續(xù)系統(tǒng)解析。例如一個提示詞片段可能是你是Alex一個風險厭惡型但渴望提升社會地位的零售商。你當前的現(xiàn)金是1000元持有50單位谷物。過去三次經(jīng)濟衰退中谷物價格都先跌后漲來自你的語義記憶。昨天你剛和批發(fā)商Bob進行了一次不愉快的議價他顯得很急躁來自情景記憶。 當前市場谷物價格15元/單位木材價格30元/單位。宏觀經(jīng)濟新聞央行暗示可能加息。 請基于你的記憶、人格和目標決定本周期的行動。請先進行思考再輸出最終動作。 可選動作格式BUY [商品] [數(shù)量], SELL [商品] [數(shù)量], NEGOTIATE [對方] [商品] [目標價格], HOLD。 思考3.3 行動執(zhí)行與記憶更新階段LLM輸出規(guī)范化的動作后環(huán)境引擎會驗證并執(zhí)行該動作例如檢查現(xiàn)金是否足夠完成購買計算結果更新資產(chǎn)、可能觸發(fā)市場波動并將結果反饋給智能體。緊接著必須立即進行記憶更新。這是一個關鍵步驟決定了智能體能否學習。系統(tǒng)需要自動生成一條新的情景記憶“在時間T觀察到市場狀態(tài)S經(jīng)過思考T采取了行動A導致了結果R成功/失敗獲利XX?！?然后可以將這條新記憶與近期記憶一起定期如每10個周期觸發(fā)一次LLM進行記憶總結與提煉生成或更新語義記憶和核心偏好。例如如果智能體連續(xù)三次在價格低點買入谷物并在高點賣出成功LLM總結后可能會生成一條新的語義記憶“我似乎掌握了谷物價格季節(jié)性波動的規(guī)律并在實踐中驗證了低買高賣策略的有效性。” 同時其“風險厭惡”參數(shù)可能會略微下調(diào)“自信”參數(shù)可能會上升。4. 多智能體交互與社會網(wǎng)絡涌現(xiàn)單個智能體的學習固然有趣但EconAI的真正魅力在于多智能體互動中涌現(xiàn)出的復雜社會經(jīng)濟現(xiàn)象。這涉及到智能體間的通信、信任建立、合作與競爭。4.1 交互協(xié)議的設計智能體之間不能隨意“腦電波交流”。需要設計一套簡潔有效的交互協(xié)議定義交互的類型、格式和規(guī)則。常見的交互類型包括交易一方發(fā)起包含具體條款商品、數(shù)量、價格的要約另一方可以接受、拒絕或還盤。借貸涉及本金、利率、期限和抵押物需要記錄在案并影響信用記憶。信息交換智能體可以有選擇地分享自己的市場觀點或私有信息可能是真也可能是假這需要設計信譽機制。合作形成例如多個智能體可以組成“公司”共同完成一個需要多種技能的生產(chǎn)項目。所有這些交互在系統(tǒng)層面都體現(xiàn)為結構化消息的傳遞。LLM的角色是生成和理解這些消息的內(nèi)容。例如在談判中LLM需要根據(jù)當前上下文和人格是強硬還是隨和生成一段討價還價的自然語言文本同時系統(tǒng)會將其核心條款解析為結構化數(shù)據(jù)供引擎處理。4.2 信任與信譽模型的構建在動態(tài)環(huán)境中信任是降低交易成本的關鍵。一個簡單的信譽模型可以為每個智能體維護一個“信譽分”其更新規(guī)則基于歷史交互成功履行合約如按時還款、交貨達標增加雙方信譽分。違約嚴重降低違約方信譽分。信息真實性事后可驗證的信息如果為真則增加提供者信譽如果為假則降低。LLM在決策時可以將對方的信譽分作為重要輸入。一個信譽極低的智能體可能根本收不到交易邀約或者只能以極高利率借款這模擬了現(xiàn)實中的信用懲罰。4.3 宏觀現(xiàn)象的涌現(xiàn)觀察當數(shù)十上百個具備記憶和人格的智能體在一個動態(tài)經(jīng)濟環(huán)境中長期運行一些有趣的宏觀現(xiàn)象可能會自發(fā)涌現(xiàn)市場周期智能體的一致樂觀導致過度投資和資產(chǎn)泡沫隨后泡沫破裂進入衰退集體學習后又開始復蘇。社會分層由于初始條件、人格和運氣差異部分智能體積累更多財富和資源形成階層。行業(yè)興衰智能體根據(jù)利潤信號進入或退出某個行業(yè)導致該行業(yè)產(chǎn)能和價格的周期性波動。制度自發(fā)形成為了降低交易風險智能體們可能“約定俗成”某些規(guī)則或選舉出“仲裁者”這類似于原始制度的萌芽。觀察和量化這些涌現(xiàn)現(xiàn)象是EconAI項目最重要的研究價值所在。我們需要設計一系列觀測指標和可視化工具來跟蹤財富基尼系數(shù)、行業(yè)集中度、平均價格波動率等。5. 實操搭建從零構建一個簡化版EconAI理論聊了很多現(xiàn)在我們動手搭建一個極度簡化的版本以理解整個技術棧和流程。我們將使用Python作為主要語言。5.1 技術棧選型與環(huán)境準備編程語言Python 3.10。生態(tài)豐富是AI和科學計算的首選。LLM接口OpenAI API (GPT-4) 或開源模型本地部署如使用ollama運行Llama 3或vLLM部署Qwen系列。對于實驗API更方便對于大規(guī)模模擬需考慮本地模型的成本和可控性。向量數(shù)據(jù)庫用于存儲和檢索記憶。輕量級可選ChromaDB或FAISS。ChromaDB易于集成適合原型。開發(fā)框架雖然有很多新興的Agent框架如LangChain, AutoGen但為了深入理解底層機制我建議初期不用重型框架而是用基礎庫自建核心循環(huán)。這能避免“黑箱”讓你清楚每一個環(huán)節(jié)?？梢灾饕蕾噐equests(調(diào)用API)、numpy/pandas(數(shù)據(jù)處理)、logging(記錄) 等基礎庫。環(huán)境配置# 創(chuàng)建虛擬環(huán)境 python -m venv econai_env source econai_env/bin/activate # Linux/Mac # econai_env\Scripts\activate # Windows # 安裝核心庫 pip install openai chromadb numpy pandas # 如果使用本地LLM例如ollama # pip install ollama5.2 基礎數(shù)據(jù)結構定義我們首先定義幾個核心的Python類來表征系統(tǒng)的基本元素。import uuid from datetime import datetime from typing import Dict, List, Any, Optional from pydantic import BaseModel class MemoryItem(BaseModel): 單條記憶項 id: str type: str # episodic, semantic, preference content: str # 記憶的文本內(nèi)容 embedding: Optional[List[float]] None # 向量嵌入 metadata: Dict[str, Any] # 時間、關聯(lián)實體、情感標簽等 importance: float 1.0 # 記憶重要性權重 class Agent: 智能體類 def __init__(self, name: str, persona_desc: str): self.id str(uuid.uuid4()) self.name name self.persona_desc persona_desc self.persona_params self._initialize_persona(persona_desc) # 人格參數(shù)字典 self.memory_store: List[MemoryItem] [] # 簡化版實際應接向量庫 self.inventory: Dict[str, float] {cash: 1000.0} # 庫存/資產(chǎn) self.known_agents {} # 已知的其他智能體及信譽 def _initialize_persona(self, desc: str) - Dict[str, float]: 調(diào)用LLM根據(jù)描述初始化人格參數(shù) # 這里簡化處理實際需要調(diào)用LLM并解析輸出 # 示例返回一個參數(shù)字典 return { risk_aversion: 0.7, # 0-1, 越高越厭惡風險 consumption_propensity: 0.6, # 消費傾向 social_activity: 0.5, # 社交活躍度 learning_rate: 0.1, # 從經(jīng)驗中學習的速度 } class Market: 單一商品市場 def __init__(self, good_name: str): self.good_name good_name self.current_price 10.0 self.supply 100.0 self.demand 0.0 def update_price(self, net_transaction: float): 根據(jù)凈交易量更新價格非常簡單的線性模型 # net_transaction: 正表示凈買入負表示凈賣出 price_adjustment net_transaction * 0.01 self.current_price max(0.1, self.current_price price_adjustment)5.3 核心決策循環(huán)的實現(xiàn)這是智能體的“大腦”部分。我們實現(xiàn)一個make_decision方法。import openai # 假設已設置 openai.api_key class Agent(Agent): # 續(xù)接上面的Agent類 def retrieve_relevant_memories(self, query: str, top_k: int 5) - List[MemoryItem]: 從記憶中檢索最相關的幾條簡化版未實現(xiàn)真實向量檢索 # 在實際中這里應使用向量數(shù)據(jù)庫查詢 # 此處僅返回最近幾條記憶作為演示 return self.memory_store[-top_k:] if self.memory_store else [] def construct_context(self, market_info: Dict, other_agents_info: List) - str: 構建LLM決策上下文 # 1. 檢索相關記憶 query fmarket prices: {market_info}, my status: {self.inventory} relevant_mems self.retrieve_relevant_memories(query) memory_text \n.join([f- {m.content} for m in relevant_mems]) # 2. 構建上下文提示詞 context f 你是一個經(jīng)濟模擬中的智能體名叫{self.name}。 你的核心人格特征是{self.persona_desc}。 你當前的人格參數(shù)是{self.persona_params}。 【你當前的資產(chǎn)與狀態(tài)】 {self.inventory} 【你相關的過往記憶】 {memory_text} 【當前市場環(huán)境】 {market_info} 【可交互的其他智能體】 {other_agents_info} 【你可以執(zhí)行的動作】 1. BUY [商品名稱] [數(shù)量] - 以市場價格購買商品 2. SELL [商品名稱] [數(shù)量] - 以市場價格出售商品 3. HOLD - 什么也不做保持觀望 請仔細分析你的記憶、當前狀況和人格做出最符合你角色和利益的決策。 首先以“思考”開頭詳細寫出你的推理過程引用你的記憶和人格特質(zhì)。 最后以“動作”開頭輸出你的最終動作嚴格遵循上述動作格式。 return context def make_decision(self, market_info: Dict) - str: 調(diào)用LLM做出決策 context_prompt self.construct_context(market_info, []) try: response openai.ChatCompletion.create( modelgpt-3.5-turbo, # 實驗可用gpt-3.5-turbo研究建議用gpt-4 messages[{role: user, content: context_prompt}], temperature0.7, # 控制創(chuàng)造性決策時可稍低如0.3-0.5 max_tokens500 ) full_response response.choices[0].message.content # 解析出動作部分 lines full_response.split(\n) for line in lines: if line.startswith(動作): return line.strip() return 動作HOLD # 默認動作 except Exception as e: print(fLLM調(diào)用失敗: {e}) return 動作HOLD5.4 記憶的存儲、檢索與總結實現(xiàn)記憶系統(tǒng)是智能體學習的核心。我們需要實現(xiàn)一個與向量數(shù)據(jù)庫交互的MemorySystem類。import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer # 用于生成向量 class MemorySystem: def __init__(self, agent_id: str): self.agent_id agent_id self.client chromadb.Client(Settings(chroma_db_implduckdbparquet, persist_directory./chroma_db)) # 獲取或創(chuàng)建該智能體的記憶集合 self.collection self.client.get_or_create_collection(namefagent_memories_{agent_id}) self.embedder SentenceTransformer(all-MiniLM-L6-v2) # 輕量級嵌入模型 def add_memory(self, memory_item: MemoryItem): 添加一條記憶到向量數(shù)據(jù)庫 # 生成內(nèi)容向量 embedding self.embedder.encode(memory_item.content).tolist() memory_item.embedding embedding # 存儲到ChromaDB self.collection.add( documents[memory_item.content], metadatas[memory_item.metadata], embeddings[embedding], ids[memory_item.id] ) # 同時保留在本地列表供快速訪問可選 # self.memory_store.append(memory_item) def retrieve_memories(self, query_text: str, n_results: int 5, memory_type: str None) - List[MemoryItem]: 根據(jù)查詢文本檢索相關記憶 query_embedding self.embedder.encode(query_text).tolist() # 構建過濾條件 where_clause None if memory_type: where_clause {type: memory_type} results self.collection.query( query_embeddings[query_embedding], n_resultsn_results, wherewhere_clause ) retrieved_memories [] if results[documents]: for i in range(len(results[documents][0])): mem MemoryItem( idresults[ids][0][i], contentresults[documents][0][i], metadataresults[metadatas][0][i], typeresults[metadatas][0][i].get(type, episodic), embeddingresults[embeddings][0][i] if results[embeddings] else None ) retrieved_memories.append(mem) return retrieved_memories def summarize_memories(self, recent_memories: List[MemoryItem]) - Optional[str]: 定期調(diào)用LLM對近期記憶進行總結生成語義記憶 if not recent_memories: return None memory_text \n.join([f- {m.content} for m in recent_memories[-10:]]) # 總結最近10條 prompt f 以下是智能體{self.agent_id}在過去一段時間內(nèi)的經(jīng)歷 {memory_text} 請從這些具體經(jīng)歷中總結出1-3條普遍性的經(jīng)驗、教訓或更新你對世界的認知?？偨Y應簡潔、抽象適用于指導未來行動。 輸出格式為 總結1[你的總結] 總結2[你的總結] ... try: response openai.ChatCompletion.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.5, max_tokens300 ) summary response.choices[0].message.content # 將總結作為一條新的“語義記憶”存儲 semantic_memory MemoryItem( idstr(uuid.uuid4()), typesemantic, contentsummary, metadata{summary_time: datetime.now().isoformat(), source_memories: [m.id for m in recent_memories]} ) self.add_memory(semantic_memory) return summary except Exception as e: print(f記憶總結失敗: {e}) return None6. 系統(tǒng)集成、運行與觀測將各個模塊組合起來形成一個可以運行的模擬世界。6.1 主模擬循環(huán)搭建import time import random class Simulation: def __init__(self, num_agents5): self.agents: List[Agent] [] self.markets: Dict[str, Market] {grain: Market(grain), lumber: Market(lumber)} self.current_step 0 self.data_log [] # 用于記錄每一步的宏觀數(shù)據(jù) # 初始化智能體 personas [ 一個保守的農(nóng)民重視糧食安全, 一個激進的木材商人追求高利潤, 一個平衡的工匠注重穩(wěn)定收入, 一個年輕的投機者樂于冒險, 一個經(jīng)驗豐富的倉儲商善于把握時機 ] for i, p in enumerate(personas[:num_agents]): agent Agent(namefAgent_{i}, persona_descp) agent.memory_system MemorySystem(agent.id) # 為每個智能體掛載記憶系統(tǒng) self.agents.append(agent) def run_step(self): 運行一個模擬周期 self.current_step 1 print(f\n 模擬步數(shù) {self.current_step} ) # 1. 環(huán)境更新例如隨機事件 if random.random() 0.1: # 10%概率發(fā)生隨機事件 event random.choice([豐收, 干旱, 政策利好]) print(f隨機事件: {event}) # 這里可以調(diào)整市場參數(shù) if event 豐收: self.markets[grain].supply * 1.2 elif event 干旱: self.markets[grain].supply * 0.8 market_snapshot {name: market.current_price for name, market in self.markets.items()} # 2. 每個智能體決策并行動 total_buy_sell {good: 0.0 for good in self.markets} for agent in self.agents: # 獲取決策 action_str agent.make_decision(market_snapshot) print(f{agent.name}: {action_str}) # 解析并執(zhí)行動作這里簡化處理僅處理BUY/SELL try: if action_str.startswith(動作BUY): _, good, qty action_str.replace(動作BUY, ).strip().split() qty float(qty) good good.lower() cost self.markets[good].current_price * qty if agent.inventory[cash] cost: agent.inventory[cash] - cost agent.inventory[good] agent.inventory.get(good, 0) qty total_buy_sell[good] qty # 凈買入為正 # 記錄記憶 memory MemoryItem( idstr(uuid.uuid4()), typeepisodic, contentf在步驟{self.current_step}以{self.markets[good].current_price}的價格購買了{qty}單位{good}。, metadata{step: self.current_step, action: BUY, good: good, quantity: qty, price: self.markets[good].current_price} ) agent.memory_system.add_memory(memory) elif action_str.startswith(動作SELL): _, good, qty action_str.replace(動作SELL, ).strip().split() qty float(qty) good good.lower() if agent.inventory.get(good, 0) qty: revenue self.markets[good].current_price * qty agent.inventory[cash] revenue agent.inventory[good] - qty total_buy_sell[good] - qty # 凈賣出為負 # 記錄記憶... except Exception as e: print(f解析/執(zhí)行動作失敗: {action_str}, 錯誤: {e}) # 每隔一定步數(shù)觸發(fā)記憶總結 if self.current_step % 5 0: recent_mems agent.memory_system.retrieve_memories(query_textrecent, n_results10) summary agent.memory_system.summarize_memories(recent_mems) if summary: print(f{agent.name} 的記憶總結: {summary[:100]}...) # 3. 市場根據(jù)總交易量更新價格 for good, net in total_buy_sell.items(): self.markets[good].update_price(net) print(f{good}市場: 凈交易量{net:.2f}, 新價格{self.markets[good].current_price:.2f}) # 4. 記錄宏觀數(shù)據(jù) step_data { step: self.current_step, grain_price: self.markets[grain].current_price, lumber_price: self.markets[lumber].current_price, total_cash: sum([a.inventory[cash] for a in self.agents]), # ... 其他指標 } self.data_log.append(step_data) def run(self, steps20): 運行多步模擬 for _ in range(steps): self.run_step() time.sleep(0.5) # 方便觀察 # 運行模擬 if __name__ __main__: sim Simulation(num_agents3) sim.run(steps10)6.2 可視化與結果分析模擬運行后data_log里積累了時間序列數(shù)據(jù)。我們可以用matplotlib進行簡單的可視化觀察涌現(xiàn)現(xiàn)象。import pandas as pd import matplotlib.pyplot as plt # 假設simulation.data_log已經(jīng)存在 df pd.DataFrame(sim.data_log) fig, axes plt.subplots(2, 1, figsize(10, 8)) # 價格走勢 axes[0].plot(df[step], df[grain_price], labelGrain Price, markero) axes[0].plot(df[step], df[lumber_price], labelLumber Price, markers) axes[0].set_xlabel(Simulation Step) axes[0].set_ylabel(Price) axes[0].set_title(Market Price Dynamics) axes[0].legend() axes[0].grid(True, linestyle--, alpha0.7) # 總現(xiàn)金量簡單衡量經(jīng)濟活躍度 axes[1].plot(df[step], df[total_cash], labelTotal Cash in System, colorgreen, marker^) axes[1].set_xlabel(Simulation Step) axes[1].set_ylabel(Total Cash) axes[1].set_title(System Liquidity) axes[1].legend() axes[1].grid(True, linestyle--, alpha0.7) plt.tight_layout() plt.show()通過圖表你可以觀察價格是否出現(xiàn)波動周期總現(xiàn)金量如何變化如果設計了生產(chǎn)消費總現(xiàn)金可能不變但資產(chǎn)價值會變。更復雜的分析可以包括計算每個智能體的財富變化、交易網(wǎng)絡的演化等。7. 避坑指南與進階優(yōu)化方向在實踐EconAI這類項目時你會遇到許多預料之中和預料之外的挑戰(zhàn)。以下是一些關鍵的注意事項和進階思路。7.1 常見問題與調(diào)試技巧LLM調(diào)用成本與延遲這是最大的實踐瓶頸。頻繁調(diào)用GPT-4 API進行模擬成本會迅速攀升。技巧對于實驗大量使用gpt-3.5-turbo。將多個智能體的決策請求批量處理如果它們的環(huán)境上下文獨立。設置合理的max_tokens和temperature以控制輸出長度和穩(wěn)定性?？紤]使用開源模型本地部署雖然初期設置復雜但長期成本可控且隱私性好。智能體行為失控或趨同所有智能體可能做出相似的非理性決策或者行為完全脫離預設人格。調(diào)試首先詳細記錄每個智能體每個周期的完整提示詞和LLM響應。這是調(diào)試的黃金標準。檢查提示詞是否清晰鎖定了角色人格參數(shù)是否被有效嵌入到上下文中temperature參數(shù)是否設置得當太低導致死板太高導致隨機其次為決策引入少量隨機噪聲或個性化偏差防止趨同。記憶檢索不相關或效率低技巧優(yōu)化查詢生成。不要直接用原始觀察作為查詢而是讓LLM先根據(jù)當前決策任務生成一個更聚焦的查詢問題例如“根據(jù)當前市場我是否應該賣出谷物”再用這個問題去檢索記憶。調(diào)整向量模型對于經(jīng)濟文本可以嘗試用經(jīng)濟新聞或報告微調(diào)過的嵌入模型。模擬運行速度慢瓶頸分析主要瓶頸通常是LLM調(diào)用和向量檢索。對于向量檢索確保使用索引如FAISS的IVF索引。對于LLM如前所述考慮批量調(diào)用、緩存常見響應、或使用更小的模型。經(jīng)濟系統(tǒng)失衡市場價格飛漲或暴跌至不合理范圍或者某個資源被單一智能體壟斷。對策在環(huán)境引擎中引入穩(wěn)定機制如價格限幅、基礎貨幣注入/回收、反壟斷規(guī)則模擬稅收或補貼。這并非“作弊”而是模擬現(xiàn)實中的宏觀調(diào)控。7.2 性能優(yōu)化與擴展建議當基礎版本跑通后你可以從以下方向深化分層決策與快速路徑不是每個決策都需要勞駕LLM?？梢詾橹悄荏w設計一個快速決策系統(tǒng)對于常規(guī)、低風險決策如按習慣購買日用品使用基于規(guī)則的快速響應。只有遇到復雜、高風險或新情況時才觸發(fā)完整的LLM推理循環(huán)。這能大幅降低計算開銷。更復雜的環(huán)境與動作空間引入生產(chǎn)將原材料加工為商品、研發(fā)提升生產(chǎn)效率、金融借貸、債券、政府稅收、公共品等模塊。動作空間從簡單的買賣擴展到更復雜的合同簽訂、長期投資等。情感與社交關系的建模在記憶中增加情感標簽喜悅、憤怒、后悔并讓情感影響人格參數(shù)和決策。智能體之間可以建立“友誼”、“競爭”或“信任”關系這些關系會影響信息分享的真實性和交易條件。離線學習與策略改進讓智能體不僅從在線經(jīng)驗中學習還能進行“離線反思”。定期用一批歷史記憶訓練一個輕量級的策略網(wǎng)絡一個小型神經(jīng)網(wǎng)絡讓它學習在什么狀態(tài)下采取什么動作收益更高。這個策略網(wǎng)絡可以作為LLM決策的輔助或快速決策模塊。引入外部數(shù)據(jù)讓經(jīng)濟環(huán)境接收真實世界的新聞摘要或經(jīng)濟數(shù)據(jù)作為輸入使模擬環(huán)境與真實世界產(chǎn)生弱關聯(lián)觀察智能體如何應對“真實”的沖擊。EconAI是一個宏大的愿景我們構建的只是一個極其簡化的雛形。它的價值不在于預測真實經(jīng)濟而在于為我們提供一個可控的“數(shù)字實驗室”去檢驗關于人類經(jīng)濟行為、制度演化、復雜系統(tǒng)動力學的各種理論和猜想。每一步深入的實現(xiàn)都會讓你對智能體、LLM以及社會經(jīng)濟系統(tǒng)的復雜性有更深刻的理解。從這個玩具系統(tǒng)出發(fā)你可以選擇任何一個感興趣的方向深挖下去比如優(yōu)化記憶架構、設計更精巧的市場機制或是探索多智能體協(xié)作的涌現(xiàn)每一條路都充滿了挑戰(zhàn)和樂趣。