bowei/skills-win
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

更新了langgraph-plan-view，支持windows下使用 langgraph cli；更新了langgraph-vivado-install，要求安装langgraph cli，确定conda环境的名称为langgraph

Browse Source
This commit is contained in:
Myfro9
2026-05-25 13:13:00 +08:00
parent fda6c9a114
commit 48b965ec4e
9 changed files with 980 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
langgraph-plan-review/scripts/run_workflow.py
+241 -128
View File
@@ -1,3 +1,12 @@
"""run_workflow.py — submit workflow runs to the local langgraph dev server.
Instead of running the graph directly in-process, every invocation sends a
run to the langgraph dev server (http://127.0.0.1:2024) via the LangGraph
SDK sync client. This lets LangGraph Studio track execution in real time.
The Codex review command is forwarded in the run's config so the server does
not need to be restarted when the command changes between projects.
"""
from __future__ import annotations
import argparse
@@ -5,63 +14,68 @@ import base64
import json
import logging
import sys
import time
from pathlib import Path
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import Command
from agents import AgentRunner
from workflow import build_workflow, resolve_notes_dir, resolve_work_dir
SERVER_URL = "http://127.0.0.1:2024"
ASSISTANT_ID = "plan_review"
THREAD_MAP_FILENAME = "workflow_threads.json"
DB_FILENAME = "workflow_state.db"
# ── path helpers ──────────────────────────────────────────────────────────────
def _resolve_work_dir(value: str) -> Path:
return Path(value or ".").resolve()
def parse_command(value: str | None) -> list[str] | None:
if not value:
return None
try:
parsed = json.loads(value)
except json.JSONDecodeError:
# Fallback: the string may contain Windows paths with unescaped backslashes
# that make it invalid JSON. Re-escape backslashes and retry before giving up.
def _resolve_notes_dir(work_dir: Path, value: str) -> Path:
notes_dir = Path(value or "Notes")
if not notes_dir.is_absolute():
notes_dir = work_dir / notes_dir
return notes_dir.resolve()
# ── thread map (task_id → server thread_id) ──────────────────────────────────
def _load_thread_map(notes_dir: Path) -> dict:
path = notes_dir / THREAD_MAP_FILENAME
if path.exists():
try:
import re
# Replace bare backslashes (not already doubled) with \\
re_escaped = re.sub(r'\\(?![\\"/bfnrtu])', r'\\\\', value)
parsed = json.loads(re_escaped)
except json.JSONDecodeError:
stripped = value.strip()
if stripped.startswith("[") and stripped.endswith("]"):
# Last-resort comma split; strip surrounding quotes from each item.
parsed = [item.strip().strip('"').strip("'")
for item in stripped[1:-1].split(",")]
else:
raise
if not isinstance(parsed, list) or not all(isinstance(item, str) for item in parsed):
raise argparse.ArgumentTypeError("command must be a JSON string array")
return parsed
return json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return {}
return {}
def parse_command_b64(value: str | None) -> list[str] | None:
if not value:
return None
try:
decoded = base64.b64decode(value).decode("utf-8")
except Exception as exc:
raise argparse.ArgumentTypeError(f"invalid base64 command: {exc}") from exc
return parse_command(decoded)
def _save_thread_map(notes_dir: Path, mapping: dict) -> None:
path = notes_dir / THREAD_MAP_FILENAME
path.write_text(json.dumps(mapping, indent=2, ensure_ascii=False), encoding="utf-8")
def parse_json_b64(value: str | None):
if not value:
return None
try:
decoded = base64.b64decode(value).decode("utf-8")
except Exception as exc:
raise argparse.ArgumentTypeError(f"invalid base64 JSON: {exc}") from exc
return json.loads(decoded)
# ── interrupt extraction ──────────────────────────────────────────────────────
def _extract_interrupt(state: dict):
"""Return the first interrupt value from a thread-state dict, or None."""
# Primary: tasks[].interrupts (langgraph-api >= 0.5)
for task in state.get("tasks", []):
for intr in task.get("interrupts", []):
if isinstance(intr, dict):
return intr.get("value", intr)
return intr
# Fallback: __interrupt__ channel value
interrupt_list = state.get("values", {}).get("__interrupt__", [])
if interrupt_list:
intr = interrupt_list[0]
return intr.value if hasattr(intr, "value") else intr
return None
def _is_interrupted(state: dict) -> bool:
"""Return True when the thread is paused at an interrupt."""
return bool(state.get("next")) and _extract_interrupt(state) is not None
# ── core workflow runner ──────────────────────────────────────────────────────
def run_workflow(
*,
@@ -69,113 +83,210 @@ def run_workflow(
user_prompt: str = "",
notes_dir: str = "Notes",
work_dir: str = "",
runner: AgentRunner,
codex_review_command: list[str],
thread_id: str | None = None,
resume_payload=None,
timeout: int = 600,
server_url: str = SERVER_URL,
) -> dict:
effective_thread = thread_id or task_id
resolved_work_dir = resolve_work_dir(work_dir)
resolved_notes_dir = resolve_notes_dir(resolved_work_dir, notes_dir)
try:
from langgraph_sdk import get_sync_client
except ImportError:
raise ImportError(
"langgraph_sdk not installed. Run: "
"conda run -n langgraph pip install langgraph-sdk"
)
import httpx
effective_task_id = thread_id or task_id
resolved_work_dir = _resolve_work_dir(work_dir)
resolved_notes_dir = _resolve_notes_dir(resolved_work_dir, notes_dir)
resolved_notes_dir.mkdir(parents=True, exist_ok=True)
db_path = resolved_notes_dir / DB_FILENAME
with SqliteSaver.from_conn_string(str(db_path)) as checkpointer:
app = build_workflow(runner, checkpointer=checkpointer)
config = {"configurable": {"thread_id": effective_thread}}
# ── verify server is up ───────────────────────────────────────────────────
try:
httpx.get(f"{server_url}/ok", timeout=5).raise_for_status()
except Exception as exc:
raise RuntimeError(
f"langgraph dev server not reachable at {server_url}.\n"
f"Start it first:\n"
f" PowerShell -File C:\\Users\\bowei\\.claude\\skills\\langgraph-plan-review\\scripts\\manage_dev_server.ps1 start\n"
f"Error: {exc}"
)
if resume_payload is not None:
result = app.invoke(Command(resume=resume_payload), config=config)
else:
result = app.invoke(
{
"task_id": task_id,
"user_prompt": user_prompt,
"notes_dir": str(resolved_notes_dir),
"work_dir": str(resolved_work_dir),
},
config=config,
client = get_sync_client(url=server_url)
# ── resolve server thread_id ──────────────────────────────────────────────
thread_map = _load_thread_map(resolved_notes_dir)
server_thread_id: str | None = thread_map.get(effective_task_id)
# Verify the stored thread still exists on the server (server may have restarted)
if server_thread_id:
try:
client.threads.get(server_thread_id)
except Exception:
logging.warning(
"Stored thread %s no longer exists on server (server restarted?); "
"creating a new thread.",
server_thread_id,
)
server_thread_id = None
interrupts = result.get("__interrupt__", [])
if interrupts:
intr = interrupts[0]
action_required = intr.value if hasattr(intr, "value") else intr
if not server_thread_id:
thread = client.threads.create(
metadata={"task_id": effective_task_id}
)
server_thread_id = thread["thread_id"]
thread_map[effective_task_id] = server_thread_id
_save_thread_map(resolved_notes_dir, thread_map)
# ── submit run ────────────────────────────────────────────────────────────
run_config = {
"configurable": {
"codex_review_command": codex_review_command,
"codex_timeout": timeout,
}
}
if resume_payload is not None:
run = client.runs.create(
server_thread_id,
ASSISTANT_ID,
command={"resume": resume_payload},
config=run_config,
)
else:
run = client.runs.create(
server_thread_id,
ASSISTANT_ID,
input={
"task_id": effective_task_id,
"user_prompt": user_prompt,
"notes_dir": str(resolved_notes_dir),
"work_dir": str(resolved_work_dir),
},
config=run_config,
)
run_id = run["run_id"]
logging.info("Submitted run %s on thread %s", run_id, server_thread_id)
# ── wait for completion or interrupt ──────────────────────────────────────
ACTIVE = {"pending", "running"}
deadline = time.monotonic() + timeout
run_status = run.get("status", "pending")
while run_status in ACTIVE and time.monotonic() < deadline:
time.sleep(1)
try:
final_run = client.runs.get(server_thread_id, run_id)
run_status = final_run.get("status", run_status)
except Exception as exc:
logging.debug("runs.get error (will retry): %s", exc)
if run_status in ACTIVE:
raise TimeoutError(
f"Run {run_id} did not complete within {timeout}s (status={run_status})"
)
# ── detect interrupt from thread state ────────────────────────────────────
state = client.threads.get_state(server_thread_id)
if _is_interrupted(state):
interrupt_value = _extract_interrupt(state)
return {
"status": "waiting",
"thread_id": effective_thread,
"db_path": str(db_path),
"action_required": action_required,
"thread_id": effective_task_id,
"server_thread_id": server_thread_id,
"action_required": interrupt_value,
}
if run_status not in ("success",):
error = final_run.get("error") or run_status
raise RuntimeError(f"Run ended with status '{run_status}': {error}")
# ── extract final state ───────────────────────────────────────────────────
values = state.get("values", {})
return {
"status": "done",
"thread_id": effective_thread,
"db_path": str(db_path),
"task_path": result.get("task_path", ""),
"latest_discuss_path": result.get("latest_discuss_path", ""),
"work_dir": result.get("work_dir", ""),
"codex_session_id": result.get("codex_session_id", ""),
"has_serious_issue": result.get("has_serious_issue", False),
"should_exit": result.get("should_exit", True),
"thread_id": effective_task_id,
"server_thread_id": server_thread_id,
"task_path": values.get("task_path", ""),
"latest_discuss_path": values.get("latest_discuss_path", ""),
"work_dir": values.get("work_dir", ""),
"codex_session_id": values.get("codex_session_id", ""),
"has_serious_issue": values.get("has_serious_issue", False),
"should_exit": values.get("should_exit", True),
}
# ── CLI ───────────────────────────────────────────────────────────────────────
def _parse_command(value: str | None) -> list[str] | None:
if not value:
return None
try:
parsed = json.loads(value)
except json.JSONDecodeError:
import re
try:
parsed = json.loads(re.sub(r'\\(?![\\"/bfnrtu])', r'\\\\', value))
except json.JSONDecodeError:
stripped = value.strip()
if stripped.startswith("[") and stripped.endswith("]"):
parsed = [i.strip().strip('"').strip("'") for i in stripped[1:-1].split(",")]
else:
raise
if not isinstance(parsed, list) or not all(isinstance(i, str) for i in parsed):
raise argparse.ArgumentTypeError("command must be a JSON string array")
return parsed
def _parse_b64(value: str | None) -> list[str] | None:
if not value:
return None
return _parse_command(base64.b64decode(value).decode("utf-8"))
def _parse_json_b64(value: str | None):
if not value:
return None
return json.loads(base64.b64decode(value).decode("utf-8"))
def main() -> int:
parser = argparse.ArgumentParser(
description="Run the LangGraph Claude/Codex review workflow."
)
parser.add_argument("--task-id", required=True, help="Task id, for example 02")
parser.add_argument("--prompt", default="", help="User request recorded for traceability")
parser.add_argument("--notes-dir", default="Notes", help="Directory for Task/discuss files")
parser.add_argument("--work-dir", default="", help="Claude Code launch working directory")
parser.add_argument(
"--codex-review-command",
type=parse_command,
default=None,
help="JSON argv for Codex review step.",
)
parser.add_argument(
"--codex-review-command-b64",
type=parse_command_b64,
default=None,
help="Base64-encoded JSON argv for Codex review step.",
description="Submit a plan-review workflow run to the local langgraph dev server."
)
parser.add_argument("--task-id", required=True)
parser.add_argument("--prompt", default="")
parser.add_argument("--notes-dir", default="Notes")
parser.add_argument("--work-dir", default="")
parser.add_argument("--codex-review-command", type=_parse_command, default=None)
parser.add_argument("--codex-review-command-b64", type=_parse_b64, default=None)
parser.add_argument("--timeout", type=int, default=600)
parser.add_argument(
"--thread-id",
default="",
help="Thread ID for resuming an existing run (defaults to task-id)",
)
parser.add_argument(
"--resume",
default="",
help=(
"JSON payload for Command(resume=...). "
"Use true for plan/revise/rejection steps; {\"accepts\":true} for decision step."
),
)
parser.add_argument(
"--resume-b64",
default="",
help="Base64-encoded JSON payload for Command(resume=...).",
)
parser.add_argument("--verbose", "-v", action="store_true", help="Enable debug logging to stderr.")
parser.add_argument("--thread-id", default="")
parser.add_argument("--resume", default="")
parser.add_argument("--resume-b64", default="")
parser.add_argument("--server-url", default=SERVER_URL)
parser.add_argument("--verbose", "-v", action="store_true")
args = parser.parse_args()
if args.verbose:
logging.basicConfig(level=logging.DEBUG, stream=sys.stderr,
format="%(asctime)s %(levelname)s %(message)s")
codex_review_command = args.codex_review_command_b64 or args.codex_review_command
if not codex_review_command:
logging.basicConfig(
level=logging.DEBUG, stream=sys.stderr,
format="%(asctime)s %(levelname)s %(message)s",
)
codex_cmd = args.codex_review_command_b64 or args.codex_review_command
if not codex_cmd:
parser.error("--codex-review-command or --codex-review-command-b64 is required")
runner = AgentRunner(
codex_review_command=codex_review_command,
timeout=args.timeout,
)
# resume=True tells LangGraph to resume from the last interrupt without a specific value.
# Callers should pass 'true' (JSON boolean) for plan/revise/rejection steps.
resume_payload = parse_json_b64(args.resume_b64) if args.resume_b64 else (
json.loads(args.resume) if args.resume else None
resume_payload = (
_parse_json_b64(args.resume_b64) if args.resume_b64
else (json.loads(args.resume) if args.resume else None)
)
output = run_workflow(
@@ -183,9 +294,11 @@ def main() -> int:
user_prompt=args.prompt,
notes_dir=args.notes_dir,
work_dir=args.work_dir,
runner=runner,
codex_review_command=codex_cmd,
thread_id=args.thread_id or None,
resume_payload=resume_payload,
timeout=args.timeout,
server_url=args.server_url,
)
print(json.dumps(output, ensure_ascii=False, indent=2))