当前位置：首页 > news >正文

法律 AI Agent：从架构到案例匹配的技术方案与工程实践

news 2026/7/1 4:23:57

普通人遇到法律问题，第一反应往往不是"找律师"，而是——“我这事算什么性质？该怎么办？”

劳动争议、民间借贷、婚姻纠纷、交通事故……这些高频法律场景，真正走到诉讼阶段的只是少数。大多数人卡在第一步：不知道自己的情况对应什么法律问题，也不知道该从哪开始。

这篇文章从整体架构讲到案例匹配模块的技术实现——LangGraph 工作流编排、ES + Qdrant 多路召回、LLM 精排、Guardrail 三层校验链。不做概念科普，只讲设计和代码。

一、整体定位：四级漏斗

用户描述问题 ↓ 第一层：案情导诊 判断属于什么法律问题，评估严重程度，给出行动方向 ↓ 第二层：案例匹配（核心能力） 检索相似案例，分析判决结果，提供辩护/应诉思路参考 ↓ 第三层：自助工具 生成法律文书，提供流程指引和证据清单 ↓ 第四层：律师匹配 按领域/地区推荐律师

用户从导诊进入，根据问题的复杂程度停在不同的层级。简单问题导诊 + 自助工具就够了，复杂问题走到底层找律师。

二、整体架构

用户界面（小程序 / Web / 公众号） │ ┌─────▼────────────────────────────────────┐ │ Agent 核心层 │ │ │ │ 案情理解 → 导诊分类 → 行动建议 │ │ 事实提取 风险评估 分步指南 │ │ 要素识别 紧急度 材料清单 │ │ │ │ 案例匹配 → 辩护思路 │ │ 语义检索 策略推荐 │ │ 判决分析 法条引用 │ │ │ │ 文书生成 → 律师匹配 → 流程跟踪 │ └────┬───────────┬───────────┬─────────────┘ │ │ │ ┌────▼───┐ ┌────▼────┐ ┌───▼───────────┐ │ 知识库 │ │ 案例库 │ │ 外部服务 │ │ │ │ │ │ │ │ 法律条文 │ │ 裁判文书 │ │ 律师信息 │ │ 司法解释 │ │ 按案由 │ │ 法院信息 │ │ 流程知识 │ │ 按语义 │ │ 工商信息 │ │ 文书模板 │ │ 判决结果 │ │ │ └─────────┘ └─────────┘ └───────────────┘

三、案例匹配：完整技术实现

案例匹配是 Agent 最有价值也最难的模块。用户说"老板不给工资还把我辞了"，Agent 能从裁判文书库中找到事实最接近的案例，告诉用户类似情况法院怎么判的、赔偿范围多少、怎么主张权利。

3.1 LangGraph 工作流编排

整个流程有 4 个 Stage，但不是简单的线性流水线。Stage 1 发现关键信息缺失要回退问用户，Stage 3 精排结果太差要放宽召回条件重试。用 LangGraph 的状态图来编排是最自然的选择。

from langgraph.graph import StateGraph, END from typing import TypedDict, Optional, List # === 全局状态定义 === class LegalCaseState(TypedDict): user_input: str structured: Optional[dict] # Stage 1 结构化结果 missing_info: List[str] # 缺失的关键信息 tag_filtered_ids: List[str] # 标签过滤后的候选 ID recall_candidates: List[dict] # 向量召回结果 reranked_top5: List[dict] # 精排结果 output: Optional[dict] # 最终产出 retry_count: int # 重试次数 max_retries: int # 最大重试 # === 构建状态图 === workflow = StateGraph(LegalCaseState) workflow.add_node("extract_legal_elements", extract_legal_elements) workflow.add_node("ask_missing_info", ask_missing_info) workflow.add_node("tag_filter", tag_filter) workflow.add_node("vector_recall", vector_recall) workflow.add_node("llm_rerank", llm_rerank) workflow.add_node("relax_recall", relax_recall) workflow.add_node("generate_output", generate_output) workflow.set_entry_point("extract_legal_elements") # 有条件边：信息缺失 → 反问用户 workflow.add_conditional_edges( "extract_legal_elements", decide_missing_info, {"ask_user": "ask_missing_info", "proceed": "tag_filter"} ) workflow.add_edge("ask_missing_info", "extract_legal_elements") # 召回链路 workflow.add_edge("tag_filter", "vector_recall") workflow.add_edge("vector_recall", "llm_rerank") # 精排不达标 → 放宽条件重试 workflow.add_conditional_edges( "llm_rerank", decide_retry, {"acceptable": "generate_output", "retry": "relax_recall", "failed": END} ) workflow.add_edge("relax_recall", "vector_recall") app = workflow.compile()

为什么用 LangGraph 而不是 LangChain Chain？因为流程中有多个条件分支和回退环路——信息缺失要回退到反问、精排不达标要放宽条件重试。Chain 是 DAG 结构，表达不了带环的流程；StateGraph 的循环边天然适合这种场景。

路由函数实现：

def decide_missing_info(state: LegalCaseState) -> str: """判断是否需要反问用户补充信息""" if state["missing_info"] and state["retry_count"] == 0: critical = ["amount_involved", "has_contract"] for info in state["missing_info"]: if info in critical: return "ask_user" return "proceed" def decide_retry(state: LegalCaseState) -> str: """判断精排结果是否可接受""" if not state["reranked_top5"]: return "retry" if state["retry_count"] < state["max_retries"] else "failed" top_score = state["reranked_top5"][0].get("overall_score", 0) if top_score < 6.0 and state["retry_count"] < state["max_retries"]: state["retry_count"] += 1 return "retry" return "acceptable"

3.2 Stage 1：案情结构化

用户的口语化描述先转化为结构化法律要素。这一步做不好，后面全是错的。

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field class LegalElements(BaseModel): case_type: str = Field(description="案件类型：民事/刑事/行政") case_category: str = Field(description="案由大类，如劳动争议") sub_category: str = Field(description="案由小类") employment_duration: Optional[str] = None has_contract: Optional[bool] = None unpaid_wages: Optional[dict] = None termination_type: Optional[str] = None amount_involved: Optional[str] = None keywords: List[str] = Field(default_factory=list) applicable_laws: List[str] = Field(default_factory=list) missing_info: List[str] = Field(default_factory=list) parser = PydanticOutputParser(pydantic_object=LegalElements) prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个法律案情分析专家。将用户的自然语言描述转化为结构化法律要素。 注意： 1. **只抽取明确提到的信息，不要猜测** 2. **如果关键信息缺失，在 missing_info 中标注** 3. **案由分类参考《民事案件案由规定》** {format_instructions}"""), ("human", "{user_input}") ]) chain = prompt | llm | parser

输出结构化结果的同时，missing_info字段驱动 LangGraph 的ask_missing_info节点反问用户补全信息：

def ask_missing_info(state: LegalCaseState): questions = [] for k in state["missing_info"][:3]: if k == "amount_involved": questions.append("您的月工资是多少？这关系到赔偿金额的计算。") elif k == "has_social_security": questions.append("公司有没有给您缴纳社保？") elif k == "has_contract": questions.append("您和公司签过劳动合同吗？") return {"messages": [AIMessage(content="/n".join(questions))]}

3.3 Stage 2：多路召回

裁判文书网有上亿份判决书，不能全量做向量检索。分两级：ES 标签过滤 + Qdrant 向量语义检索。

第一级：ES 标签精确过滤

用 Stage 1 的结构化标签做过滤，快速缩小候选集。选用 ES 而不是关系数据库，因为裁判文书除了结构化字段还有全文内容，ES 的倒排索引可以同时在结构化字段和文本字段上做混合过滤。

from elasticsearch import Elasticsearch es = Elasticsearch("http://localhost:9200") def tag_filter(state: LegalCaseState) -> LegalCaseState: elements = state["structured"] must_clauses = [ {"term": {"case_category": elements["case_category"]}}, {"terms": {"sub_category": [elements["sub_category"]]}}, {"range": {"year": {"gte": 2022, "lte": 2026}}}, ] if elements.get("has_contract") is False: must_clauses.append({"term": {"has_contract": False}}) if elements.get("amount_involved") and elements["amount_involved"] != "未知": amount = extract_number(elements["amount_involved"]) must_clauses.append({"range": {"amount_involved": {"lte": amount * 2}}}) body = { "query": {"bool": {"must": must_clauses}}, "size": 5000, "_source": ["case_id", "case_category", "fact_summary", "judgment"] } resp = es.search(index="judgments", body=body) case_ids = [hit["_source"]["case_id"] for hit in resp["hits"]["hits"]] return {"tag_filtered_ids": case_ids}

第二级：Qdrant 向量语义检索

在标签过滤后的候选集上，用事实描述做语义检索。选 Qdrant 而不是 Milvus：这个场景只需要单机部署、百亿级以内的向量量，Qdrant 一个 Docker 容器搞定，部署成本比 Milvus 低得多。

Embedding 模型选 bge-m3（1024 维）。法律文书中法条编号（“第82条”）、案例编号（“(2023)京01民终1234号”）中英文混写，bge-m3 的多语言能力在 1024 维下检索精度比 text-embedding-3-small 高 3-5 个百分点（在我们的 2000 条标注测试集上）。

from qdrant_client import QdrantClient from sentence_transformers import SentenceTransformer model = SentenceTransformer("BAAI/bge-m3") client = QdrantClient(host="localhost", port=6333) def vector_recall(state: LegalCaseState) -> LegalCaseState: case_ids = state["tag_filtered_ids"] query_text = state["structured"].get("fact_summary", state["user_input"]) query_vector = model.encode(query_text).tolist() # Qdrant 支持用 payload filter 缩小搜索范围 hits = client.search( collection_name="judgments_by_category", query_vector=query_vector, query_filter={"must": [{"has_id": case_ids}]}, limit=50, ) # 混合权重：0.6 × 向量相似度 + 0.4 × 标签匹配度 candidates = [] for hit in hits: vector_score = hit.score tag_score = compute_tag_match(state["structured"], hit.payload) combined = 0.6 * vector_score + 0.4 * tag_score candidates.append({ "case_id": hit.payload["case_id"], "fact_summary": hit.payload["fact_summary"], "judgment": hit.payload["judgment"], "combined_score": combined, }) candidates.sort(key=lambda x: x["combined_score"], reverse=True) return {"recall_candidates": candidates[:50]}

为什么先做标签过滤再做向量检索？因为"法律关系相似"和"事实描述相似"是两回事。两个案例都提到"欠薪"，但一个是劳动合同纠纷（适用劳动法），一个是劳务合同纠纷（适用民法典），法律关系不同，判决依据完全不同。标签过滤确保候选集在法律关系上是同类的。

混合权重

def compute_tag_match(structured: dict, payload: dict) -> float: score = 0.0 # 子案由一致给 0.5 if structured.get("sub_category") == payload.get("sub_category"): score += 0.5 # 要素重叠比例 elements = structured.get("legal_elements", {}) payload_elems = payload.get("legal_elements", {}) overlap = sum(1 for k, v in elements.items() if v and v != "未知" and payload_elems.get(k) == v) total = sum(1 for v in elements.values() if v and v != "未知") if total > 0: score += 0.5 * (overlap / total) return score

重试时的放宽策略

def relax_recall(state: LegalCaseState) -> LegalCaseState: retry = state["retry_count"] if retry == 1: state["tag_filtered_ids"] = expand_category(state) # 案由扩大到父类 elif retry == 2: state["tag_filtered_ids"] = remove_year_filter(state) # 去掉年份限制 elif retry >= 3: state["tag_filtered_ids"] = None # 全库向量检索 return state

3.4 Stage 3：LLM 精排

向量召回回来的 TOP 50，还需要做一次精细比对。向量相似度衡量的是"措辞相似"，不是"法律关系相似"。

用 LangChain 的with_structured_output实现多维评分：

from langchain_core.pydantic_v1 import BaseModel, Field class CaseScore(BaseModel): cause_match: int = Field(ge=1, le=10, description="案由匹配度") fact_similarity: int = Field(ge=1, le=10, description="事实相似度") dispute_focus: int = Field(ge=1, le=10, description="争议焦点") reference_value: int = Field(ge=1, le=10, description="参考价值") class RerankResult(BaseModel): scores: List[CaseScore] rerank_prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个法律案例比对专家。请从以下维度评分（1-10）： A. 案由匹配度 B. 事实相似度 C. 争议焦点 D. 参考价值 要严格评分，事实不相似就给低分。"""), ("human", "用户案情：{user_case}/n候选案例：{case_info}") ]) structured_llm = llm.with_structured_output(RerankResult) rerank_chain = rerank_prompt | structured_llm

批量评分并加权聚合：

def llm_rerank(state: LegalCaseState) -> LegalCaseState: candidates = state["recall_candidates"][:50] scored = [] for i in range(0, len(candidates), 5): batch = candidates[i:i+5] result = rerank_chain.invoke({ "user_case": state["user_input"], "case_info": [ {"case_fact": c["fact_summary"][:500], "case_judgment": c["judgment"][:300]} for c in batch ] }) for idx, score in enumerate(result.scores): scored.append({ batch[idx], "overall_score": ( score.cause_match * 0.3 + score.fact_similarity * 0.3 + score.dispute_focus * 0.2 + score.reference_value * 0.2 ) }) scored.sort(key=lambda x: x["overall_score"], reverse=True) return {"reranked_top5": scored[:5]}

这份精排结果同时输入到判决统计（用 TOP 50 的判决结果做统计分析）和策略建议（用 TOP 5 的法院观点生成）。

3.5 Guardrail 实现

法律 Agent 的 Guardrail 和其他场景不同——它防护的不是"执行了危险操作"，而是输出了误导性法律内容。

用 LangGraph 的 Condition 做第一道闸，自定义规则引擎做第二道，知识库检索做第三道：

def guardrail_check(state: LegalCaseState) -> str: output = state["output"] # 第一道：Agent 自检 — 绝对化表述拦截 if any(p in str(output) for p in ["保证", "一定", "100%", "肯定赢"]): return "rewrite" # 第二道：规则拦截 if state["structured"]["case_type"] in ("刑事", "行政"): if "律师" not in str(output): return "block" if "不构成法律意见" not in str(output): return "rewrite" # 第三道：法条引用校验 cited_laws = extract_law_citations(str(output)) for law in cited_laws: if not verify_citation(query_law_knowledge_base(law), str(output)): return "rewrite" return "pass" # 注册为 LangGraph 条件边 workflow.add_conditional_edges( "generate_output", guardrail_check, {"pass": END, "rewrite": "generate_output", "block": "blocked_output"} )

法条引用校验的实现——用精确查询，不用向量检索。法条原文一字不能差，向量检索的召回率做不到 100%：

def extract_law_citations(text: str) -> List[str]: import re pattern = r"[《]?([^》]+?法)[》]?第(/d+)条" matches = re.findall(pattern, text) return [f"{m[0]}第{m[1]}条" for m in matches] def query_law_knowledge_base(law_citation: str) -> str: result = es.get(index="law_knowledge_base", id=law_citation) return result["_source"]["original_text"]

四、安全性设计

法律 Agent 和其他场景最大的不同是：给错了建议，用户可能真的会输掉官司。

三条红线

不能给确定性结论。不说"你一定能赢"，说"类似案例胜诉率约 80%"
不能替代律师。刑事案件、重大商事纠纷必须建议咨询律师
不能建议违法操作。不指导毁灭证据、虚假陈述等

免责声明的分级

DISCLAIMERS = { "civil": ( "⚠️ 以上分析基于公开数据和通用法律知识，不构成正式法律意见。" "每个案件情况不同，建议咨询执业律师。" ), "criminal": ( "🚨 您描述的情况可能涉及刑事/重大民事纠纷。" "建议立即委托执业律师处理。本工具仅提供参考信息，" "不能替代专业法律服务。" ), }

这些是 Guardrail 代码强制追加的，不是 LLM 自觉加上的。如果 Guardrail 检测到输出中缺少免责声明，会触发rewrite强制补上。

五、关键的工程决策

案由分库 vs 全库统一检索

每个高频案由（劳动争议、民间借贷、婚姻家事、交通事故、买卖合同纠纷）在 Qdrant 里各一个 collection。这五类占了民事诉讼的 70% 以上。

COLLECTION_MAP = { "劳动争议": "judgments_labor", "民间借贷": "judgments_loan", "婚姻家事": "judgments_family", "交通事故": "judgments_traffic", "买卖合同纠纷": "judgments_contract", }

好处：检索范围小速度快、无跨案由语义噪音、不同案由可独立调优。代价：案由识别必须准确，新增案由需要建新库。

裁判文书预处理

原始裁判文书有几大问题：长（少则几千字多则几万字）、模板化（大量格式文本）、噪声多（OCR 错误）。预处理必须做 LLM 提取，只把fact_summary（300 字以内）拿去 Embedding。

超过 512 token 长度的文本，bge-m3 的向量表示质量会明显下降。所以提取内容控制在 500 字以内做 Embedding。

冷启动策略

从最高人民法院发布的典型案例和指导案例开始——这些案例质量高、权威性强，数量不多（几千份）但覆盖主要案由。然后扩展到各省高院的裁判文书，最后再扩展到基层法院。

增量更新：每周跑一次增量 pipeline，新文书经过预处理 → Embedding → Qdrant upsert，不需要全量重建。

时效性

2020 年的案例和 2025 年的案例，法律可能有修改（如 2024 年民法典合同编司法解释）。默认只检索近 3 年，案例太少才放宽。

六、V2.0 升级：五个关键优化

上面这套方案上线跑通（以下简称 V1.0）之后，有几个问题会暴露出来。我们在迭代过程中识别了五个核心短板，对应做了 V2.0 的架构升级。

6.1 CrossEncoder 初排：解决 LLM 精排成本

V1.0 的问题： TOP 50 直接送 LLM 打分。如果用的 GPT-4.1 或 Claude，50 个 case 逐个看，单次查询仅精排就要几万 token，成本和延迟都扛不住。月活一万用户，光精排每个月烧掉几千美元。

V2.0 的改进：在向量检索和 LLM 精排之间加一层 CrossEncoder 初排。

V1.0: ES → Qdrant (50) → LLM 精排 → TOP 5 V2.0: ES → Qdrant (50) → CrossEncoder (50→20) → LLM 精排 → TOP 5

CrossEncoder 比向量检索（Bi-Encoder）精度高，比 LLM 成本低一个数量级。

from sentence_transformers import CrossEncoder # bge-reranker-v2 或 Qwen3-Reranker 均可 reranker = CrossEncoder("BAAI/bge-reranker-v2-m3") def cross_encoder_rerank(state: LegalCaseState) -> LegalCaseState: candidates = state["recall_candidates"][:50] query = state["structured"].get("fact_summary", state["user_input"]) pairs = [(query, c["fact_summary"][:512]) for c in candidates] scores = reranker.predict(pairs) for idx, score in enumerate(scores): candidates[idx]["crossencoder_score"] = float(score) # 融合向量得分和 CrossEncoder 得分 candidates[idx]["combined_score"] = ( 0.3 * candidates[idx]["combined_score"] + 0.7 * float(score) ) candidates.sort(key=lambda x: x["combined_score"], reverse=True) return {"recall_candidates": candidates[:20]} # 只保留 TOP 20 给 LLM

方案	单次成本	延迟	精度
LLM 直接看 50 个	$$$	10-20s	最高
CrossEncoder 50→20 + LLM 看 20 个	$	2-3s	≈ 持平
纯 CrossEncoder	$	~1s	略低但有瓶颈

CrossEncoder 的得分（01）和向量余弦相似度（01）做加权融合。线上实验下来，0.7 × CrossEncoder + 0.3 × 向量相似度效果最好，既保留了语义信息，又大幅压低了 LLM 调用量。

选型上 bge-reranker-v2-m3 和 Qwen3-Reranker 都可用。前者对中英文混写文本（法律文书场景）表现更好，后者在纯中文场景略优。

6.2 法条优先 + 联合推理：纠正成文法系逻辑

V1.0 的问题：案例匹配→策略推荐，这套逻辑是美国判例法思维。中国是成文法系，裁判逻辑是法条 → 司法解释 → 指导案例，案例只是参考，不能作为判决依据。

只给用户看"类似案例怎么判"，不给"法律依据是什么"，在中国法律场景下是半成品。

V2.0 的改进：召回链路改成法条召回 → 案例召回 → 联合推理。

V1.0: 案情结构化 → 案例检索 → 策略推荐 V2.0: 案情结构化 → ┌── 法条检索 (Law RAG) ├── 案例检索 (Case RAG) └── 联合推理 (Legal Reasoner)

法条检索独立成一条召回链路：

def law_recall(state: LegalCaseState) -> LegalCaseState: """从法律知识库中检索相关法条""" elements = state["structured"] # 方法 1：基于案由的精确匹配 law_ids = es.search(index="law_knowledge_base", body={ "query": {"bool": {"must": [ {"term": {"applicable_categories": elements["case_category"]}}, ]}}, "size": 10, }) # 方法 2：基于关键要素的语义检索 query_text = " ".join(elements.get("keywords", [])) law_vectors = law_model.encode(query_text).tolist() law_hits = law_client.search( collection_name="laws", query_vector=law_vectors, limit=5, ) return {"law_candidates": combine_law_results(law_ids, law_hits)}

联合推理阶段，LLM 同时看到法条原文和相似案例，生成的建议结构变为：

📋 分析结果 【法律依据】 劳动合同法第10条：建立劳动关系应当订立书面劳动合同 劳动合同法第82条：未签合同超过一个月，应支付双倍工资 劳动合同法第87条：违法解除劳动合同，应支付二倍经济补偿金 【相似案例参考】 案例1：张三诉XX公司……（以下为参考，不作为判决依据） 【策略建议】 基于上述法律依据和参考案例……

法条是依据，案例是参考。用户先看到法律依据，再看案例，逻辑上更符合中国法律体系。

6.3 法律知识图谱：连接法条、概念与案例

V1.0 的问题：法条库和案例库各自独立。当法条更新时（如 2024 年民法典合同编司法解释发布），旧案例引用的法条版本可能已经不适用。而且法条和场景之间的关联是隐式的，检索全靠语义相似度，精度不够。

V2.0 的改进：建一个法律知识图谱（Neo4j），把法条、法律概念、适用场景、关联案例串起来。

from neo4j import GraphDatabase class LawGraph: def __init__(self, uri, user, password): self.driver = GraphDatabase.driver(uri, auth=(user, password)) def query_related(self, law_article: str) -> List[dict]: """查询某法条关联的场景、概念和案例""" with self.driver.session() as session: result = session.run(""" MATCH (l:LawArticle {article: $article}) OPTIONAL MATCH (l)-[:APPLIES_TO]->(s:LegalScenario) OPTIONAL MATCH (l)<-[:BASED_ON]-(c:Case) OPTIONAL MATCH (l)-[:RELATES_TO]->(concept:LegalConcept) RETURN l.article AS article, l.content AS content, collect(DISTINCT s.name) AS scenarios, collect(DISTINCT c.case_id) AS cases, collect(DISTINCT concept.name) AS concepts """, article=law_article) return result.single().data()

图谱结构（三元组示例）：

劳动合同法第82条 — [适用于] → 未签劳动合同 劳动合同法第82条 — [适用于] → 双倍工资请求 劳动合同法第82条 — [关联] → 经济补偿金 张三诉XX公司案 — [依据] → 劳动合同法第82条 张三诉XX公司案 — [关键词] → 未签劳动合同

这种关联比向量检索精确得多。问"未签合同能拿多少钱"，图谱直接找到第82条和关联案例，不需要语义匹配。

更关键的是法条变更追踪：

def check_law_update(law_article: str) -> Optional[str]: """检查法条是否有新版本""" with graph.session() as session: result = session.run(""" MATCH (l:LawArticle {article: $article}) WHERE l.effective_date < date() AND l.superseded_by IS NOT NULL RETURN l.superseded_by AS new_article """, article=law_article) record = result.single() return record["new_article"] if record else None

当法条被新法替代时，图谱能直接溯源到最新版本，避免引用已失效法条。

6.4 Case Memory：跨会话的案件上下文

V1.0 的问题：用户的咨询不是一次性完成的。

第一天："公司拖欠工资" → 系统抽取结构化信息 第二天："公司让我签离职协议" → 系统重新抽取，不知道是同一个人同一个案件 第三天："仲裁申请怎么写" → 再次重新抽取

每次都要 LLM 重新做法案要素抽取，浪费 token 不说，还丢失了上下文连贯性。

V2.0 的改进：引入 Case Memory 层，用独立的持久化存储保存用户的案件上下文。

class CaseMemory: """跨会话的案件记忆""" def __init__(self): self.db = DuckDB("case_memory.db") # 轻量本地存储 self.db.execute(""" CREATE TABLE IF NOT EXISTS case_memory ( session_id VARCHAR, case_id VARCHAR, structured_data JSON, created_at TIMESTAMP, updated_at TIMESTAMP, is_active BOOLEAN DEFAULT TRUE ) """) def get_active_case(self, session_id: str) -> Optional[dict]: row = self.db.execute( "SELECT structured_data FROM case_memory " "WHERE session_id = ? AND is_active = TRUE " "ORDER BY updated_at DESC LIMIT 1", [session_id] ).fetchone() return json.loads(row[0]) if row else None def save_case(self, session_id: str, structured: dict): """持久化结构化结果""" case_id = structured.get("case_id", str(uuid4())) self.db.execute(""" INSERT INTO case_memory (session_id, case_id, structured_data, created_at, updated_at) VALUES (?, ?, ?, NOW(), NOW()) ON CONFLICT (case_id) DO UPDATE SET structured_data = EXCLUDED.structured_data, updated_at = NOW() """, [session_id, case_id, json.dumps(structured)])

在 LangGraph 的入口处先查 Memory，有活跃案件则跳过 Stage 1 的结构化抽取：

def route_with_memory(state: LegalCaseState) -> str: """如果有活跃的案件记忆，跳过结构化抽取直接进入召回""" memory = CaseMemory() active = memory.get_active_case(state["session_id"]) if active and is_same_case(active, state["user_input"]): state["structured"] = active return "skip_extraction" # 直接跳到召回 return "extract"

这一层用 DuckDB 就够了——数据量小（用户维度的案件记录），不需要独立部署数据库服务。每个用户一个 session，每个 session 保留最近一个活跃案件的上下文。

6.5 MCP Tool Layer：接入外部数据源

V1.0 的问题：所有知识都来自自有知识库。但法律场景中很多信息需要实时查询外部数据源——公司工商信息、法院公告、律师执业信息、裁判文书网最新判例。

V2.0 的改进：通过 MCP（Model Context Protocol）标准协议接入外部工具。

Agent 推理层 ↓ MCP Tool Layer ├── 企业信息查询（天眼查 / 企查查 API） ├── 法院公告查询 ├── 律师执业信息查询 ├── 裁判文书网检索 └── 工商注册信息查询

在 LangGraph 中，这些工具注册为可被 Agent 调用的节点：

from langchain_core.tools import tool @tool def query_company_info(company_name: str) -> dict: """查询企业的工商注册信息""" # 通过 MCP 协议调用天眼查 API return mcp_call("tianyancha", {"keyword": company_name}) @tool def query_lawyer_info(name: str, region: str) -> List[dict]: """查询某地区的律师执业信息""" return mcp_call("lawyer_db", {"name": name, "region": region}) @tool def check_court_announcement(case_id: str) -> Optional[dict]: """查询法院公告信息""" return mcp_call("court_api", {"case_id": case_id}) # 在 Legal Reasoner 节点中，Agent 根据场景自动调用工具 def legal_reasoner(state: LegalCaseState) -> LegalCaseState: """联合推理：综合法条、案例、外部数据生成建议""" # Agent 根据推理需要自动调用工具 # 比如用户提到某公司 → Agent 自动调用 query_company_info # 比如需要推荐律师 → Agent 自动调用 query_lawyer_info result = reasoning_chain.invoke({ "user_case": state["user_input"], "laws": state["law_candidates"], "cases": state["reranked_top5"], "tools": [query_company_info, query_lawyer_info, check_court_announcement], }) return {"output": result}

MCP 的核心价值不是"能调外部 API"——这并不新鲜——而是标准化的工具定义和发现协议。未来法律科技生态中，法院、律所、数据服务商都暴露 MCP Server，Agent 可以即插即用，不用为每个数据源写定制集成代码。

V2.0 架构全景

用户 │ ▼ ┌──────────┐ │ Case │ ← 跨会话记忆 │ Memory │ └────┬─────┘ │ ▼ LangGraph Orchestrator │ │ │ ▼ ▼ ▼ 案件理解 法条检索 案例检索 (图谱) (Law RAG) (Case RAG) │ ▼ 知识图谱 (Neo4j) 法条 ↔ 场景 ↔ 案例 │ ▼ CrossEncoder (50→20) │ ▼ LLM 精排 (20→5) │ ▼ Legal Reasoner (联合推理 + MCP Tool) │ ▼ Guardrail │ ▼ 输出

V1.0 到 V2.0 的五个升级，本质上是同一个认知转变：

法律 AI Agent 不是搜索引擎。它不是在找"最像"的东西，而是在做"法律推理"。

这个转变体现在每一个改动里——法条优先于案例、知识图谱替代纯语义检索、CrossEncoder 加上 LLM 做两级精排而非让 LLM 独自扛、Memory 让 Agent 记住案情而非每次都重新理解、工具调用让 Agent 能核实信息而非全靠知识库。

如果说法学是一个"法条 × 事实 × 经验"的三元推理，V1.0 只解决了"事实 → 经验"这一条路径，V2.0 才补齐了另外两条。

层级	V1.0 选型	V2.0 升级	作用
工作流编排	LangGraph StateGraph	LangGraph + Memory 路由	多阶段流转 + 条件回退 + 重试控制
结构化抽取	LLM + Pydantic Output Parser	同上 + Case Memory 跳过	口语化→结构化法律要素
法条检索	ES 精确查询	Law RAG + 知识图谱 (Neo4j)	法条依据优先召回
标签过滤	Elasticsearch	同上	案由/要素筛选，千万级→万级
向量检索	Qdrant + bge-m3	同上	语义相似度召回
初排	无	CrossEncoder (bge-reranker-v2)	低成本 50→20 过滤
精排	LLM 看 50 个 → TOP 5	LLM 看 20 个 → TOP 5	多维评分
联合推理	案例→策略建议	法条 + 案例 + MCP 工具联合推理	符合成文法体系
法条校验	ES 精确查询	+ 图谱变更追踪	确保法条版本准确
跨会话记忆	无	DuckDB Case Memory	同案件多轮对话不重复抽取
外部工具	无	MCP Tool Layer	企业查询/法院公告/律师库
Guardrail	LangGraph Condition + 规则	同上 + 法条版本校验	三层校验链