今天做了什么

系统分层

今天的工作是给系统分层，以前我在写系统的时候更多的关注于系统功能，但是这样在工程项目里面是不对的，一个系统不光要有功能，还要有系统层次，让系统好维护。

所以今天的主要工作是给咱们的系统做个升级，增加一个编排层。讲到这里我需要讲讲什么是编排层，编排层是一个系统将自身复杂的内部操作封装，对外暴露出仅一个接口的这样一个模块。

这样做最直接的好处是功能内部逻辑和上层接口解开，实现系统的模块化，这样系统功能容易插拔比较日后的换功能和系统维护升级。

我在网上找到了一个关于编排层的解释案例，觉得能很好的帮助理解编排层的作用，从下面的案例可以看出来，让顾客去操作底层业务逻辑，比如指挥厨师1或厨师2完成某项工作是很不合适的，所以比较合理的方式是将上层请求与底层业务逻辑分开，加上编排层，那么系统分层后，日后维护（例如换厨师），那么就不需要修改大量的调用代码了。

项目像一家餐厅：
厨房 1：RAG 检索
厨房 2：time tool
厨房 3：calculator tool
调度流程：LangGraph
顾客：API / 前端 / Django
那“编排层入口”就像前台接单员。

前面咱们在main.py里面写build_agent_graph(...)然后再graph.invoke(...)，这样系统能跑是没错，但是等于是FastAPI的上层请求，直接调用底层逻辑，系统没有分层的，分层后的系统对于上传调用只留有一个接口，最佳状态是用户传入一个问题和对话ID后系统会编排层会直接给出一个最终结果。下面就来创建编排层首先，需要在app/graph/目录下创建一个workflow.py文件用写我们的编排层逻辑：

from app.graph.builder import build_agent_graph


class AgentWorkflow:
    def __init__(self, tools, rag=None):
        self.tools = tools
        self.rag = rag
        self.graph = build_agent_graph(tools, rag=rag)

    def invoke(self, session_id: str, query: str, chat_history=None):
        if chat_history is None:
            chat_history = []

        state = {
            "session_id": session_id,
            "query": query,
            "chat_history": chat_history,
        }

        result = self.graph.invoke(state)
        return result["final_answer"]

最后需要改main.py文件：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from app.config import DATA_DIR
from app.data_loader import load_pdfs, process_documents
from app.rag_system import RAGSystem
from app.tools import TOOLS
from app.session_manager import SessionManager
from app.logger_config import setup_logger

from app.graph.workflow import AgentWorkflow

logger = setup_logger()

app = FastAPI()

session_manager = SessionManager(max_turns=3)

rag = None
workflow = None


class QueryRequest(BaseModel):
    session_id: str
    question: str


@app.on_event("startup")
def startup_event():
    global rag, workflow

    logger.info("Loading RAG system...")

    docs = load_pdfs(DATA_DIR)
    logger.info(f"docs数量: {len(docs)}")

    chunks = process_documents(docs)
    logger.info(f"chunks数量: {len(chunks)}")

    rag = RAGSystem(chunks)
    rag.build_index()

    workflow = AgentWorkflow(TOOLS, rag=rag)

    logger.info("RAG + LangGraph ready!")


@app.post("/ask")
def ask_question(req: QueryRequest):
    try:
        history = session_manager.get_history(req.session_id)

        answer = workflow.invoke(
            session_id=req.session_id,
            query=req.question,
            chat_history=history
        )

        session_manager.append_turn(
            req.session_id,
            req.question,
            answer
        )

        return {
            "session_id": req.session_id,
            "question": req.question,
            "answer": answer
        }

    except Exception as e:
        logger.exception("Error occurred in /ask")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/clear/{session_id}")
def clear_session(session_id: str):
    session_manager.clear_session(session_id)
    return {
        "session_id": session_id,
        "message": "session cleared"
    }

补异常处理和日志

这里没有什么多说的，咱们需要给现有的node.py补充异常处理，也就是加try/except和日志：

import json
from typing import Any

from app.graph.state import AgentState
from app.llm_utils import client
from app.config import CHAT_MODEL
from app.logger_config import setup_logger

logger = setup_logger()


def build_choose_tool_node(tools: list[dict[str, Any]]):
    def choose_tool_node(state: AgentState) -> AgentState:
        query = state["query"]
        logger.info(f"[choose_tool_node] query: {query}")

        tool_desc = "\n".join([
            f"{t['name']}: {t['description']}" for t in tools
        ])

        prompt = f"""
                    You are an AI agent.
                    
                    Available tools:
                    {tool_desc}
                    
                    User question:
                    {query}
                    
                    Return JSON:
                    {{"tool": "...", "input": "..."}}
                  """

        content = ""
        try:
            response = client.chat.completions.create(
                model=CHAT_MODEL,
                messages=[{"role": "user", "content": prompt}]
            )
            content = response.choices[0].message.content
            decision = json.loads(content)

            logger.info(f"[choose_tool_node] decision: {decision}")

            return {
                "decision": decision
            }

        except Exception as e:
            logger.exception("choose_tool_node failed")
            return {
                "decision": {"tool": "llm", "input": query},
                "error": f"choose_tool_node failed: {str(e)}"
            }

    return choose_tool_node


def build_execute_tool_node(tools: list[dict[str, Any]], rag=None):
    def execute_tool_node(state: AgentState) -> AgentState:
        try:
            decision = state["decision"]
            chat_history = state.get("chat_history", [])

            tool_name = decision["tool"]
            tool_input = decision["input"]

            logger.info(f"[execute_tool_node] tool_name: {tool_name}, tool_input: {tool_input}")

            for t in tools:
                if t["name"] == tool_name:
                    if tool_name == "rag":
                        result = t["func"](tool_input, rag, chat_history=chat_history)
                    elif tool_name == "llm":
                        result = t["func"](tool_input, chat_history=chat_history)
                    else:
                        result = t["func"](tool_input)

                    logger.info(f"[execute_tool_node] tool_output: {result}")

                    return {
                        "tool_result": {
                            "tool_name": tool_name,
                            "tool_input": tool_input,
                            "tool_output": result
                        }
                    }

            logger.warning(f"[execute_tool_node] tool not found: {tool_name}")

            return {
                "tool_result": {
                    "tool_name": "none",
                    "tool_input": tool_input,
                    "tool_output": "No valid tool found."
                },
                "error": f"Tool not found: {tool_name}"
            }

        except Exception as e:
            logger.exception("execute_tool_node failed")
            return {
                "tool_result": {
                    "tool_name": "error",
                    "tool_input": "",
                    "tool_output": "Tool execution failed."
                },
                "error": f"execute_tool_node failed: {str(e)}"
            }

    return execute_tool_node


def generate_answer_node(state: AgentState) -> AgentState:
    if state.get("error"):
        logger.warning(f"[generate_answer_node] error found in state: {state['error']}")
        return {
            "final_answer": f"系统执行过程中出现问题：{state['error']}"
        }

    tool_result = state["tool_result"]
    logger.info(f"[generate_answer_node] final_answer: {tool_result['tool_output']}")

    return {
        "final_answer": str(tool_result["tool_output"])
    }

遇到的问题

在做编排层的时候其实我有三个点是有疑惑的：
1、首先是为什么需要LangGraph？

2、定义的Graph是如何接State的？

3、Node的返回结果总是给人一种割裂的感觉，但是又说不清是什么？

当前理解

1、为什么用LangGraph，其实现在的理解是为了扩展方便，以前做手写Agent的时候，选工具、执行再到给返回结果这样的工作流其实是写死的，但是用了LangGraph后，这样工作流变了，变得更灵活了，每一个对数据的操作都是一个节点，那么我们可以创建很多的节点，然后补断增加边来建立新的工作流，其实现在想LangGraph的意义在于，可以从工作流角度设计Agent，相当于站在一个高处设计业务，手写Agent系统，每次都盯着具体的函数和变量怎么传递，其实业务少还好，如果业务多了，在一堆函数之间来回调用，其实设计会非常繁琐和烧脑，因为你需要想各种变量怎么传，但是有了LangGraph你只需要又一个状态图State，然后补断设计业务节点怎么串起来就好。

2、后来查了一些资料，似乎我们在定义图的时候，这就相当于建立了一个工作流水线，虽然创建流水线的代码没有明确定义传入state参数：

def build_agent_graph(tools, rag=None):
    graph_builder = StateGraph(AgentState)
    ...

但是graph_builder = StateGraph(AgentState)这句话描述了传入图数据的样子，当使用invoke()函数时，这个函数做了封装，会自动将state作为参数传给node函数。

3、关于Node的返回值，后来看node.py我有注意到，node的返回值不是某个参数，node的返回值在LangGraph这个框架下是对State的修改操作，这里有个坑，他跟一般理解上的函数返回一个值不一样，他是返回一个修改操作。

如果这篇文章对你有帮助，可以点个赞～
完整代码地址：https://github.com/1186141415/Paper-RAG-Agent-with-LangGraph