add main.py
This commit is contained in:
@@ -1,198 +1,120 @@
|
|||||||
"""
|
# main.py
|
||||||
LangGraph Code Review Agent
|
|
||||||
===========================
|
|
||||||
|
|
||||||
This repository contains a minimal LangGraph implementation that
|
|
||||||
performs a code review on a Python function. The graph consists of
|
|
||||||
three nodes:
|
|
||||||
|
|
||||||
* ``draft_review`` – generates an initial review.
|
|
||||||
* ``reflect`` – a critic that scores the review on four criteria
|
|
||||||
(PEP8, type hints, edge cases, naming) and decides whether a
|
|
||||||
rewrite is required.
|
|
||||||
* ``rewrite`` – rewrites the weakest part of the review.
|
|
||||||
|
|
||||||
The graph runs for a maximum of ``max_rounds`` (default 2). The
|
|
||||||
demo can be executed with ``python -m main``.
|
|
||||||
"""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import os
|
import os
|
||||||
from typing import TypedDict, Dict
|
from typing import TypedDict, Dict
|
||||||
|
|
||||||
from langgraph.graph import StateGraph, END
|
from langgraph.graph import StateGraph, END
|
||||||
from langgraph.prebuilt import create_react_agent
|
from langgraph.prebuilt import create_chat_agent
|
||||||
from langchain_openai import ChatOpenAI
|
from langchain_openai import ChatOpenAI
|
||||||
from langchain_core.messages import HumanMessage
|
from langchain_core.messages import HumanMessage, AIMessage
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
# State definition
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
|
|
||||||
|
# Define state
|
||||||
class CodeReviewState(TypedDict):
|
class CodeReviewState(TypedDict):
|
||||||
code: str
|
code: str
|
||||||
draft_review: str
|
draft_review: str
|
||||||
criteria_scores: Dict[str, int]
|
criteria_scores: Dict[str, int]
|
||||||
weakest_criterion: str
|
weakest_criterion: str
|
||||||
verdict: str # "ok" | "needs_revision"
|
verdict: str
|
||||||
round: int
|
round: int
|
||||||
max_rounds: int
|
max_rounds: int
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# LLM
|
||||||
# LLM configuration
|
llm = ChatOpenAI(temperature=0)
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
|
|
||||||
# The user must set the OPENAI_API_KEY environment variable.
|
|
||||||
llm = ChatOpenAI(temperature=0.0, model="gpt-4o-mini")
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
# Node implementations
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
|
|
||||||
|
# Draft review node
|
||||||
async def draft_review(state: CodeReviewState) -> CodeReviewState:
|
async def draft_review(state: CodeReviewState) -> CodeReviewState:
|
||||||
"""Generate an initial review of the provided code.
|
prompt = f"""
|
||||||
|
You are a senior Python developer. Review the following code and provide a concise code review (3-6 bullet points) highlighting what is good and what can be improved.
|
||||||
|
|
||||||
The review is a short list of 3–6 bullet points.
|
Code:
|
||||||
"""
|
{state['code']}
|
||||||
code = state["code"]
|
|
||||||
prompt = (
|
Review:
|
||||||
"You are a senior Python developer.\n"
|
"""
|
||||||
"Review the following function and provide a concise list of 3–6 points\n"
|
response = await llm.ainvoke([HumanMessage(content=prompt)])
|
||||||
"highlighting what is good and what could be improved.\n"
|
state['draft_review'] = response.content
|
||||||
"Do not mention the criteria – just give the review.\n"
|
|
||||||
f"Function:\n{code}\n"
|
|
||||||
"Review:" # LLM will continue after this
|
|
||||||
)
|
|
||||||
review = await llm.ainvoke([HumanMessage(content=prompt)])
|
|
||||||
state["draft_review"] = review.content.strip()
|
|
||||||
return state
|
return state
|
||||||
|
|
||||||
|
# Reflect node
|
||||||
async def reflect(state: CodeReviewState) -> CodeReviewState:
|
async def reflect(state: CodeReviewState) -> CodeReviewState:
|
||||||
"""Critic that scores the draft review on four criteria.
|
prompt = f"""
|
||||||
|
You are an AI critic evaluating a code review. Assign a score 0-10 for each of the following criteria based on the draft review:
|
||||||
|
- pep8
|
||||||
|
- type_hints
|
||||||
|
- edge_cases
|
||||||
|
- naming
|
||||||
|
|
||||||
The output is a JSON object with keys:
|
Provide a JSON object with keys "pep8", "type_hints", "edge_cases", "naming" and integer values.
|
||||||
* pep8, type_hints, edge_cases, naming – integers 0‑10
|
Also determine the weakest criterion (the one with lowest score) and a verdict: "ok" if all scores >=7, otherwise "needs_revision".
|
||||||
* weakest_criterion – one of the four keys
|
|
||||||
* verdict – "ok" or "needs_revision"
|
Draft review:
|
||||||
"""
|
{state['draft_review']}
|
||||||
review = state["draft_review"]
|
|
||||||
code = state["code"]
|
Output JSON:
|
||||||
prompt = (
|
"""
|
||||||
"You are a code quality critic.\n"
|
|
||||||
"Given the following code and its draft review, score the review on\n"
|
|
||||||
"four criteria: PEP8, type hints, edge cases, naming.\n"
|
|
||||||
"Return a JSON object with keys: pep8, type_hints, edge_cases, naming,\n"
|
|
||||||
"weakest_criterion, verdict.\n"
|
|
||||||
f"Code:\n{code}\n"
|
|
||||||
f"Draft review:\n{review}\n"
|
|
||||||
"Answer in JSON only."
|
|
||||||
)
|
|
||||||
response = await llm.ainvoke([HumanMessage(content=prompt)])
|
response = await llm.ainvoke([HumanMessage(content=prompt)])
|
||||||
import json
|
import json
|
||||||
try:
|
scores = json.loads(response.content)
|
||||||
scores = json.loads(response.content)
|
state['criteria_scores'] = scores
|
||||||
except Exception as e:
|
weakest = min(scores, key=scores.get)
|
||||||
# Fallback: if parsing fails, treat as needs_revision
|
state['weakest_criterion'] = weakest
|
||||||
scores = {
|
state['verdict'] = "ok" if all(v >= 7 for v in scores.values()) else "needs_revision"
|
||||||
"pep8": 0,
|
|
||||||
"type_hints": 0,
|
|
||||||
"edge_cases": 0,
|
|
||||||
"naming": 0,
|
|
||||||
"weakest_criterion": "pep8",
|
|
||||||
"verdict": "needs_revision",
|
|
||||||
}
|
|
||||||
state["criteria_scores"] = {
|
|
||||||
"pep8": int(scores.get("pep8", 0)),
|
|
||||||
"type_hints": int(scores.get("type_hints", 0)),
|
|
||||||
"edge_cases": int(scores.get("edge_cases", 0)),
|
|
||||||
"naming": int(scores.get("naming", 0)),
|
|
||||||
}
|
|
||||||
state["weakest_criterion"] = scores.get("weakest_criterion", "pep8")
|
|
||||||
state["verdict"] = scores.get("verdict", "needs_revision")
|
|
||||||
return state
|
return state
|
||||||
|
|
||||||
|
# Rewrite node
|
||||||
async def rewrite(state: CodeReviewState) -> CodeReviewState:
|
async def rewrite(state: CodeReviewState) -> CodeReviewState:
|
||||||
"""Rewrite the weakest part of the review.
|
crit = state['weakest_criterion']
|
||||||
|
prompt = f"""
|
||||||
|
You are a senior Python developer. Rewrite the section of the code review that addresses the {crit} criterion, improving it. Keep the rest of the review unchanged.
|
||||||
|
|
||||||
The node receives the current state and the weakest criterion.
|
Original review:
|
||||||
It generates a new review that specifically addresses that criterion.
|
{state['draft_review']}
|
||||||
"""
|
|
||||||
code = state["code"]
|
Rewrite only the part related to {crit}:
|
||||||
weakest = state["weakest_criterion"]
|
"""
|
||||||
prompt = (
|
response = await llm.ainvoke([HumanMessage(content=prompt)])
|
||||||
"You are a senior Python developer.\n"
|
# Replace the part in draft_review that mentions crit
|
||||||
"Rewrite the draft review to improve the part related to the following criterion: "
|
# For simplicity, just append the new part
|
||||||
f"{weakest}.\n"
|
state['draft_review'] = state['draft_review'] + "\n" + response.content
|
||||||
"Keep the rest of the review unchanged.\n"
|
state['round'] += 1
|
||||||
"Output only the updated review.\n"
|
|
||||||
f"Current draft review:\n{state['draft_review']}\n"
|
|
||||||
)
|
|
||||||
new_review = await llm.ainvoke([HumanMessage(content=prompt)])
|
|
||||||
state["draft_review"] = new_review.content.strip()
|
|
||||||
state["round"] += 1
|
|
||||||
return state
|
return state
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# Build graph
|
||||||
# Graph construction
|
builder = StateGraph(CodeReviewState)
|
||||||
# ---------------------------------------------------------------------------
|
builder.add_node("draft_review", draft_review)
|
||||||
|
builder.add_node("reflect", reflect)
|
||||||
|
builder.add_node("rewrite", rewrite)
|
||||||
|
|
||||||
def build_graph() -> StateGraph[CodeReviewState]:
|
builder.set_entry_point("draft_review")
|
||||||
graph = StateGraph(CodeReviewState)
|
builder.add_edge("draft_review", "reflect")
|
||||||
graph.add_node("draft_review", draft_review)
|
builder.add_conditional_edges(
|
||||||
graph.add_node("reflect", reflect)
|
"reflect",
|
||||||
graph.add_node("rewrite", rewrite)
|
lambda x: "rewrite" if x["verdict"] == "needs_revision" and x["round"] < x["max_rounds"] else "END",
|
||||||
|
)
|
||||||
|
builder.add_edge("rewrite", "reflect")
|
||||||
|
|
||||||
# Entry point
|
graph = builder.compile()
|
||||||
graph.set_entry_point("draft_review")
|
|
||||||
|
|
||||||
# Transitions
|
# Demo function
|
||||||
graph.add_edge("draft_review", "reflect")
|
async def run_demo():
|
||||||
graph.add_conditional_edges(
|
code = """
|
||||||
"reflect",
|
# Example function to sort numbers
|
||||||
lambda state: "rewrite" if state["verdict"] == "needs_revision" and state["round"] < state["max_rounds"] else "END",
|
|
||||||
)
|
|
||||||
graph.add_edge("rewrite", "reflect")
|
|
||||||
|
|
||||||
return graph
|
def sort_numbers(arr):
|
||||||
|
return sorted(arr)
|
||||||
# ---------------------------------------------------------------------------
|
"""
|
||||||
# Demo execution
|
init_state: CodeReviewState = {
|
||||||
# ---------------------------------------------------------------------------
|
"code": code,
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
import argparse
|
|
||||||
import textwrap
|
|
||||||
|
|
||||||
parser = argparse.ArgumentParser(description="Run the code review graph on a demo function.")
|
|
||||||
parser.add_argument("--max-rounds", type=int, default=2, help="Maximum number of rewrite rounds")
|
|
||||||
args = parser.parse_args()
|
|
||||||
|
|
||||||
# Demo function – can be replaced by any user code
|
|
||||||
demo_code = textwrap.dedent(
|
|
||||||
"""
|
|
||||||
def sort_numbers(arr):
|
|
||||||
return sorted(arr)
|
|
||||||
"""
|
|
||||||
).strip()
|
|
||||||
|
|
||||||
initial_state: CodeReviewState = {
|
|
||||||
"code": demo_code,
|
|
||||||
"draft_review": "",
|
"draft_review": "",
|
||||||
"criteria_scores": {},
|
"criteria_scores": {},
|
||||||
"weakest_criterion": "",
|
"weakest_criterion": "",
|
||||||
"verdict": "",
|
"verdict": "",
|
||||||
"round": 0,
|
"round": 0,
|
||||||
"max_rounds": args.max_rounds,
|
"max_rounds": 2,
|
||||||
}
|
}
|
||||||
|
result = await graph.ainvoke(init_state)
|
||||||
|
print("Final Review:\n", result["draft_review"])
|
||||||
|
print("Scores:\n", result["criteria_scores"])
|
||||||
|
print("Verdict:\n", result["verdict"])
|
||||||
|
|
||||||
graph = build_graph()
|
if __name__ == "__main__":
|
||||||
final_state = graph.invoke(initial_state)
|
import asyncio
|
||||||
|
asyncio.run(run_demo())
|
||||||
print("\n=== Final Review ===")
|
|
||||||
print(final_state["draft_review"])
|
|
||||||
print("\n=== Scores ===")
|
|
||||||
for k, v in final_state["criteria_scores"].items():
|
|
||||||
print(f"{k}: {v}")
|
|
||||||
print(f"Verdict: {final_state['verdict']}")
|
|
||||||
print(f"Rounds: {final_state['round']}")
|
|
||||||
|
|||||||
Reference in New Issue
Block a user