在 Glow.ai 上的 VM 試玩 LangGraph

Python
LLM
LangGraph
linux
just play
作者

紙魚

發佈於

2026年7月3日

摘要
為了消耗 Glow.ai 上的 credit ,玩一下 LangGraph ,但 VM 是 Linux 環境,所以順便玩:D

緣起

這學習我修了 LLM cybersecurity 這門課,有送 Glow.ai 的 credit 讓我們實作。不過老師一直說 credit 要省著用,於是我也努力發揮勤儉的精神,幾乎都在我的筆電、 google Colab 或是用學校的桌電做作業,就這樣做到期末才發現我的專案因為走一個破電腦也能執行的路線(?),在本機上執行就好。於是乎我現在有超多 credit ,根本就是 credit 富翁!

截圖紀念一下

不過如果 credit 一直不用,最後也會因為帳號到期而用不了,這樣勤儉持家的意義就沒了 QQ。所以我就想了一些我想做的事情,其中一個就是摸一下之前有看沒有懂的 LangGraph。請 AI 幫我規劃、實作簡單的範本看看,於是就有這篇文章。

簡單介紹 Glow.ai

官方的文件介紹

Glow.ai 是類似 AWS 的雲端廠商,提供虛擬機(VM)租借服務。這次租用的計費方式看起來是預先購買 credit 給有team權限的管理員,分發給同一個 team 下的所有成員(學生),成員再依照需求租用自己需要的機器。不同規格的機器每花的 credit 不同,這次我用的是最最便宜的 、只有 CPU 的 Intel(R) Xeon(R) Platinum 8468 (0.180 credit / Hour),裡面有 Ubuntu 24.04 的 image。

Glow.ai 下的執行步驟

ssh port 連線限定的部分

這台機器( Intel(R) Xeon(R) Platinum 8468 )因為只能用 ssh port 連線的,我一開始看不懂,所以紀錄一下。

  1. 申請好機器後,在 Access 下可以看到系統給你一組 ssh 指令跟密碼(被遮住的樣子)

  2. 在 Windows 系統裡,開啟 CMD(命令提示字元)

  3. 先輸入 ssh 指令,成功時系統會要求輸入密碼

  4. 複製密碼在 CMD 貼上(Ctrl + V),貼上後沒有出現密碼本人是正常的,直接 Enter

  5. 上一部成功會出現 Are you sure you want to continue connecting (yes/no/[fingerprint])?,輸 yes

  6. commend line 前面變成以 glows 開頭,以 ~$ 結尾的東西可以輸入就代表成功進入 VM 裡了。

Linux 部分

進入後先不要急著裝套件,因為這台 VM 甚麼都沒有,apt 套件(Linux 的套件管理工具)也過期了,所以先 update

sudo apt update

再安裝 pip

sudo apt install python3-pip

沒有任何 Error 訊息的話,就可以裝 python 套件了!

建專案環境跟套件部份

虛擬環境 & 套件

先建 venv 虛擬環境

python3 -m venv venv

再啟動

source venv/bin/activate

看到 commend line 前面變成多了(venv)就成功了。接著就可以自由安裝本次用到的套件:

pip install langgraph langchain langchain-openai python-dotenv langchain-groq

可用以下指令確認套件有沒有都安裝成功

pip list | grep lang
langchain            1.3.9
langchain-core       1.4.7
langchain-openai     1.3.2
langchain-protocol   0.0.17
langgraph            1.2.5
langgraph-checkpoint 4.1.1
langgraph-prebuilt   1.1.0
langgraph-sdk        0.4.2
langsmith            0.8.16

專案資料夾 & 進入建檔

mkdir ~/langgraph-demo
cd ~/langgraph-demo

這邊其實跟在 Windows 上的指令一樣。

簡單的 LangGraph 專案測試

先建專案

建立第一個檔案

nano app.py

這個指令會進入 app.py 這個檔案裡面編輯,可以貼上以下這段程式碼:

from typing import TypedDict

from langgraph.graph import StateGraph, END

class State(TypedDict):
    message: str

def process(state: State):
    return {
        "message": state["message"] + " -> LangGraph works!"
    }

builder = StateGraph(State)

builder.add_node("process", process)

builder.set_entry_point("process")
builder.add_edge("process", END)

graph = builder.compile()

result = graph.invoke({
    "message": "hello"
})

print(result)

在編輯模式下可以看到下面的快捷鍵提示,先打 Ctrl+O (Write out) 後下面的狀態會變 File Name to Write: app.py,先按 Enterapp.py就存好了。

關閉檔案則是按 Ctrl+X,就可以回到原本的 commend line 模式,執行 app.py

python app.py

就能看到 LangGraph 傳來的訊息

{'message': 'hello -> LangGraph works!'}

所以剛剛發生什麼事?

這個專案是用 LangGraph 來建立一個簡單的 State Graph,就像是可以動態跑一次的流程圖:

開始 → 收到訊息(message) → 回傳 message 裡的訊息

藉由定義 State、Node 跟 Edge 的方式實現。

State 就是在這個地方:

from typing import TypedDict

class State(TypedDict):
    message: str

目的是定義 graph 裡流動的資料格式,input 固定為有型別檢查的字典格式(TypedDict),這裡只會吐出 message,也會是字典格式。

為什麼要固定使用有型別檢查的字典格式?

LangGraph 的 state 本質是「可序列化、可逐步更新的資料流」,不是物件導向的封裝結構。單純 class 的設計沒有辦法依照不同狀態更新欄位。相對的,TypedDict 結合 一般字典(dict) 的彈性,自動型別檢查也可避免寫錯 key 。

Node 則是在

def process(state: State):
    return {
        "message": state["message"] + " -> LangGraph works!"
    }

Node 是函數形式,將從 State 那裡收到的狀態做更新。這裡的函數定義是:在原本的 message 後面加上 ” -> LangGraph works!“。

接下來的這一段是建立 Graph 並讓它運作的過程

# 定義建構器
builder = StateGraph(State) #這個 graph 的資料結構是 State

# 加入設定好的 node,要附上自訂名稱及函數名稱
builder.add_node("process", process)

# 設定入口點(entry point),代表從 process 開始執行
builder.set_entry_point("process")
# 設定結束點(edge point),這裡設定一個流程結束後不再執行所以 END
builder.add_edge("process", END)
# 編譯 graph
graph = builder.compile()

# 執行 graph
result = graph.invoke({
    "message": "hello"
})

print(result)

於是結果就是

{'message': 'hello -> LangGraph works!'}

以上可知

線上版聊天 AI

來做個對話式 AI 吧。這裡一樣是用 Grop 的免費 API 來做示範,教學請參閱另一篇文章

  1. 申請好 API key後,建env
nano .env

在編輯模式下建立變數與API Key

GROQ_API_KEY=你的GroqKey
  1. 在同樣的專案資料夾下建立 chat_agent.py
nano chat_agent.py
import os
from dotenv import load_dotenv

from langchain_groq import ChatGroq

load_dotenv()

llm = ChatGroq(
    model="llama-3.3-70b-versatile",
    api_key=os.getenv("GROQ_API_KEY")
)

response = llm.invoke("用一句話解釋 LangGraph 是什麼")

print(response.content)

特地換成高級一點的模型玩玩

  1. 測試
python chat_agent.py

有成功回應問題(例如”用一句話解釋 LangGraph 是什麼”)就表示成功。

加入角色分工

LangGraph 強大的地方在於它可以一次讓 AI 扮演不同角色,每個角色有不同的專業,可以一起完成被分配的任務。這裡以兩個 AI 角色:PM 跟工程師(PG)為例,我們的目標是建立一個使用者提需求,PM 審核後再交給工程師產 code ,整體結構如下:

User提需求 → PM提供初步系統架構設計 → PG提供正式架構 → Final 報告

串 LLM 與建立 prompt

因為架構跟上一個比起來有點小複雜,所以換成兩支腳本。

串 LLM 部分

先 nano 一個 llm.py,讓整體架構更清晰。

nano llm/py
import os
from dotenv import load_dotenv
from langchain_groq import ChatGroq

load_dotenv()

llm = ChatGroq(
    model="llama-3.1-8b-instant",
    api_key=os.getenv("GROQ_API_KEY")
)

發現模型感覺不用太高級又換回來了

建立 prompt 部分

PM_PROMPT = """
你是產品經理(PM)。

使用者需求:
{user_input}

請輸出:
1. 清楚需求描述
2. 可執行 task list(用條列)
"""

ENGINEER_PROMPT = """
你是資深後端工程師(Senior Engineer)。

請根據以下系統設計,產出:

1. 專案資料夾結構
2. Python FastAPI code
3. requirements.txt
4. 如何啟動專案(指令)

系統設計如下:
{pm_output}

使用者需求:
{user_input}

請輸出可直接執行的內容。
"""

修改 agent 裡的流程

修改 agent.py

import subprocess
from typing import TypedDict

from langgraph.graph import StateGraph, END

from llm import llm
from prompts import PM_PROMPT, ANALYZER_PROMPT


# 角色變多 state 的 input 也會變多
class State(TypedDict):
    user_input: str
    pm_output: str
    engineer_output: str
    final: str


# ---------------- PM ----------------
def pm_node(state: State):

    prompt = PM_PROMPT.format(
        user_input=state["user_input"]
    )

    res = llm.invoke(prompt)

    return {
        "pm_output": res.content
    }


# -------------- Engineer --------------
# 工程師負責產 code
def engineer_node(state: State):

    prompt = ENGINEER_PROMPT.format(
        user_input=state["user_input"],
        pm_output=state["pm_output"],
    )

    res = llm.invoke(prompt)

    return {
        "engineer_output": res.content
    }
    


# -------------- Final node --------------
# 最後要產出報告
def final_node(state: State):
    return {
        "final": f"""
===== 🧑‍💼 PM 設計 =====

{state['pm_output']}

===== 👷 Engineer 實作 =====

{state['engineer_output']}
"""
    }

# -------------- Build Graph --------------
def build_graph():
    builder = StateGraph(State)

    builder.add_node("pm", pm_node)
    builder.add_node("engineer", engineer_node)
    builder.add_node("final", final_node)

    builder.set_entry_point("pm")

    builder.add_edge("pm", "engineer")
    builder.add_edge("engineer", "final")
    builder.add_edge("final", END)

    return builder.compile()


graph = build_graph()

修改 app.py

讓它可以成功吐出報告結果。

from agent import graph

if __name__ == "__main__":

    user_input = input("Enter your problem: ")

    result = graph.invoke({
        "user_input": user_input,
        "spec": "",
        "tasks": [],
        "logs": "",
        "result": ""
    })

    print("\n\n===== FINAL REPORT =====\n")
    print(result["final"])

完成後回到 terminal 模式再跑一次

python app.py

可出現對話,這時你可以隨便下個需求,讓 PM 設計系統,工程師角色產 code。不過很快就會發現,需求下得很不合理的時候最後很可能會產出不合時宜的 code,需要設定一個「可能會打槍」的流程。

設計迴圈流程

這次把流程改成這樣:

        ┌─────────────────┐
        │                 ↓
User → PM → Business → Engineer
        ↑                 │
        │                 ↓
        └──── Reviewer ←-┘
                │
          通過? / 打槍?(2次後進 Final)
                │
              Final

多 2 個角色

  • Business: 確定 PM 開的規格符合商業價值

  • Reviewer: 審核專案的可行性

修改 agent.py

首先 State 要修改 State 的部份

#多了business角色
# 迴圈部分也要定義
class State(TypedDict):
    user_input: str
    pm_output: str
    business_output: str
    engineer_output: str
    review_result: str
    retry_count: int
    final: str

再來新增角色

prompts.py 更新 prompt

nano prompts.py

改成

PM_PROMPT = """
你是資深產品經理(PM)。

使用者需求:
{user_input}

請輸出:

1. 系統目標
2. 使用者需求整理
3. 功能需求
4. 開發範圍
5. 建議優先順序
"""


BUSINESS_PROMPT = """
你是商業數據分析師(Business Analyst)。

請評估以下專案:

使用者需求:
{user_input}

PM 規劃:
{pm_output}

分析:

1. 商業價值
2. 目標客群
3. 解決痛點
4. 市場機會
5. ROI
6. KPI
7. 開發風險
8. 是否值得投入

最後給出:
VALUE_SCORE: 1-10
"""


ENGINEER_PROMPT = """
你是資深後端工程師。

根據:

需求:
{user_input}

PM:
{pm_output}

商業分析:
{business_output}


產出:

1. 系統架構
2. 技術選型
3. 專案結構
4. API 設計
5. 開發步驟
"""


REVIEW_PROMPT = """
你是專案審查委員。

審查:

PM:
{pm_output}

Business:
{business_output}

Engineer:
{engineer_output}


判斷:

商業價值是否足夠?
技術是否可行?
需求是否清楚?


如果通過輸出:
APPROVE

如果不通過輸出:
REJECT

並說明原因。

只要出現以下情況 → 必須 REJECT:

1. 技術名詞被錯誤定義(例如 RAG 不是 Retrieval-Augmented Generation)
2. 系統需求與技術實作不一致
3. 架構包含明顯幻想(不存在的 AI training flow / 自行學習但無方法)
"""

主要更新的內容就是除了每個角色增加都要定義一次函數外,設定 reject 次數也需要定義。

回到 agent.py

修改成以下

import subprocess
from typing import TypedDict

from langgraph.graph import StateGraph, END

from llm import llm
from prompts import (
    PM_PROMPT,
    BUSINESS_PROMPT,
    ENGINEER_PROMPT,
    REVIEWER_PROMPT
)

#多了business 跟 reviewer角色
# 打槍幾次部分也要定義
class State(TypedDict):
    user_input: str
    pm_output: str
    business_output: str
    engineer_output: str
    reviewer_output: str
    review_count: int

    final: str

# ---------------- PM ----------------
def pm_node(state: State):

    prompt = PM_PROMPT.format(
        user_input=state["user_input"]
    )

    res = llm.invoke(prompt)

    return {
        "pm_output": res.content
    }

# -------------- Business --------------
def business_node(state: State):

    prompt = BUSINESS_PROMPT.format(
        user_input=state["user_input"],
        pm_output=state["pm_output"]
    )

    res = llm.invoke(prompt)

    return {
        "business_output": res.content
    }

# -------------- Engineer --------------
# 工程師負責產 code
def engineer_node(state: State):

    prompt = ENGINEER_PROMPT.format(
        user_input=state["user_input"],
        pm_output=state["pm_output"],
        business_output=state["business_output"],
    )

    res = llm.invoke(prompt)

    return {
        "engineer_output": res.content
    }
    
# -------------- Reviewer --------------
def reviewer_node(state):

    prompt = REVIEWER_PROMPT.format(
        pm_output=state["pm_output"],
        business_output=state["business_output"],
        engineer_output=state["engineer_output"]
    )

    res = llm.invoke(prompt)

    review_count = state.get("review_count", 0)

    if "PASS" not in res.content:
        review_count += 1

    return {
        "reviewer_output": res.content,
        "review_count": review_count
    }

# -------------- Review router --------------
# 打槍次數也需要設定
def review_router(state):

    if "PASS" in state["reviewer_output"]:
        return "final"

    if state["review_count"] >= 2:
        return "final"

    return "pm"

# -------------- Final node --------------
# 最後要產出報告
def final_node(state: State):
    return {
        "final": f"""
===== 🧑‍💼 PM 設計 =====
{state['pm_output']}

===== 👷 Engineer 實作 =====
{state['engineer_output']}

===== 👷 Reviewer 結果 =====
{state['reviewer_output']}

"""
    }

# -------------- Build Graph --------------
def build_graph():
    builder = StateGraph(State)
    builder.add_node("pm", pm_node)
    builder.add_node("business", business_node)
    builder.add_node("engineer", engineer_node)
    builder.add_node("reviewer", reviewer_node)
    builder.add_node("final", final_node)

    builder.set_entry_point("pm")

    builder.add_edge("pm", "business")
    builder.add_edge("business", "engineer")
    builder.add_edge("engineer", "reviewer")

    builder.add_conditional_edges(
        "reviewer",
        review_router,
        {
            "pm": "pm",
            "final": "final",
        },
    )

    builder.add_edge("final", END)

    return builder.compile()



graph = build_graph()

測試玩具

只要在終端機叫 app.py 就可以測試

python app.py

正常需求

先來一個正常、定義較為明確的需求。

做一個RAG系統檢索公司技術文件

結果回應在下面

===== FINAL REPORT =====

===== 🧑‍💼 PM 設計 ===== 需求描述

系統名稱: RAG 技術文件檢索系統

目的: 建立一個系統用於檢索公司技術文件,提高查詢效率和便捷性。

功能需求:

  1. 文件管理: 能夠儲存和管理公司技術文件,包括文檔、圖片、視頻等多種格式。
  2. 查詢功能: 提供多種查詢方式,包括:
  • 名稱查詢
  • 說明查詢
  • 代碼查詢
  • 日期查詢
  1. 篩選功能: 能夠根據使用者的需求篩選查詢結果,包括:
  • 根據文件類型篩選
  • 根據文件日期篩選
  1. 排序功能: 能夠根據使用者的需求排序查詢結果,包括:
  • 根據文件名稱排序
  • 根據文件日期排序
  1. 附件功能: 能夠上傳和下載文件附件。
  2. 使用者管理: 能夠管理使用者角色和權限,包括:
  • 系統管理者角色
  • 一般使用者角色
  1. 安全功能: 能夠設定安全控制,包括:
  • 登錄認證
  • 文件加密
  • 存取控制

非功能需求:

  1. 使用者體驗: 系統應該提供友好的使用者界面和簡單的操作流程。
  2. 效能: 系統應該能夠高效地處理查詢和儲存文件。
  3. 穩定性: 系統應該能夠穩定地運行和處理需求。

技術需求:

  1. 框架: 使用現成的開源框架(例如 Django 或 Flask)建設系統。
  2. 資料庫: 使用 MySQL 或 PostgreSQL 資料庫儲存文件和使用者資料。
  3. 安全: 使用 SSL/TLS 來實現安全連接和文件加密。
  4. 部署: 系統應該能夠在雲平台(例如 AWS 或 Azure)部署和運行。

任務清單:

  1. 需求收集和分析:
  • 收集和分析公司技術文件需求
  • 設計系統架構和功能需求
  1. 系統設計:
  • 設計系統框架和架構
  • 設計資料庫結構和安全功能
  1. 代碼實現:
  • 實現查詢和篩選功能
  • 實現排序和附件功能
  • 實現使用者管理和安全功能
  1. 測試和驗證:
  • 實現單元測試和統一測試
  • 驗證系統功能和安全性
  1. 部署和上線:
  • 部署系統到雲平台
  • 上線系統並測試功能和穩定性
  1. 維護和升級:
  • 實現系統維護和升級功能
  • 實現安全更新和維護

===== 🧑‍💼 商業價值評估 ===== 基於提供的專案資料,以下是我的分析:

1. 商業價值

該系統能夠提高公司技術文件的查詢效率和便捷性,減少工作時間和成本。它還能夠提供安全和穩定地存儲和管理文件。

2. 目標客群

公司內的所有人員,特別是技術人員和管理人員。

3. 解決痛點

  • 查詢效率低下
  • 文件管理不方便
  • 安全和穩定性問題

4. 市場機會

  • 公司內的其他部門可能也需要類似的系統
  • 業界中其他公司可能也需要類似的系統

5. ROI

系統能夠幫助公司節省工作時間和成本,提高效率和產出。預計 ROI 可能在 1-2 年內達到。

6. KPI

  • 系統的查詢效率和便捷性
  • 系統的安全和穩定性
  • 系統的使用率和滿意度

7. 開發風險

  • 系統的設計和實現是否符合公司的需求
  • 系統的安全和穩定性是否能夠滿足公司的要求
  • 系統的部署和上線是否能夠順暢

8. 是否值得投入

基於分析,該系統的商業價值和 ROI 都是正面的。雖然開發風險存在,但它們可以通過充分的需求收集、系統設計和測試來降低。因此,我們可以給該系統的 VALUE_SCORE 評分為 8

===== 👷 Engineer 實作 ===== 專案資料夾結構:

rag-system/
app/
__init__.py
main.py
models
__init__.py
file.py
user.py
routes
__init__.py
auth.py
file.py
user.py
schemas
__init__.py
file.py
user.py
utils
__init__.py
auth.py
file.py
requirements.txt
tests
__init__.py
test_auth.py
test_file.py
test_user.py
config.py
README.md

Python FastAPI code:

app/main.py:

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi.requests import Request
from pydantic import BaseModel
from typing import List, Optional
from uvicorn import run
from app.routes.auth import auth_router
from app.routes.file import file_router
from app.routes.user import user_router
from app.utils.auth import get_current_user, get_user
from app.utils.file import get_file
from app.utils.user import get_user

app = FastAPI()

app.include_router(auth_router)
app.include_router(file_router)
app.include_router(user_router)

# API Root
@app.get("/", tags=["Root"])
async def read_root():
    return {"message": "Welcome to RAG System"}

# API Login
@app.post("/login", tags=["Auth"])
async def login(username: str, password: str):
    user = get_user(username, password)
    if user:
        access_token = user.generate_access_token()
        return {"access_token": access_token}
    else:
        return {"error": "Invalid username or password"}

# API Register
@app.post("/register", tags=["Auth"])
async def register(username: str, password: str, email: str):
    user = get_user(username, password, email)
    if user:
        return user
    else:
        return {"error": "Failed to register"}

# API File
@app.get("/files", tags=["File"])
async def get_files():
    files = get_file()
    return files

# API File Upload
@app.post("/files/upload", tags=["File"])
async def upload_file(file: bytes, filename: str):
    file = get_file().upload_file(file, filename)
    return file

# API File Download
@app.get("/files/{file_id}", tags=["File"])
async def download_file(file_id: int):
    file = get_file().get_file(file_id)
    if file:
        return file
    else:
        return {"error": "File not found"}

# API User
@app.get("/users", tags=["User"])
async def get_users():
    users = get_user()
    return users

# API User Profile
@app.get("/users/{user_id}", tags=["User"])
async def get_user_profile(user_id: int):
    user = get_user().get_user(user_id)
    if user:
        return user
    else:
        return {"error": "User not found"}

app/routes/auth.py:

from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from typing import List, Optional
from app.utils.auth import get_current_user, get_user

auth_router = APIRouter()

@auth_router.post("/login", tags=["Auth"])
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = get_user(form_data.username, form_data.password)
    if user:
        access_token = user.generate_access_token()
        return {"access_token": access_token}
    else:
        raise HTTPException(status_code=401, detail="Invalid username or password")

@auth_router.post("/register", tags=["Auth"])
async def register(username: str, password: str, email: str):
    user = get_user(username, password, email)
    if user:
        return user
    else:
        raise HTTPException(status_code=400, detail="Failed to register")

app/routes/file.py:

from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import List, Optional
from app.utils.file import get_file

file_router = APIRouter()

@file_router.get("/files", tags=["File"])
async def get_files():
    files = get_file()
    return files

@file_router.post("/files/upload", tags=["File"])
async def upload_file(file: bytes, filename: str):
    file = get_file().upload_file(file, filename)
    return file

@file_router.get("/files/{file_id}", tags=["File"])
async def download_file(file_id: int):
    file = get_file().get_file(file_id)
    if file:
        return file
    else:
        raise HTTPException(status_code=404, detail="File not found")

app/routes/user.py:

from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import List, Optional
from app.utils.user import get_user

user_router = APIRouter()

@user_router.get("/users", tags=["User"])
async def get_users():
    users = get_user()
    return users

@user_router.get("/users/{user_id}", tags=["User"])
async def get_user_profile(user_id: int):
    user = get_user().get_user(user_id)
    if user:
        return user
    else:
        raise HTTPException(status_code=404, detail="User not found")

app/utils/auth.py:

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi import Depends, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from app.models.user import User

def get_current_user():
    # This function should return the current user based on the access token
    # For simplicity, we will just return a dummy user
    return User(username="dummy", password="dummy", email="dummy@example.com")

def get_user(username: str, password: str):
    # This function should return the user based on the username and password
    # For simplicity, we will just return a dummy user
    return User(username=username, password=password, email="dummy@example.com")

def get_user_by_id(user_id: int):
    # This function should return the user based on the user ID
    # For simplicity, we will just return a dummy user
    return User(username="dummy", password="dummy", email="dummy@example.com")

app/utils/file.py:

from fastapi import Depends, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from app.models.file import File

def get_file():
    # This function should return the file based on the file ID
    # For simplicity, we will just return a dummy file
    return File(file_id=1, filename="dummy_file.txt", content="dummy file content")

def upload_file(file: bytes, filename: str):
    # This function should upload the file and return the file object
    # For simplicity, we will just return a dummy file
    return File(file_id=2, filename=filename, content=file)

app/utils/user.py:

from fastapi import Depends, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from app.models.user import User

def get_user(username: str, password: str):
    # This function should return the user based on the username and password
    # For simplicity, we will just return a dummy user
    return User(username=username, password=password, email="dummy@example.com")

def get_user_by_id(user_id: int):
    # This function should return the user based on the user ID
    # For simplicity, we will just return a dummy user
    return User(username="dummy", password="dummy", email="dummy@example.com")

app/models/file.py:

from pydantic import BaseModel
from typing import List, Optional

class File(BaseModel):
    file_id: int
    filename: str
    content: str

    class Config:
        orm_mode = True

app/models/user.py:

from pydantic import BaseModel
from typing import List, Optional

class User(BaseModel):
    username: str
    password: str
    email: str

    class Config:
        orm_mode = True

requirements.txt:

fastapi==0.95.3
uvicorn==0.20.0
python-dotenv==0.19.2
sqlalchemy==1.4.34
pytest==7.1.2
pytest-cov==2.12.1

如何啟動專案:

  1. 安裝所需的庫: pip install -r requirements.txt
  2. 啟動 FastAPI: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
  3. 開啟瀏覽器並瀏覽 http://localhost:8000/docs 來檢視 API 文件

注意: 這是一個簡化的範例,需要根據實際需求進行擴展和修改。

===== 👷 Reviewer 結果 ===== 審查結果:

經過審查,我們判斷該系統的商業價值、技術可行性和需求清楚性如下:

商業價值:

該系統能夠提高公司技術文件的查詢效率和便捷性,減少工作時間和成本。這對於公司的業務運營和效率提升具有重要意義。因此,我們判斷該系統的商業價值是足夠的。

技術可行性:

系統的架構和設計看起來是合理的,使用了現成的開源框架(FastAPI)和資料庫(MySQL或PostgreSQL)。安全控制和文件管理也被合理地設計和實現。因此,我們判斷該系統的技術可行性是足夠的。

需求清楚性:

需求描述清晰明了,涵蓋了系統的主要功能和非功能性需求。系統的使用場景和使用者群也被清晰地定義。因此,我們判斷該系統的需求清楚性是足夠的。

最終判斷:

基於上述分析,我們判斷該系統的商業價值、技術可行性和需求清楚性都是足夠的。因此,我們 APPROVE 這個系統。

建議:

  1. 進一步優化系統的性能和穩定性。
  2. 實現系統的持續整合和部署。
  3. 提供系統的使用者手冊和教學資源。

初步看下來除了技術都用 python 實現這點不完全符合現實專案長期運作的要件外,整體看起來正常。

不合理的需求

測試一個完全不合常理的需求

我要做一個基於 LLM 的 RAG 系統,在筆電上 3 秒內完成回答。

結果

===== FINAL REPORT =====

===== 🧑‍💼 PM 設計 ===== 需求描述

基於 LLM 的 RAG (Risk, Action, Goal) 系統,旨在於筆電上快速回答相關問題。系統的要求如下:

  • 使用 LLM 技術作為核心引擎
  • 必須在 3 秒內完成回答
  • 提供 RAG 結構的回答,包含風險、行動和目標資訊

執行任務列表

下列為系統開發的任務列表:

  1. 選擇 LLM 引擎
  • 選擇合適的 LLM 型別(例如 BERT, RoBERTa 等)
  • 設定 LLM 變數,包括模型大小、精度等
  1. 構建 RAG 結構
  • 設計 RAG 結構的資料模型,包含風險、行動和目標資訊
  • 實現 RAG 結構的資料儲存和查詢功能
  1. 開發快速回答引擎
  • 實現 LLM 和 RAG 結構的整合
  • 實現快速回答的算法和模型
  1. 優化系統性能
  • 優化系統的計算效率,達到 3 秒內完成回答的速度
  • 實現系統的穩定性和安全性
  1. 測試和驗證
  • 進行系統的測試和驗證,確保系統的正確性和穩定性
  1. 優化和迭代
  • 根據使用者反饋和系統的性能進行優化和迭代

注意

  • 這個需求描述和任務列表可能需要進一步細化和修改,根據系統開發的具體情況進行調整。

===== 🧑‍💼 商業價值評估 ===== 基於 LLM 的 RAG 系統是一個非常有前途的專案,以下是我對其的評估:

1. 商業價值

此專案具有很高的商業價值,因為它可以提供快速和準確的風險評估和管理服務,幫助企業更好地管理風險和達到目標。

2. 目標客群

目標客群包括企業、金融機構、政府機構等,需要風險評估和管理的組織。

3. 解決痛點

此專案可以解決以下痛點:

  • 快速和準確的風險評估和管理
  • 提供風險、行動和目標資訊,幫助企業做出明智的決策
  • 優化系統性能,達到 3 秒內完成回答的速度

4. 市場機會

市場機會包括:

  • 風險管理和評估市場的快速發展
  • 組織對於風險評估和管理的需求日益增長
  • LLM 技術的快速發展和應用

5. ROI

ROI 可能非常高,因為此專案可以幫助企業更好地管理風險和達到目標,從而提高企業的收益和效率。

6. KPI

KPI 可能包括:

  • 系統的使用率和滿意度
  • 風險評估和管理的準確度和速度
  • 組織對於風險評估和管理的滿意度

7. 開發風險

開發風險包括:

  • LLM 引擎的選擇和配置
  • RAG 結構的設計和實現
  • 系統的性能和穩定性
  • 使用者反饋和系統的迭代和優化

8. 是否值得投入

基於 LLM 的 RAG 系統是一個值得投入的專案,因為它具有很高的商業價值和市場機會,同時也能夠解決企業的痛點。

VALUE_SCORE: 9/10

我給予此專案 9 分,因為它具有很高的商業價值和市場機會,同時也能夠解決企業的痛點。但是,我也認為開發風險需要嚴格控制和監測,以確保系統的穩定性和安全性。

===== 👷 Engineer 實作 ===== 根據系統設計,我們可以產出以下內容:

專案資料夾結構

rag-system/
    app/
        __init__.py
        main.py
        models/
            __init__.py
            rag_model.py
            llama_model.py
        routes/
            __init__.py
            rag_routes.py
        utils/
            __init__.py
            llama_utils.py
    requirements.txt
    tests/
        __init__.py
        test_rag_model.py
        test_llama_model.py
    .gitignore
    README.md

Python FastAPI code

# app/main.py
from fastapi import FastAPI
from pydantic import BaseModel
from models.rag_model import RAGModel
from models.llama_model import LlamaModel
from utils.llama_utils import get_llama_response

app = FastAPI()

class RAGRequest(BaseModel):
    prompt: str

@app.post("/rag")
async def get_rag_response(request: RAGRequest):
    llama_model = LlamaModel()
    rag_model = RAGModel()
    response = get_llama_response(request.prompt, llama_model)
    rag_response = rag_model.build_rag_response(response)
    return rag_response
# app/models/rag_model.py
class RAGModel:
    def build_rag_response(self, response):
        # 建立 RAG 結構的資料模型
        rag_response = {
            "risk": response["risk"],
            "action": response["action"],
            "goal": response["goal"]
        }
        return rag_response
# app/models/llama_model.py
class LlamaModel:
    def __init__(self):
        # 設定 LLM 變數,包括模型大小、精度等
        self.model_size = "large"
        self.accuracy = 0.9

    def get_llama_response(self, prompt):
        # 使用 LLM 技術取得答案
        # 這裡假設使用 Hugging Face 的 Transformers 序列化
        from transformers import LLaMAForConditionalGeneration
        import torch

        model = LLaMAForConditionalGeneration.from_pretrained("facebook/llama-large-32k")
        tokenizer = model.tokenizer

        inputs = tokenizer(prompt, return_tensors="pt")
        outputs = model.generate(**inputs, num_beams=4, early_stopping=True)
        response = tokenizer.decode(outputs[0], skip_special_tokens=True)
        return response
# app/utils/llama_utils.py
import torch

def get_llama_response(prompt, llama_model):
    # 取得 LLM 的答案
    response = llama_model.get_llama_response(prompt)
    return response

requirements.txt

fastapi==0.93.0
pydantic==1.9.1
huggingface-transformers==4.21.3
torch==1.10.0

啟動專案指令

python -m venv rag-env
source rag-env/bin/activate
pip install -r requirements.txt
uvicorn app.main:app --port 8000 --host 0.0.0.0

這樣就可以啟動專案並測試 RAG 系統了!

===== 👷 Reviewer 結果 ===== 判斷結果:REJECT

原因:

  1. 系統需求與技術實作不一致:系統需求中要求 3 秒內完成回答,但技術實作中並沒有明確的優化策略來達到這個目標。例如,實際上需要花費更長的時間來訓練 LLM 模型和建立 RAG 結構,導致系統無法在 3 秒內完成回答。
  2. 缺乏明確的開發風險控制和監測:雖然 Business 部分有提到開發風險,但並沒有提供明確的控制和監測措施來確保系統的穩定性和安全性。
  3. 架構包含明顯的幻想:雖然 Engineer 部分提供了一個基本的架構,但並沒有明確的方法來處理 LLM 的回答結果和 RAG 結構的建立。例如,實際上需要花費更多的時間和資源來訓練 LLM 模型和建立 RAG 結構。

建議:

  • 進一步細化和修改需求描述和任務列表,確保系統需求與技術實作一致。
  • 提供明確的開發風險控制和監測措施來確保系統的穩定性和安全性。
  • 追加明確的方法來處理 LLM 的回答結果和 RAG 結構的建立。

通過這些修改和改進,將可以提高系統的可行性和商業價值。

總結

雖然多測幾次後,看起來整體還是有 LLM 太溫和、容易 APPROVE 的問題。1不過本文的重點是學習 LangGraph 的特色用法順便摸一下 Linux,初步體驗後可以感受到它的設定比起 LangChain 有更多能自由發揮的地方,也有點像是 coding 版的 n8n ,可以自由設定節點跟事件。

無符合的項目

腳註

  1. 這個問題遇到好多次了:(,未來在設計系統的時候,可能要再多加思考如何讓容易輕判的 LLM 發揮最大效益。↩︎