他是模型，是 AI 能力的核心。 他是語言模型，其核心能力在于語言能力。 他是大語言模型，與傳統模型相比，它最大的特點就是“大”。 AI應用，就是以 LLM 為核心的各種應用 從API的角度理解GPT模型，它最核心的參數就是輸入一個或多個字符串，然后，大模型輸出一個字符串 與傳統的應用開發不同的是，這個 API 并非是傳統應用開發中按照特定預期處理的結果。 使用傳統的 API，我們需要關注的是接口文檔，而想要發揮 LLM 的威力，我們需要對大模型有一定的了解 一次添加一個詞 下一個詞是什么 token 溫度（Temperature）：表示隨機性強弱的概念 Embedding ：在大模型內部處理的是向量，，Embedding 是一種將高維數據（如文本、圖像、視頻等）轉換為低維向量表示的技術。這種技術在自然語言處理（NLP）、計算機視覺等領域有著廣泛的應用。Embedding 的核心思想是將離散數據映射到連續的向量空間，使得相似的數據點在向量空間中的距離較近，而不相似的數據點則距離較遠。 提示詞 = 定義角色 + 背景信息 + 任務目標 + 輸出要求 要求：大模型處理復雜任務場景的能力 原因：Agent背后的技術能讓大模型推斷下一步行為，利用大模型的推理能力，依賴于promot工程 起源：Natural Language Processing（NLP）:如果給予 AI 適當的引導，它能更準確地理解我們的意圖，響應我們的指令 零樣本提示（Zero-Shot Prompting）:適合簡單的任務。比如，一些簡單查詢就可以使用零樣本提示。我們需要做的就是調整提示詞 少樣本提示（Few-Shot Prompting）:適合復雜的任務。比如，我們需要讓 AI 回答一個問題，我們需要提供一些例子，讓 AI 學習這些例子，然后再回答問題。 思維鏈提示（Chain-of-Thought Prompting）:思維鏈提示給出的答案則是帶有完整的思考過程，是一個“慢下來”的答案，于是得到了一個正確的答案 ReAct 框架（Reasoning + Acting） :推理 + 行動=大模型為了完成一個大目標，需要不斷地做一些任務。每個任務都會經歷思考（Thought）、行動（Action）、觀察（Observation）三個階段。 Text Generation：生成和處理文本 Embeddings：文本轉向量 Speech to Text：語音轉文本 Image Generation：生成圖像 Vision：處理圖像輸入 SSE 是服務器發送事件（Server-Sent Event），它是一種服務器推送技術，客戶端通過 HTTP 連接接收來自服務器的自動更新 它描述了服務器如何在建立初始客戶端連接后向客戶端發起數據傳輸。 為啥不用WebSocket SSE 的技術特點契合流式應答的需求:客戶端與大模型的交互是一次性的，每產生一個 token，服務端就可以給客戶端推送一次，當生成內容結束時，斷掉連接，無需考慮客戶端的存活情況 如果采用 WebSocket 的話，服務端就需要維護連接，像 OpenAI 這樣的服務體量，維護連接就會造成很大的服務器壓力，而且，在生成內容場景下，也沒有向服務端進一步發送內容，WebSocket 的雙向通信在這里也是多余的 SSE 這項技術而言，它存在已經很長時間了，2004 年就有人提出,大模型才流行起來 ChatModel：整個框架的核心，根據輸入的內容生成輸出 PromptTemplate： 負責處理輸入，有效拆分開發者提示詞和用戶提示詞 OutputParser：負責處理輸出，許多輸出解析器里包含了格式指令 from langchain_openai import ChatOpenAI
from langchain_core. messages import HumanMessage
from langchain_core. chat_history import BaseChatMessageHistory
from langchain_core. prompts import ChatPromptTemplate
from langchain_core. prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core. runnables import RunnableWithMessageHistory
api_key = "你的密鑰"
chat_model = ChatOpenAI( model= "gpt-4o-mini" , api_key= api_key)
store = { }
def get_session_history ( session_id: str ) - > BaseChatMessageHistory: if session_id not in store: store[ session_id] = InMemoryChatMessageHistory( ) return store[ session_id]
prompt = ChatPromptTemplate. from_messages( [ ( "system" , "你現在扮演程序員的角色，可以直接生成代碼" , ) , MessagesPlaceholder( variable_name= "messages" ) , ]
)
with_message_history = RunnableWithMessageHistory( prompt | chat_model, get_session_history
)
config = { "configurable" : { "session_id" : "chatLLMCode" } }
while True : user_input = input ( "You:> " ) if user_input. lower( ) == 'exit' : break stream = with_message_history. stream( { "messages" : [ HumanMessage( content= user_input) ] } , config= config) for chunk in stream: print ( chunk. content, end= '' , flush= True ) print ( ) 檢索增強生成（Retrieval Augmented Generation，RAG）是一種結合了檢索和生成的技術，它可以在生成文本時，利用外部的知識庫來增強生成的內容。 檢索增強生成:在本地檢索到相關的內容，把它增強到提示詞里，然后再去做內容生成 模型微調：使用業務信息對已經訓練好的模型進行微調 RAG：在上下文中帶有業務信息，讓大模型據此進行整合 Embeddings：Embeddings 是一種將高維數據（如文本、圖像、視頻等）轉換為低維向量表示的技術。這種技術在自然語言處理（NLP）、計算機視覺等領域有著廣泛的應用。Embeddings 的核心思想是將離散數據映射到連續的向量空間，使得相似的數據點在向量空間中的距離較近，而不相似的數據點則距離較遠。 VectorStore：VectorStore 是一種用于存儲和檢索向量數據的技術。它可以將高維向量數據存儲在一個向量空間中，以便快速查找相似的向量。VectorStore 的核心思想是將高維數據映射到低維向量空間，以便進行高效的相似度搜索。 相似度搜索（Similarity Search）：相似度搜索是一種用于查找與給定向量最相似的向量的技術。它可以將給定的向量與索引中的向量進行比較，以便找到最相似的向量。相似度搜索的核心思想是將給定的向量映射到索引空間中，以便進行高效的查找。 首先，我們需要將文本數據轉換為向量。這可以通過使用 Embeddings 技術來實現。 然后，我們需要將向量存儲在一個向量空間中。這可以通過使用 VectorStore 技術來實現。 最后，索引把信息放到向量數據庫中，而檢索就是把信息提取出來，提取出來的信息與用戶提示詞合并起來，再到大模型去完成生成
from operator import itemgetter
from typing import List
import tiktoken
from langchain_core. messages import BaseMessage, HumanMessage, AIMessage, ToolMessage, SystemMessage, trim_messages
from langchain_core. chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core. prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core. runnables import RunnablePassthrough
from langchain_core. runnables. history import RunnableWithMessageHistory
from langchain_openai import OpenAIEmbeddings
from langchain_openai. chat_models import ChatOpenAI
from langchain_chroma import Chroma
vectorstore = Chroma( collection_name= "rag" , embedding_function= OpenAIEmbeddings( ) , persist_directory= "vectordb"
)
retriever = vectorstore. as_retriever( search_type= "similarity" )
def str_token_counter ( text: str ) - > int : enc = tiktoken. get_encoding( "o200k_base" ) return len ( enc. encode( text) )
def tiktoken_counter ( messages: List[ BaseMessage] ) - > int : num_tokens = 3 tokens_per_message = 3 tokens_per_name = 1 for msg in messages: if isinstance ( msg, HumanMessage) : role = "user" elif isinstance ( msg, AIMessage) : role = "assistant" elif isinstance ( msg, ToolMessage) : role = "tool" elif isinstance ( msg, SystemMessage) : role = "system" else : raise ValueError( f"Unsupported messages type { msg. __class__} " ) num_tokens += ( tokens_per_message+ str_token_counter( role) + str_token_counter( msg. content) ) if msg. name: num_tokens += tokens_per_name + str_token_counter( msg. name) return num_tokens
trimmer = trim_messages( max_tokens= 4096 , strategy= "last" , token_counter= tiktoken_counter, include_system= True ,
)
store = { }
def get_session_history ( session_id: str ) - > BaseChatMessageHistory: if session_id not in store: store[ session_id] = InMemoryChatMessageHistory( ) return store[ session_id]
model = ChatOpenAI( )
prompt = ChatPromptTemplate. from_messages( [ ( "system" , """You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. Use three sentences maximum and keep the answer concise.Context: {context}""" , ) , MessagesPlaceholder( variable_name= "history" ) , ( "human" , "{question}" ) , ]
)
def format_docs ( docs) : return "\n\n" . join( doc. page_content for doc in docs)
context = itemgetter( "question" ) | retriever | format_docs
first_step = RunnablePassthrough. assign( context= context)
chain = first_step | prompt | trimmer | model
with_message_history = RunnableWithMessageHistory( chain, get_session_history= get_session_history, input_messages_key= "question" , history_messages_key= "history" ,
)
config = { "configurable" : { "session_id" : "dreamhead" } }
while True : user_input = input ( "You:> " ) if user_input. lower( ) == 'exit' : break if user_input. strip( ) == "" : continue stream = with_message_history. stream( { "question" : user_input} , config= config) for chunk in stream: print ( chunk. content, end= '' , flush= True ) print ( ) 
