Day 4 我們有了 Global Constraints State。但 State 只是數據結構。怎麼讓它實際執行?
最簡單的做法是寫一個大函數:
srs_text = read_file()
chunks = chunk(srs_text)
constraints = extract(chunks)
conflicts = detect(constraints)
report = generate(conflicts)
問題是,後來你想加條件分支(「如果矛盾超過 10 個,進行詳細分析」),或者你想在某個環節插入快取、重試、LLM 降級,整個函數就變成一個 if/else 的迷宮。
LangGraph 解決這個問題。它用「狀態機」的思路組織工作流:定義節點(做什麼),定義邊(流向哪裡),State 在節點間自動流轉。
節點(Node)— 執行任務的函數:
def extract_constraints(state: LangGraphState) -> LangGraphState:
# 拿到 state,修改它,返回它
state["global_constraints"].append(...)
return state
邊(Edge)— 連接節點:
graph.add_edge("extract", "detect") # extract 完成後進入 detect
條件邊(Conditional Edge)— 根據狀態決定下一步:
def check_conflict_count(state: LangGraphState):
if len(state["conflicts"]) > 10:
return "detail_analysis"
return "report"
graph.add_conditional_edges("detect", check_conflict_count)
就這麼簡單。
五個階段:
[讀取 & 分塊]
↓ state 初始化
[提取約束](extract_constraints 節點)
遍歷每個 chunk,用 LLM 提取 REQ-xxx
↓ 約束池增長
[檢測衝突](detect_conflicts 節點)
比對全局約束,找矛盾
↓ 矛盾列表增長
[條件分支]
矛盾超過 10 個?
├─ YES → [詳細分析]
└─ NO → [快速報告]
↓
[生成報告](generate_report 節點)
統計、分類、輸出 Markdown
↓ 完成
這是關鍵節點。約束提取的品質決定了整個系統的品質。
def extract_constraints(state: LangGraphState) -> LangGraphState:
"""遍歷每個 chunk,用 LLM 提取需求"""
state["analysis_trace"].append("[extract] 開始")
for chunk_id, chunk in enumerate(state["chunks"]):
# 用 LLM 提取
prompt = f"""
從以下文本提取所有需求。
格式:REQ-編號|類型|文本|依賴1,依賴2
{chunk.content}
"""
llm_response = call_ollama(prompt)
constraints = parse_constraints(llm_response, chunk_id)
# 加入全局池
state["global_constraints"].extend(constraints)
state["analysis_trace"].append(
f"[extract] Chunk {chunk_id}: 提取 {len(constraints)} 個"
)
return state
為什麼關鍵?因為這裡是 LLM 直接參與的地方。Prompt 的品質、Parser 的健壯性、錯誤恢復——全都影響下游。
兩種策略並用:
def detect_conflicts(state: LangGraphState) -> LangGraphState:
"""檢測約束間的矛盾"""
constraints = state["global_constraints"]
# 策略 1:規則匹配(快)
# 硬編碼的模式:「多用戶」vs「單用戶」、「加密」vs「明文」
for i, c1 in enumerate(constraints):
for c2 in constraints[i+1:]:
if detect_by_rules(c1, c2):
state["conflicts"].append(Conflict(
req_id_1=c1.id,
req_id_2=c2.id,
conflict_type=ConflictType.LOGIC,
severity=Severity.HIGH,
))
# 策略 2:LLM 推理(精確但慢)
# 只對高價值對進行
if len(state["conflicts"]) == 0:
# 用 LLM 做細緻比對
for c1 in constraints[:5]:
prompt = f"需求 {c1.id} 與其他需求有無衝突?\n..."
response = call_ollama(prompt)
# 解析並記錄
return state
實際上 O(n²) 的對比。對 45 個約束來說,大約 1000 次對比。規則層先過篩,LLM 層只檢查疑似對。
簡單的統計和 Markdown 格式化:
def generate_report(state: LangGraphState) -> LangGraphState:
"""生成 Markdown 報告"""
high = len([c for c in state["conflicts"] if c.severity == Severity.HIGH])
medium = len([c for c in state["conflicts"] if c.severity == Severity.MEDIUM])
report = f"""# SRS 審查報告
## 概覽
- 約束數:{len(state['global_constraints'])}
- 矛盾數:{len(state['conflicts'])} (高:{high}, 中:{medium})
## 矛盾清單
"""
for conflict in state["conflicts"]:
report += f"\n### {conflict.req_id_1} vs {conflict.req_id_2}\n{conflict.description}\n"
return state
src/workflow.py:
from langgraph.graph import StateGraph, END
from src.models import LangGraphState, Constraint, Conflict, ConflictType, Severity
def create_srs_review_workflow():
"""創建 SRS 審查工作流"""
workflow = StateGraph(LangGraphState)
# 定義節點
workflow.add_node("extract", extract_constraints)
workflow.add_node("detect", detect_conflicts)
workflow.add_node("report", generate_report)
# 定義邊
workflow.add_edge("extract", "detect")
# 條件邊:根據矛盾數決定是否詳細分析
def should_detail(state: LangGraphState):
return "detail" if len(state["conflicts"]) > 10 else "report"
workflow.add_conditional_edges("detect", should_detail, {
"report": "report",
"detail": "report", # 簡化版,都進入 report
})
workflow.add_edge("report", END)
workflow.set_entry_point("extract")
return workflow.compile()
def analyze_srs_with_workflow(srs_text: str) -> LangGraphState:
"""執行工作流"""
initial_state: LangGraphState = {
"srs_file_path": "input.md",
"srs_content": srs_text,
"chunks": [], # 從 chunking 模塊獲得
"current_chunk_id": 0,
"global_constraints": [],
"conflicts": [],
"analysis_trace": ["[workflow] 初始化"],
"processing_status": "初始化中",
"error_message": None,
}
workflow = create_srs_review_workflow()
return workflow.invoke(initial_state)
在 src/main.py 加入:
@app.command()
def analyze_workflow(
filepath: str = typer.Argument(..., help="SRS 檔案路徑"),
model: str = typer.Option("mistral", "--model", "-m"),
verbose: bool = typer.Option(False, "--verbose", "-v"),):
"""使用 LangGraph 工作流分析 SRS"""
try:
from src.workflow import analyze_srs_with_workflow
agent = OllamaAgent(model=model)
srs_text = agent.read_srs(filepath)
print(f"📖 讀取 {filepath}")
print(f"🔄 執行工作流...")
final_state = analyze_srs_with_workflow(srs_text)
print(f"\n✅ 完成")
print(f" 約束:{len(final_state['global_constraints'])}")
print(f" 矛盾:{len(final_state['conflicts'])}")
for conflict in final_state["conflicts"]:
print(f" - {conflict.req_id_1} vs {conflict.req_id_2}")
if verbose:
print("\n執行追蹤:")
for trace in final_state["analysis_trace"]:
print(f" {trace}")
except Exception as e:
print(f"❌ 錯誤: {e}")
raise typer.Exit(1)
執行:
python -m src.main analyze-workflow tests/fixtures/srs_medium.md
簡單說:清晰 + 可追蹤 + 可擴展。
清晰:節點和邊一眼看出邏輯流。不是一堆嵌套的 if/else。
可追蹤:analysis_trace 記錄每一步。出問題時,你知道卡在哪。
可擴展:想加錯誤重試?加新節點。想加快取?改提取節點。不需要改其他地方。
Day 6 我們會加 MCP 工具,讓 LLM 直接調用這些節點,而不是硬編碼調用。
第 1 週:透明的 SRS 分析