
2024年您產品必備的10大AI API推薦
價格方面,DeepSeek-V2 API 的價格為?每百萬 tokens 輸入 1 元、輸出 2 元(32K 上下文),價格僅為 GPT-4-Turbo 的近百分之一,LLaMA3-70B 出來時的成本已經震驚我了,DeepSeek-V2 這波屬于是殺紅眼了 ??,對應用層是極大的利好。
DeepSeek-V2 在中文方面是最好用的,價格也是最便宜的(社區里微調 LLaMA3-70B 中文版的朋友估計哭暈在廁所了 ??)
對 Transformer 架構中的自注意力機制進行了全方位的創新
更詳細的技術解讀請看這篇 ???幻方發布全球最強 MOE 大模型!DeepSeek-V2
一句話總結:參數更多、能力更強、成本更低!當然能不能干活,我得試了才知道,所以我決定使用 LangChain 的 LangGraph 組件構建一個編碼類 Agent 測試下(編碼能力、工具調用能力、推理能力),順便提下,自從上次 LLaMA3-70B 發布,我還在測試其在各類開源 Agent 項目中的實際效果,接下來要替換為 DeepSeek-V2 了。
LangGraph 是一套構建于 LangChain 框架之上的開發組件,可以與 LangChain 現有的鏈(Chain)、LCEL(LangChain Express Language)等無縫協作,與其生態系統完全兼容。
LangGraph 受到 NetworkX 框架的啟發,將應用流程抽象為一個圖,其中節點表示一個函數調用或一次大模型調用,而邊(則代表了節點的執行順序(即數據流),并且可以通過設置條件邊(conditional edge)來實現流程的分支控制,進而基于此圖編譯生成相應的應用。在任務執行期間,系統維護一個中央狀態對象(state),該對象會隨著節點跳轉而持續更新,其包含的屬性可根據需求自由定義,這種方式使得創建具有狀態管理和可控循環的 LLM 應用變得容易,比如多 Agent 應用。LangGraph 的作用包括:
下面是官方通過使用 LangGraph 構建 RAG 應用的示意圖,晰地闡釋了 LangGraph 中的 3 個核心概念:
下面的代碼展示了如何基于 DeepSeek-V2 和 LangGraph 實現一個 AI 編碼代理,為 Python 代碼自動編寫單元測試。部分代碼參考自?Build an AI Coding Agent with LangGraph by LangChain[3]
首先配置環境,并安裝基礎依賴。
mkdir langgraph_test && cd langgraph_test
python3 -m venv env
source env/bin/activate
pip install langgraph langchain langchain_openai colorama
創建示例文件
import os
# 定義搜索路徑,即app目錄的絕對路徑
search_path = os.path.join(os.getcwd(), "app")
# 定義crud.py文件的路徑,該文件位于search_path/src目錄下
code_file = os.path.join(search_path, "src/crud.py")
# 定義測試文件test_crud.py的路徑,該文件位于search_path/test目錄下
test_file = os.path.join(search_path, "test/test_crud.py")
# 檢查search_path路徑是否存在,如果不存在則創建
if not os.path.exists(search_path):
os.mkdir(search_path) # 創建search_path目錄
os.mkdir(os.path.join(search_path, "src")) # 在search_path下創建src目錄
os.mkdir(os.path.join(search_path, "test")) # 在search_path下創建test目錄
# 定義一個包含Item類和CRUDApp類的字符串,這些類將被寫入到code_file文件中
code = """
class Item:
def __init__(self, id, name, description=None):
self.id = id # 初始化Item對象的id屬性
self.name = name # 初始化Item對象的name屬性
self.description = description # 初始化Item對象的description屬性,可省略
def __repr__(self):
# 定義對象的字符串表示方法,便于打印和調試
return f"Item(id={self.id}, name={self.name}, description={self.description})"
class CRUDApp:
def __init__(self):
self.items = [] # 初始化一個空列表,用于存儲Item對象
def create_item(self, id, name, description=None):
# 創建一個Item對象,并將其添加到items列表中
item = Item(id, name, description)
self.items.append(item)
return item # 返回創建的Item對象
def read_item(self, id):
# 根據id讀取Item對象
for item in self.items:
if item.id == id:
return item # 如果找到匹配的id,返回Item對象
return None # 如果沒有找到匹配的id,返回None
def update_item(self, id, name=None, description=None):
# 根據id更新Item對象的name和/或description屬性
for item in self.items:
if item.id == id:
if name:
item.name = name # 如果提供了name,更新Item對象的name屬性
if description:
item.description = description # 如果提供了description,更新Item對象的description屬性
return item # 更新完成后返回Item對象
return None # 如果沒有找到匹配的id,返回None
def delete_item(self, id):
# 根據id刪除Item對象
for index, item in enumerate(self.items):
if item.id == id:
return self.items.pop(index) # 如果找到匹配的id,從列表中移除Item對象并返回
return None # 如果沒有找到匹配的id,返回None
def list_items(self):
# 返回所有存儲的Item對象的列表
return self.items
"""
# 使用with語句打開code_file文件進行寫入
with open(code_file, 'w') as f:
f.write(code) # 將code字符串寫入到文件中
示例文件創建好后,接下來編寫 Agent 處理邏輯,即為這段代碼(code 字符串內容)自動生成單元測試。
開始之前先設置好 DeepSeek-V2 大模型,由于 API 格式與 OpenAI 兼容 ,所以可以直接使用 OpenAI SDK (langchain_openai)來訪問 DeepSeek API。
llm = ChatOpenAI(base_url="https://api.deepseek.com/v1",
api_key=os.getenv("DEEPSEEK_KEY"),
model="deepseek-chat")
首先定義一個 AgentState,負責在執行過程中跟蹤代理的狀態。這主要是一個 TypedDict 類,其中包含 Agent 狀態相關屬性。
class AgentState(TypedDict): # 定義AgentState類型,用于存儲代理的狀態
class_source: str
class_methods: List[str]
tests_source: str
# 創建StateGraph
workflow = StateGraph(AgentState)
class_source 存儲代碼中的 Python 類名稱,class_methods 用于存儲 Python 類類的方法,tests_source 用于存儲生成的單元測試代碼。
現在為 AgentState 添加節點,這里需要定義 3 個節點,一個用于查找類方法的節點,一個用于更新單元測試到狀態對象的節點,一個用于將生成的單元測試寫入測試文件的節點
代碼解析函數
開始之前先聲明一個提取源代碼的工具函數,這里假設代碼片段在“`內
def extract_code_from_message(message):
lines = message.split("\n") # 按行分割消息
code = ""
in_code = False # 標記是否在代碼塊中
for line in lines:
if "```" in line: # 檢查是否是代碼塊的開始或結束
in_code = not in_code
elif in_code: # 如果在代碼塊中,則累加代碼
code += line + "\n"
return code # 返回提取的代碼
現在開始定義節點
節點 1 用于查找類方法的節點
# 系統消息模板
system_message_template = """你是一個聰明的開發者,你將使用pytest編寫高質量的單元測試。
僅用源代碼回復測試。不要在你的回復中包含類。我會自己添加導入語句。
如果沒有要寫的測試,請回復“# 沒有要寫的測試”且不要包含類。
示例:
def test_function():
...
請務必遵循指令并編寫高質量的測試,不要寫測試類,只寫方法。
"""
import_prompt_template = """
這是一條包含代碼文件路徑的信息:{code_file}。
這是一條包含測試文件路徑的信息:{test_file}。
請為文件中的類編寫正確的導入語句。
"""
# 發現類及其方法
def discover_function(state: AgentState):
assert os.path.exists(code_file) # 確保代碼文件存在
with open(code_file, "r") as f: # 打開代碼文件進行讀取
source = f.read() # 讀取文件內容
state["class_source"] = source # 將源代碼存儲在狀態中
# 獲取方法
methods = []
for line in source.split("\n"):
if "def " in line: # 如果行中包含def,表示這是一個方法定義
methods.append(line.split("def ")[1].split("(")[0])
state["class_methods"] = methods # 將方法名存儲在狀態中
# 生成導入語句并啟動代碼
import_prompt = import_prompt_template.format(
code_file=code_file, # 格式化導入提示模板
test_file=test_file
)
message = llm.invoke([HumanMessage(content=import_prompt)]).content # 調用模型生成消息
code = extract_code_from_message(message) # 提取消息中的代碼
state["tests_source"] = code + "\n\n" # 將測試源代碼存儲在狀態中
return state # 返回更新后的狀態
# 將節點添加到工作流中
workflow.add_node(
"discover", # 節點名稱
discover_function # 節點對應的函數
)
上面的代碼片段中從 AgentState 的 class_source 元素中提取代碼,并將它們傳遞給 LLM ,然后將 LLM 的響應存儲在 AgentState 的 tests_source 元素中。
節點 2 更新單元測試到狀態對象的節點
# 編寫測試模板
write_test_template = """這里是類:
'''
{class_source}
'''
為方法“{class_method}”實現一個測試。
"""
def write_tests_function(state: AgentState):
# 獲取下一個要編寫測試的方法
class_method = state["class_methods"].pop(0)
print(f"為{class_method}編寫測試。")
# 獲取源代碼
class_source = state["class_source"]
# 創建提示
write_test_prompt = write_test_template.format(
class_source=class_source,
class_method=class_method
)
print(colorama.Fore.CYAN + write_test_prompt + colorama.Style.RESET_ALL) # 打印提示信息
# 獲取測試源代碼
system_message = SystemMessage(system_message_template) # 創建系統消息
human_message = HumanMessage(write_test_prompt) # 創建人類消息
test_source = llm.invoke([system_message, human_message]).content # 調用模型生成測試代碼
test_source = extract_code_from_message(test_source) # 提取消息中的測試代碼
print(colorama.Fore.GREEN + test_source + colorama.Style.RESET_ALL) # 打印測試代碼
state["tests_source"] += test_source + "\n\n" # 將測試源代碼添加到狀態中
return state # 返回更新后的狀態
# 將編寫測試節點添加到工作流中
workflow.add_node(
"write_tests",
write_tests_function
)
讓 LLM 為每個方法編寫測試用例,并更新到 AgentState 的 tests_source 元素中。
節點 3 將生成的單元測試寫入測試文件
# 編寫文件
def write_file(state: AgentState):
with open(test_file, "w") as f: # 打開測試文件進行寫入
f.write(state["tests_source"]) # 寫入測試源代碼
return state # 返回狀態
# 將寫文件節點添加到工作流中
workflow.add_node(
"write_file",
write_file
)
現在已經有 3 個節點,接下來定義邊用于指定它們之間的執行方向。Agent 工作流從查找類方法的節點開始執行,然后轉到編寫單元測試的節點。
# 定義入口點,這是流程開始的地方
workflow.set_entry_point("discover")
# 總是從discover跳轉到write_tests
workflow.add_edge("discover", "write_tests")
add_conditional_edge 函數添加了 write_tests 函數和 should_continue 函數,該函數根據 class_methods 條目決定采取哪一步,
# 判斷是否完成
def should_continue(state: AgentState):
if len(state["class_methods"]) == 0: # 如果沒有更多的方法要測試
return "end" # 結束流程
else:
return "continue" # 繼續流程
# 添加條件邊
workflow.add_conditional_edges(
"write_tests", # 條件邊的起始節點
should_continue, # 條件函數
{
"continue": "write_tests", # 如果應該繼續,則再次執行write_tests節點
"end": "write_file" # 如果結束,則跳轉到write_file節點
}
)
# 總是從write_file跳轉到END
workflow.add_edge("write_file", END)
當從 LLM 生成所有方法的單元測試后,測試代碼被寫入測試文件。
最后剩下的就是編譯工作流并運行它。
app = workflow.compile() # 編譯工作流
inputs = {} # 輸入參數
config = RunnableConfig(recursion_limit=100) # 設置遞歸限制
try:
result = app.invoke(inputs, config) # 運行應用
print(result) # 打印結果
except GraphRecursionError: # 如果達到遞歸限制
print("達到圖遞歸限制。") # 打印錯誤信息
遞歸限制是給定工作流的 LLM 進行循環推理的次數,當超過限制時,工作流停止。
下面是最終生成的 test_crud.py 文件的內容,大家自行評判。
class TestDict(unittest.TestCase):
def test_crudapp_init():
# 創建一個CRUDApp實例
app = CRUDApp()
# 驗證items列表是否為空
assert app.items == []
def test_item_repr():
item = Item(1, 'Test Item', 'This is a test item')
expected_repr = "Item(id=1, name=Test Item, description=This is a test item)"
assert repr(item) == expected_repr
def test_crudapp_init():
# 創建一個CRUDApp實例
app = CRUDApp()
# 驗證items列表是否為空
assert app.items == []
def test_create_item():
# 初始化CRUDApp實例
app = CRUDApp()
# 創建一個新項目
item = app.create_item(1, 'Test Item')
# 斷言項目已創建并存在于項目列表中
assert item in app.list_items()
assert item.id == 1
assert item.name == 'Test Item'
assert item.description is None
# 創建另一個項目,確保ID是唯一的
another_item = app.create_item(2, 'Another Test Item')
assert another_item in app.list_items()
assert another_item.id == 2
assert another_item.name == 'Another Test Item'
assert another_item.description is None
# 斷言列表中的項目數量正確
assert len(app.list_items()) == 2
def test_read_item():
# 初始化CRUDApp實例
app = CRUDApp()
# 創建一個測試項
test_item = app.create_item(1, 'Test Item')
# 測試讀取存在的項
read_item = app.read_item(1)
assert read_item == test_item
# 測試讀取不存在的項
non_existent_item = app.read_item(99)
assert non_existent_item is None
def test_update_item():
# 初始化CRUDApp實例
app = CRUDApp()
# 創建一個測試用的Item
item = app.create_item(1, 'Test Item')
# 更新Item的名稱
updated_item = app.update_item(1, 'Updated Test Item')
# 驗證更新是否成功
assert updated_item is not None
assert updated_item.id == 1
assert updated_item.name == 'Updated Test Item'
assert updated_item.description is None
# 驗證列表中的Item是否更新
listed_items = app.list_items()
assert len(listed_items) == 1
assert listed_items[0].id == 1
assert listed_items[0].name == 'Updated Test Item'
assert listed_items[0].description is None
# 嘗試更新不存在的Item
non_existent_item = app.update_item(2, 'Non Existent Item')
assert non_existent_item is None
# 更新Item的描述
updated_item_description = app.update_item(1, description='Updated Test Description')
# 驗證更新是否成功
assert updated_item_description is not None
assert updated_item_description.id == 1
assert updated_item_description.name == 'Updated Test Item'
assert updated_item_description.description == 'Updated Test Description'
# 驗證列表中的Item是否更新
listed_items = app.list_items()
assert len(listed_items) == 1
assert listed_items[0].id == 1
assert listed_items[0].name == 'Updated Test Item'
assert listed_items[0].description == 'Updated Test Description'
def test_delete_item():
# 創建一個CRUDApp實例
app = CRUDApp()
# 創建一些測試用的Item對象
item1 = app.create_item(1, 'Item 1')
item2 = app.create_item(2, 'Item 2')
# 驗證創建的Item對象是否在列表中
assert item1 in app.list_items()
assert item2 in app.list_items()
# 刪除一個Item對象
deleted_item = app.delete_item(1)
# 驗證刪除操作是否成功
assert deleted_item == item1
assert item1 not in app.list_items()
assert item2 in app.list_items()
# 嘗試刪除一個不存在的Item對象
deleted_nonexistent_item = app.delete_item(3)
# 驗證刪除不存在的Item對象時返回None
assert deleted_nonexistent_item is None
def test_list_items():
# 創建一個CRUDApp實例
app = CRUDApp()
# 創建一些測試用的Item實例
item1 = app.create_item(1, 'Item 1')
item2 = app.create_item(2, 'Item 2', 'This is item 2')
# 調用list_items方法并檢查返回的列表是否包含我們創建的Item實例
listed_items = app.list_items()
assert item1 in listed_items
assert item2 in listed_items
# 檢查列表的長度是否正確
assert len(listed_items) == 2
生成的單元測試,屬于是一種基本可用的狀態,多次(10 次)執行輸出的結果也較為穩定,通過和 GPT-4-Turbo、GLM-4 生成的進行比較,DeepSeek-V2 在一些 Agent 應用場景確實可以做到平替,畢竟百分之一的成本,性價比在這里擺著,還要啥自行車,等我后續在其他更復雜的 Agent 項目進行進一步驗證。
[1]瘋狂的幻方:一家隱形 AI 巨頭的大模型之路:?https://36kr.com/p/2272896094586500
[2]MoE 架構:?https://baoyu.io/translations/llm/what-is-mixture-of-experts
[3]Build an AI Coding Agent with LangGraph by LangChain:?https://www.analyticsvidhya.com/blog/2024/03/build-an-ai-coding-agent-with-langgraph-by-langchain/
文章轉自微信公眾號@莫爾索隨筆