diff --git a/README.md b/README.md index b80791c..563356d 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ 一个面向软件仓库的工程变更分析与故障诊断 Agent。它把架构文档、PR 说明、日志、Runbook 和评测结果组织成可检索证据，并围绕变更影响、PR 风险、故障假设、验证计划和回滚步骤给出有边界的工程结论。 -系统核心不是普通问答，而是一条可规划、可执行、可验证、可学习的 Agent 控制流：外层 State Graph 管理流程边界，内层 Plan → Action → Observation → Reflection 负责动态工具决策；Verification Loop 在证据不足时有界重试，并保留断点、继续、编辑后重跑与终止能力。 +系统核心不是普通问答，而是一条可规划、可执行、可验证、可学习的 Agent 控制流：外层由 **LangGraph StateGraph** 管理流程边界、条件路由与 Checkpoint，内层 Plan → Action → Observation → Reflection 负责动态工具决策；Verification Loop 在证据不足时有界重试，并保留断点、继续、编辑后重跑与终止能力。 ## 主要能力 @@ -15,6 +15,7 @@ - **生产安全边界**：写操作进入 Human Approval Gate；Tool Registry 提供 idempotency key、执行状态与补偿 rollback 回调，避免重复副作用。 - **证据门控**：检查原问题与命中材料是否有直接领域信号；低证据时不调用外部 LLM。 - **有界恢复**：Query Rewrite、检索预算和全局图转换上限共同防止无界循环。 +- **LangGraph 原生编排**：9 个业务节点直接注册到 LangGraph，条件边负责证据重试与 ReAct 循环，`InMemorySaver` + `interrupt/Command` 负责断点和恢复，不再维护第二套自研图执行器。 - **执行轨迹**：Graph Path、Tool Trace、检索次数、实际 Reranker、证据质量和停止原因均可观察。 - **实时运行事件**：`/api/ask/stream` 通过 SSE 在节点完成时推送 Route、Retrieve、Tool、Reflection 与 Final 事件，而不是等整次请求结束后一次性返回。 - **可视化调试**：节点断点、内存 Checkpoint、继续、编辑查询后从 Route 重启、终止与清理。 @@ -53,7 +54,7 @@ Debug: breakpoint → checkpoint → resume / edit and restart / cancel - Backend：Python、FastAPI、Pydantic、Uvicorn - Retrieval：SentenceTransformers、BGE、Dense Cosine Search、Chroma、Cross-Encoder -- Agent：State Graph、ReAct、Planning / Reflection、Verification Loop、Skill、Native/MCP Tool Registry +- Agent：LangGraph、StateGraph、Checkpoint / Interrupt、ReAct、Planning / Reflection、Verification Loop、Skill、Native/MCP Tool Registry - Memory：Working、Episodic、Semantic、Procedural、幂等 Memory Import - Safety：Human Approval、Idempotency、Rollback、Bounded Retry / Stop Reason - LLM：OpenAI-compatible Chat API Adapter（可选） @@ -148,7 +149,7 @@ POST /api/debug/resume DELETE /api/debug/runs/{run_id} ``` -`/api/ask` 返回 `answer`、`intent`、`citations`、`trace` 和 `metrics`。调试接口在节点执行前暂停并保留 Checkpoint；完成或取消后清理。 +`/api/ask` 返回 `answer`、`intent`、`citations`、`trace` 和 `metrics`；`/api/workflow` 会明确返回 `framework=langgraph`。调试接口通过 LangGraph interrupt 在节点执行前暂停并保留 Checkpoint，使用相同 thread 恢复；完成或取消后清理。 ## 测试与评估 diff --git a/app/agent.py b/app/agent.py index 85d9018..2303c9f 100644 --- a/app/agent.py +++ b/app/agent.py @@ -325,6 +325,9 @@ def __init__(self, kb: KnowledgeBase) -> None: def workflow_spec(self) -> dict[str, Any]: return { + "framework": self.workflow.framework, + "checkpointing": "langgraph_in_memory", + "interrupts": "langgraph_dynamic", "entrypoint": "route", "nodes": [ "route", @@ -374,7 +377,9 @@ def ask( sid = self.memory.ensure(session_id) state = self._new_state(query, sid, top_k) graph_run = self.workflow.run(state, event_sink=event_sink) - return self._build_response(sid, state, graph_run, started, record_memory=True) + response = self._build_response(sid, state, graph_run, started, record_memory=True) + self.workflow.cancel(graph_run.thread_id) + return response def debug_start( self, @@ -419,6 +424,7 @@ def debug_resume( next_query = (query or checkpoint.query).strip() if restart: + self.workflow.cancel(checkpoint.run.thread_id) state = self._new_state(next_query, checkpoint.session_id, checkpoint.top_k) state.trace.append( ToolCall( @@ -440,6 +446,7 @@ def debug_resume( prior_events=checkpoint.run.events, breakpoints=active_breakpoints, skip_breakpoint_once=checkpoint.run.next_node, + thread_id=checkpoint.run.thread_id, ) return self._finish_debug_step( @@ -455,7 +462,11 @@ def debug_resume( def cancel_debug(self, run_id: str) -> bool: with self._debug_lock: - return self._debug_runs.pop(run_id, None) is not None + checkpoint = self._debug_runs.pop(run_id, None) + if checkpoint is None: + return False + self.workflow.cancel(checkpoint.run.thread_id) + return True def _new_state(self, query: str, session_id: str, top_k: int) -> AgentState: normalized_query = self._resolve_follow_up(query, session_id) @@ -495,6 +506,7 @@ def _finish_debug_step( record_memory=completed, ) if completed: + self.workflow.cancel(graph_run.thread_id) with self._debug_lock: self._debug_runs.pop(run_id, None) else: diff --git a/app/langgraph_adapter.py b/app/langgraph_adapter.py deleted file mode 100644 index 55f707d..0000000 --- a/app/langgraph_adapter.py +++ /dev/null @@ -1,98 +0,0 @@ -"""LangGraph runtime adapter for RepoPilot. - -The existing workflow implementation remains the source of node logic. This -module only provides a LangGraph execution layer so RepoPilot can expose a -standard Agent workflow with explicit state transitions. -""" - -from typing import Any, TypedDict - -from langgraph.graph import END, StateGraph - - -class RepoPilotGraphState(TypedDict, total=False): - query: str - plan: dict[str, Any] - evidence: list[Any] - observations: list[Any] - tool_results: list[Any] - reflection: dict[str, Any] - answer: str - error: str - - -def _verify_route(state: RepoPilotGraphState) -> str: - """Route retrieval failures back to retrieval instead of hallucinating.""" - - return "tool" if state.get("evidence") else "retrieve" - - -def _reflection_route(state: RepoPilotGraphState) -> str: - """Continue tool investigation until reflection marks the task complete.""" - - reflection = state.get("reflection", {}) - return "synthesize" if reflection.get("done") else "tool" - - -def build_langgraph_workflow( - planner, - retriever, - verifier, - tool_executor, - reflector, - synthesizer, -): - """Build RepoPilot's Planner-Retrieve-Tool-Reflection graph.""" - - graph = StateGraph(RepoPilotGraphState) - - graph.add_node("planner", planner) - graph.add_node("retrieve", retriever) - graph.add_node("verify", verifier) - graph.add_node("tool", tool_executor) - graph.add_node("reflect", reflector) - graph.add_node("synthesize", synthesizer) - - graph.set_entry_point("planner") - - graph.add_edge("planner", "retrieve") - graph.add_edge("retrieve", "verify") - - graph.add_conditional_edges( - "verify", - _verify_route, - { - "tool": "tool", - "retrieve": "retrieve", - }, - ) - - graph.add_edge("tool", "reflect") - - graph.add_conditional_edges( - "reflect", - _reflection_route, - { - "synthesize": "synthesize", - "tool": "tool", - }, - ) - - graph.add_edge("synthesize", END) - - return graph.compile() - - -def run_langgraph_agent(graph, query: str, **kwargs: Any) -> dict[str, Any]: - """Run RepoPilot through LangGraph with extensible runtime options.""" - - initial_state: RepoPilotGraphState = { - "query": query, - "evidence": [], - "observations": [], - "tool_results": [], - } - initial_state.update(kwargs) - - result = graph.invoke(initial_state) - return dict(result) diff --git a/app/langgraph_bridge.py b/app/langgraph_bridge.py deleted file mode 100644 index c1ab95f..0000000 --- a/app/langgraph_bridge.py +++ /dev/null @@ -1,40 +0,0 @@ -"""LangGraph execution bridge for RepoPilot. - -Keeps the existing KnowledgeAgent nodes reusable while exposing a LangGraph -workflow for Agent applications. -""" - -from __future__ import annotations - -from typing import Any, TypedDict - -from langgraph.graph import END, StateGraph - - -class RepoPilotGraphState(TypedDict, total=False): - query: str - context: list[Any] - observations: list[Any] - answer: str - - -def build_langgraph_bridge(agent: Any): - """Build a LangGraph workflow backed by existing RepoPilot capabilities.""" - - graph = StateGraph(RepoPilotGraphState) - - def retrieve(state: RepoPilotGraphState): - result = agent.kb.build_context(state["query"], top_k=5) - return {"context": result.hits} - - def synthesize(state: RepoPilotGraphState): - answer = agent._synthesize_answer(state["query"], state.get("context", [])) - return {"answer": answer} - - graph.add_node("retrieve", retrieve) - graph.add_node("synthesize", synthesize) - graph.set_entry_point("retrieve") - graph.add_edge("retrieve", "synthesize") - graph.add_edge("synthesize", END) - - return graph.compile() diff --git a/app/langgraph_entrypoint.py b/app/langgraph_entrypoint.py deleted file mode 100644 index 92b4916..0000000 --- a/app/langgraph_entrypoint.py +++ /dev/null @@ -1,77 +0,0 @@ -"""LangGraph execution entrypoint for RepoPilot. - -Provides the LangGraph orchestration boundary while keeping existing business -nodes injectable. The existing agent modules can migrate incrementally by -passing their node functions here. -""" - -from typing import Any, Callable - -from langgraph.graph import END, StateGraph -from typing_extensions import TypedDict - - -class RepoPilotRuntimeState(TypedDict, total=False): - query: str - evidence: list[Any] - observations: list[Any] - answer: str - reflection: dict[str, Any] - trace: list[dict[str, Any]] - - -Node = Callable[[RepoPilotRuntimeState], RepoPilotRuntimeState] - - -def build_repopilot_graph( - planner: Node, - retrieve: Node, - verify: Node, - execute: Node, - reflect: Node, - answer: Node, -): - """Build RepoPilot's LangGraph workflow. - - Nodes remain injected so existing retrieval/tools/memory implementations - can be reused without rewriting them into LangChain abstractions. - """ - graph = StateGraph(RepoPilotRuntimeState) - - for name, node in { - "planner": planner, - "retrieve": retrieve, - "verify": verify, - "execute": execute, - "reflect": reflect, - "answer": answer, - }.items(): - graph.add_node(name, node) - - graph.set_entry_point("planner") - graph.add_edge("planner", "retrieve") - graph.add_edge("retrieve", "verify") - - graph.add_conditional_edges( - "verify", - lambda state: "execute" if state.get("evidence") else "retrieve", - {"execute": "execute", "retrieve": "retrieve"}, - ) - - graph.add_edge("execute", "reflect") - graph.add_conditional_edges( - "reflect", - lambda state: "answer" - if state.get("reflection", {}).get("finished") - else "execute", - {"answer": "answer", "execute": "execute"}, - ) - - graph.add_edge("answer", END) - return graph.compile() - - -def run(graph, query: str, **kwargs: Any) -> dict[str, Any]: - """Execute a compiled RepoPilot LangGraph workflow.""" - payload: RepoPilotRuntimeState = {"query": query, **kwargs} - return dict(graph.invoke(payload)) diff --git a/app/langgraph_runtime.py b/app/langgraph_runtime.py deleted file mode 100644 index 87506d0..0000000 --- a/app/langgraph_runtime.py +++ /dev/null @@ -1,74 +0,0 @@ -"""Runtime bridge between RepoPilot workflow nodes and LangGraph. - -The existing workflow keeps its business logic. This module only adapts -node callables into LangGraph compatible state transitions. -""" - -from typing import Any, Callable, TypedDict - -from langgraph.graph import END, StateGraph - - -class RepoPilotState(TypedDict, total=False): - query: str - plan: dict[str, Any] - context: list[Any] - evidence: list[Any] - observations: list[Any] - reflection: dict[str, Any] - answer: str - error: str - - -Node = Callable[[RepoPilotState], RepoPilotState] - - -def build_runtime( - planner: Node, - retriever: Node, - verifier: Node, - executor: Node, - reflector: Node, - responder: Node, -): - graph = StateGraph(RepoPilotState) - - graph.add_node("planner", planner) - graph.add_node("retriever", retriever) - graph.add_node("verifier", verifier) - graph.add_node("executor", executor) - graph.add_node("reflector", reflector) - graph.add_node("responder", responder) - - graph.set_entry_point("planner") - graph.add_edge("planner", "retriever") - graph.add_edge("retriever", "verifier") - - graph.add_conditional_edges( - "verifier", - lambda state: "executor" if state.get("evidence") else "retriever", - { - "executor": "executor", - "retriever": "retriever", - }, - ) - - graph.add_edge("executor", "reflector") - graph.add_conditional_edges( - "reflector", - lambda state: "responder" - if state.get("reflection", {}).get("finished") - else "executor", - { - "responder": "responder", - "executor": "executor", - }, - ) - - graph.add_edge("responder", END) - return graph.compile() - - -def invoke_runtime(graph, query: str, **kwargs: Any) -> dict[str, Any]: - state: RepoPilotState = {"query": query, **kwargs} - return dict(graph.invoke(state)) diff --git a/app/workflow.py b/app/workflow.py index ace39cd..8f7d689 100644 --- a/app/workflow.py +++ b/app/workflow.py @@ -1,17 +1,27 @@ from __future__ import annotations +import copy +import operator +import threading import time -from dataclasses import dataclass, field -from typing import Any, Callable +import uuid +from contextvars import ContextVar +from dataclasses import dataclass, field, fields +from typing import Annotated, Any, Callable, TypedDict -from .rag_engine import SearchHit +from langgraph.checkpoint.memory import InMemorySaver +from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer +from langgraph.errors import GraphRecursionError +from langgraph.graph import END, START +from langgraph.graph import StateGraph as LangGraphStateGraph +from langgraph.types import Command, interrupt -END = "__end__" +from .rag_engine import SearchHit @dataclass class AgentState: - """Mutable state shared by every node in the agent workflow graph.""" + """Business state shared by RepoPilot's LangGraph nodes.""" query: str normalized_query: str @@ -54,6 +64,14 @@ class GraphRun: status: str next_node: str | None events: list[GraphEvent] + thread_id: str + + +class WorkflowState(TypedDict): + agent_state: AgentState + path: Annotated[list[str], operator.add] + events: Annotated[list[GraphEvent], operator.add] + breakpoints: list[str] Node = Callable[[AgentState], None] @@ -61,9 +79,10 @@ class GraphRun: EventSink = Callable[[GraphEvent], None] -def _constant_router(next_node: str) -> Router: - """Build a router that always returns the same target node.""" +_event_sink: ContextVar[EventSink | None] = ContextVar("repopilot_event_sink", default=None) + +def _constant_router(next_node: str) -> Router: def _route(_state: AgentState) -> str: return next_node @@ -71,30 +90,47 @@ def _route(_state: AgentState) -> str: class StateGraph: - """Small dependency-free state graph with explicit, bounded transitions. + """RepoPilot facade over the official LangGraph runtime. - A node mutates shared state. Its router chooses the next node, which makes - retries and stop conditions visible instead of hiding them in nested code. + The facade keeps RepoPilot's existing mutating business nodes small while + LangGraph owns conditional execution, recursion limits, checkpointing and + interrupt/resume behavior. A fresh state copy is written after each node so + every checkpoint represents a stable workflow boundary. """ + framework = "langgraph" + def __init__(self) -> None: self._nodes: dict[str, Node] = {} self._routers: dict[str, Router] = {} self._entrypoint = "" + self._compiled: Any = None + self._compile_lock = threading.Lock() + self._checkpointer = InMemorySaver( + serde=JsonPlusSerializer( + allowed_msgpack_modules=[ + ("app.workflow", "AgentState"), + ("app.workflow", "GraphEvent"), + ("app.agent", "ToolCall"), + ("app.rag_engine", "SearchHit"), + ("app.rag_engine", "Chunk"), + ] + ) + ) def add_node(self, name: str, node: Node) -> None: if name == END: raise ValueError(f"{END} is reserved") self._nodes[name] = node + self._compiled = None def add_edge(self, source: str, target: str | Router) -> None: - if callable(target): - self._routers[source] = target - else: - self._routers[source] = _constant_router(target) + self._routers[source] = target if callable(target) else _constant_router(target) + self._compiled = None def set_entrypoint(self, name: str) -> None: self._entrypoint = name + self._compiled = None def run( self, @@ -107,98 +143,163 @@ def run( breakpoints: set[str] | None = None, skip_breakpoint_once: str | None = None, event_sink: EventSink | None = None, + thread_id: str | None = None, ) -> GraphRun: + """Invoke or resume the compiled graph. + + The legacy cursor arguments remain for API compatibility. A resumed run + uses LangGraph's saved cursor rather than manually jumping to a node. + """ + if not self._entrypoint: raise RuntimeError("workflow entrypoint is not configured") - - current = start_at or self._entrypoint - path = list(prior_path or []) - events = list(prior_events or []) - active_breakpoints = breakpoints or set() - remaining = max_transitions - len(path) - if remaining <= 0: + completed_transitions = len(prior_path or []) if start_at is not None else 0 + remaining_transitions = max_transitions - completed_transitions + if remaining_transitions <= 0: state.stop_reason = "graph_transition_limit" raise RuntimeError(f"workflow exceeded {max_transitions} transitions") - for _ in range(remaining): - if current in active_breakpoints and current != skip_breakpoint_once: - event = GraphEvent( - sequence=len(events) + 1, - node=current, - status="paused", - latency_ms=0, - next_node=current, - detail={"reason": "breakpoint"}, - ) - events.append(event) - if event_sink: - event_sink(event) - return GraphRun( - path=path, - transitions=len(path), - status="paused", - next_node=current, - events=events, + compiled = self._compile() + resolved_thread_id = thread_id or str(uuid.uuid4()) + config = { + "configurable": {"thread_id": resolved_thread_id}, + # LangGraph counts the terminal superstep in addition to executed + # nodes. Add one so max_transitions keeps its public meaning. + "recursion_limit": remaining_transitions + 1, + } + active_breakpoints = sorted(breakpoints or set()) + if start_at is None: + payload: WorkflowState | Command = { + "agent_state": copy.deepcopy(state), + "path": list(prior_path or []), + "events": list(prior_events or []), + "breakpoints": active_breakpoints, + } + else: + if not thread_id: + raise ValueError("resuming a LangGraph run requires thread_id") + payload = Command( + resume={"action": "continue", "node": skip_breakpoint_once or start_at}, + update={"breakpoints": active_breakpoints}, + ) + + token = _event_sink.set(event_sink) + try: + for update in compiled.stream(payload, config, stream_mode="updates"): + if "__interrupt__" in update: + continue + for node_update in update.values(): + if not isinstance(node_update, dict): + continue + for event in node_update.get("events", []): + if event_sink: + event_sink(event) + except GraphRecursionError as exc: + state.stop_reason = "graph_transition_limit" + raise RuntimeError(f"workflow exceeded {max_transitions} transitions") from exc + finally: + _event_sink.reset(token) + + snapshot = compiled.get_state(config) + values = snapshot.values + current_state = values.get("agent_state", state) + self._copy_state(current_state, state) + path = list(values.get("path", [])) + events = list(values.get("events", [])) + next_node = snapshot.next[0] if snapshot.next else None + status = "paused" if next_node else "completed" + return GraphRun( + path=path, + transitions=len(path), + status=status, + next_node=next_node, + events=events, + thread_id=resolved_thread_id, + ) + + def cancel(self, thread_id: str) -> None: + """Delete a paused thread and all of its in-memory checkpoints.""" + + self._checkpointer.delete_thread(thread_id) + + def _compile(self): + if self._compiled is not None: + return self._compiled + with self._compile_lock: + if self._compiled is not None: + return self._compiled + if self._entrypoint not in self._nodes: + raise RuntimeError(f"workflow node is not configured: {self._entrypoint}") + + builder = LangGraphStateGraph(WorkflowState) + for name, node in self._nodes.items(): + builder.add_node(name, self._wrap_node(name, node)) + builder.add_edge(START, self._entrypoint) + for source, router in self._routers.items(): + builder.add_conditional_edges( + source, + lambda workflow_state, route=router: route( + workflow_state["agent_state"] + ), ) - skip_breakpoint_once = None - node = self._nodes.get(current) - if node is None: - raise RuntimeError(f"workflow node is not configured: {current}") - path.append(current) + self._compiled = builder.compile(checkpointer=self._checkpointer) + return self._compiled + + def _wrap_node(self, name: str, node: Node): + router = self._routers.get(name) + + def _wrapped(workflow_state: WorkflowState) -> dict[str, Any]: + if name in workflow_state.get("breakpoints", []): + interrupt({"reason": "breakpoint", "node": name}) + + current = copy.deepcopy(workflow_state["agent_state"]) started = time.perf_counter() try: - node(state) + node(current) except Exception as exc: event = GraphEvent( - sequence=len(events) + 1, - node=current, + sequence=len(workflow_state.get("events", [])) + 1, + node=name, status="failed", latency_ms=int((time.perf_counter() - started) * 1000), next_node=None, detail={"error": f"{type(exc).__name__}: {exc}"}, ) - events.append(event) - if event_sink: - event_sink(event) + sink = _event_sink.get() + if sink: + sink(event) raise - router = self._routers.get(current) if router is None: - raise RuntimeError(f"workflow edge is not configured: {current}") - next_node = router(state) + raise RuntimeError(f"workflow edge is not configured: {name}") + target = router(current) status = "completed" - if current == "rewrite_query": + if name == "rewrite_query": status = "retry" - elif current == "verify_evidence" and state.evidence_quality == "low": + elif name == "verify_evidence" and current.evidence_quality == "low": status = "rejected" - elif current == "reflect" and state.reflection.get("decision") == "revise": + elif name == "reflect" and current.reflection.get("decision") == "continue": status = "retry" event = GraphEvent( - sequence=len(events) + 1, - node=current, + sequence=len(workflow_state.get("events", [])) + 1, + node=name, status=status, latency_ms=int((time.perf_counter() - started) * 1000), - next_node=None if next_node == END else next_node, + next_node=None if target == END else target, detail={ - "attempt": state.retrieval_attempts, - "evidence_quality": state.evidence_quality, - "stop_reason": state.stop_reason, - "selected_skill": state.selected_skill, - "react_steps": state.react_steps, + "attempt": current.retrieval_attempts, + "evidence_quality": current.evidence_quality, + "stop_reason": current.stop_reason, + "selected_skill": current.selected_skill, + "react_steps": current.react_steps, }, ) - events.append(event) - if event_sink: - event_sink(event) - current = next_node - if current == END: - return GraphRun( - path=path, - transitions=len(path), - status="completed", - next_node=None, - events=events, - ) + return {"agent_state": current, "path": [name], "events": [event]} + + _wrapped.__name__ = f"repopilot_{name}" + return _wrapped - state.stop_reason = "graph_transition_limit" - raise RuntimeError(f"workflow exceeded {max_transitions} transitions") + @staticmethod + def _copy_state(source: AgentState, target: AgentState) -> None: + for item in fields(AgentState): + setattr(target, item.name, copy.deepcopy(getattr(source, item.name))) diff --git a/docs/architecture.md b/docs/architecture.md index 7983dad..c9f4157 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -23,7 +23,7 @@ flowchart TD K --> L["Answer + Citations + Episodic Memory"] ``` -`AgentState` 在节点之间传递原始问题、当前检索查询、意图、Skill、Plan、Memory hits、packed context、Tool Observation、Reflection、尝试次数、证据质量、答案和停止原因。`StateGraph` 同时限制 retrieval、ReAct step 和全局 transition，防止错误路由造成无界循环。 +`AgentState` 在节点之间传递原始问题、当前检索查询、意图、Skill、Plan、Memory hits、packed context、Tool Observation、Reflection、尝试次数、证据质量、答案和停止原因。业务节点直接注册到 LangGraph `StateGraph`；条件边表达 evidence retry 与 ReAct loop，LangGraph recursion limit 再提供全局兜底，防止错误路由造成无界循环。 ## Context Engine @@ -44,7 +44,7 @@ Tool Registry 对 Native 与 MCP 使用同一执行协议。MCP 客户端执行 ## Breakpoint and Checkpoint -调试运行可以在任意节点执行前设置断点。`StateGraph.run` 返回 `paused`、待执行节点、历史路径和节点事件；服务端用内存 Checkpoint 保存 `AgentState`，恢复时跳过当前断点一次并继续执行。暂停期间有三种操作： +调试运行可以在任意节点执行前设置断点。每个节点入口根据本次调试配置调用 LangGraph `interrupt()`；`InMemorySaver` 按 thread ID 保存状态、路径和节点事件，恢复时使用 `Command(resume=...)` 从同一 checkpoint 继续。暂停期间有三种操作： - Resume：保留当前状态继续，循环再次经过同一节点时仍会命中断点。 - Edit and Restart：替换问题并从 Route 重新计算，避免把旧意图或旧证据带入新问题。 @@ -52,7 +52,7 @@ Tool Registry 对 Native 与 MCP 使用同一执行协议。MCP 客户端执行 前端 SVG 执行图根据 `graph_events` 区分 completed、warning、retry、failed 和 paused，并将工具事件按实际顺序展开。 -普通 `/api/ask` 返回完整结果；`/api/ask/stream` 在后台线程运行同一 State Graph，并通过 SSE event sink 在每个节点完成、失败或暂停时立即推送 `graph_event`，最后推送 `final`。这里选择 SSE 是因为运行状态是服务端到客户端的单向事件流；不会为了复用 WebSocket 关键词引入不必要的双向协议。 +普通 `/api/ask` 返回完整结果；`/api/ask/stream` 在后台线程消费同一 LangGraph 的 `updates` stream，并通过 SSE 在每个节点完成、失败或暂停时立即推送 `graph_event`，最后推送 `final`。这里选择 SSE 是因为运行状态是服务端到客户端的单向事件流；不会为了复用 WebSocket 关键词引入不必要的双向协议。 ## Verification Loop diff --git a/docs/langgraph-integration.md b/docs/langgraph-integration.md index 89ba2d0..b858208 100644 --- a/docs/langgraph-integration.md +++ b/docs/langgraph-integration.md @@ -1,39 +1,28 @@ # LangGraph Integration -RepoPilot now provides a LangGraph runtime adapter on top of the existing -engineering diagnosis workflow. - -## Design - -The workflow is represented as an explicit state graph: - -``` -Planner - -> Retrieve - -> Verify Evidence - | sufficient - v - Tool Executor - -> Reflection - | complete - v - Synthesize -``` - -## Why an adapter instead of a rewrite - -The original workflow already contains bounded transitions, checkpoints, -verification and tool safety rules. LangGraph is introduced as the orchestration -layer while preserving the existing engineering controls. - -## Components - -- Planner: generates diagnosis steps. -- Retrieve: gathers repository evidence. -- Verify: checks evidence quality before actions. -- Tool Executor: performs repository analysis tools. -- Reflection: decides whether more investigation is required. -- Synthesize: produces the final engineering conclusion. - -This exposes RepoPilot as a standard LangGraph Agent workflow while keeping its -existing retrieval, memory and safety modules reusable. +RepoPilot 的主执行链已经由 LangGraph 驱动。`KnowledgeAgent.ask()`、SSE 流式问答和调试接口都进入同一个 compiled graph，不存在旁路 demo 或只供导入的 adapter。 + +## Runtime boundary + +`app/workflow.py` 是唯一编排入口： + +- LangGraph `StateGraph` 执行 Route、Memory、Plan、Retrieve、Verify、Rewrite、ReAct、Reflect 和 Synthesize。 +- 条件边根据 `AgentState` 选择证据重试、下一项工具或结束。 +- `recursion_limit` 是 retrieval budget 和 ReAct budget 之外的全局循环保护。 +- 每个业务节点写回独立 state copy，因此 checkpoint 不会被后续原位修改污染。 +- 图的 `updates` stream 被转换为已有 `GraphEvent`，继续服务 SSE 和运行指标。 + +## Checkpoint and interrupt + +每个运行分配独立 LangGraph thread ID，并使用 `InMemorySaver` 保存节点边界状态。调试断点在节点入口调用 `interrupt()`： + +1. `/api/debug/run` 执行到断点后返回 paused 和 next_node。 +2. `/api/debug/resume` 用同一 thread ID 与 `Command(resume=...)` 继续。 +3. 编辑问题并 restart 时删除旧 checkpoint，从 Route 创建全新 thread。 +4. 完成或 cancel 时删除 checkpoint，避免服务进程持续积累状态。 + +当前 checkpointer 是单进程内存实现，适合本地演示和测试。生产部署可将同一编排层替换为数据库 checkpointer，从而支持进程重启后的恢复；业务节点和 API 协议无需重写。 + +## Evidence and tool safety + +LangGraph 只负责运行时，不绕过 RepoPilot 的工程约束。Evidence Gate 仍会阻止低证据请求进入外部 LLM；Tool Registry 仍负责 Skill 白名单、MCP/Native 统一执行、human approval、idempotency 和 rollback。这样既获得标准持久化与恢复语义，也保留现有可验证的工程行为。 diff --git a/requirements.txt b/requirements.txt index 6c6df1f..317fc37 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,4 +7,4 @@ httpx[socks]==0.28.1 sentence-transformers==5.7.0 pypdf==6.16.1 python-docx==1.2.0 -langgraph==0.6.8 +langgraph==1.2.11 diff --git a/tests/test_api.py b/tests/test_api.py index 1646fb4..668726a 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -61,6 +61,8 @@ def test_kb_stats_documents_workflow(api: TestClient) -> None: docs = api.get("/api/kb/documents").json() assert docs and docs[0]["title"] == "检索服务架构" workflow = api.get("/api/workflow").json() + assert workflow["framework"] == "langgraph" + assert workflow["checkpointing"] == "langgraph_in_memory" assert workflow["entrypoint"] == "route" assert "verify_evidence" in workflow["nodes"] assert "recall_memory" in workflow["nodes"] diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 7ad0ef8..688da04 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -51,6 +51,7 @@ def test_state_graph_pauses_and_resumes_at_breakpoint() -> None: prior_events=paused.events, breakpoints={"second"}, skip_breakpoint_once=paused.next_node, + thread_id=paused.thread_id, ) assert paused.status == "paused" @@ -58,3 +59,9 @@ def test_state_graph_pauses_and_resumes_at_breakpoint() -> None: assert resumed.status == "completed" assert resumed.path == ["first", "second"] assert state.answer == "done" + + +def test_state_graph_is_backed_by_langgraph() -> None: + graph = StateGraph() + + assert graph.framework == "langgraph"