Customer Support Langgraph

客户支持LangGraph

用于处理客户支持查询的模块化、可扩展的Python管道。它验证输入，提取实体，从知识库（ChromaDB+SentenceTransformers）中检索答案，做出决策（解决与升级），并产生最终响应。它同时提供CLI工作流和FastAPI web服务器。

• 内置：Pydantic、ChromaDB、句子转换器、FastAPI
• 通过管道代理执行的YAML定义的阶段进行编排
• 知识库：基于嵌入的持久ChromaDB集合检索

______________________________________________________________________

目录

1. 特性
2. 架构概述
3. 项目结构
4. 安装
5. 快速开始
6. 运行CLI
7. 运行FastAPI服务器
8. 配置（stages.yaml）
9. 知识库与摄入
10. 管道内部
11. API终点
12. 测试
13. 故障排除
14. 开发技巧
15. 许可证

______________________________________________________________________

1） 特点

• 具有确定性、条件性和非确定性阶段的模块化管道
• 基于ChromaDB和句子转换器的知识库检索
• 输入有效载荷的验证
• 具有升级功能的简单决策逻辑
• 控制台和文件的结构化日志记录
• CLI和FastAPI集成
• 每个阶段都可以轻松扩展“能力”

______________________________________________________________________

2） 架构概述

• 使用Pydantic验证输入(InputPayload).
• 管道(LangGraphAgent)从以下位置读取阶段定义 config/stages.yaml.
• 每个阶段通过MCP式路由器运行一个或多个“能力”(mcp_client.py)，它在中调用本地函数 src/langie/abilities.py.
• 知识库检索使用 Retriever 由ChromaDB+句子转换器支持。
• 非确定性阶段（DECIDE）评估并可能升级。
• 最终输出是一个结构化的有效载荷，包括实体、分数、标志和响应。

______________________________________________________________________

3） 项目结构

.
├── app.py                         # FastAPI app with / and /chat endpoints
├── config/
│   └── stages.yaml                # Pipeline stages configuration
├── data/
│   ├── kb_faq.json                # FAQ seed data for ingestion
│   ├── tickets.json               # Ticket history (updated by FastAPI)
│   └── chroma/                    # Persisted ChromaDB store (created after ingest)
├── logs/
│   ├── pipeline.log               # Runtime logs
│   └── pipeline_test.log          # Test logs
├── pipeline/
│   └── abilities/
│       └── knowledge_base_search.py  # (legacy path used by the FastAPI app)
├── scripts/
│   ├── kb_ingest.py               # Ingest FAQ JSON into ChromaDB
|   └── run.sh                     # Convenience script to start the FastAPI server
├── src/langie/
│   ├── __main__.py                # python -m src.langie entrypoint
│   ├── abilities.py               # Ability implementations
│   ├── cli.py                     # CLI wrapper to run the pipeline
│   ├── logger.py                  # Logger configuration (console + file)
│   ├── mcp_client.py              # Ability router: COMMON/ATLAS + KB fallback
│   ├── models.py                  # Pydantic models (InputPayload)
│   ├── pipeline.py                # LangGraphAgent: loads YAML, executes stages
│   └── retriever.py               # ChromaDB + SentenceTransformers retriever
├── static/
│   └── index.html                 # Simple UI page (served under /static)
├── test_insertDB.py               # Add FAQ and rebuild ChromaDB demo
├── test_out_of_scope.py           # OOD retrieval test (uses legacy method name)
├── test_pipeline.py               # Pipeline smoke test
├── test_retriever.py              # Retrieval test (uses legacy method name)
├── pyproject.toml                 # Build metadata
├── requirements.txt               # Runtime dependencies
└── README.md                      # This document

注意：一些测试调用 Retriever.retrieve(...) 而当前的实现暴露了 search(...)。如果按原样运行这些测试，请相应调整或添加一个小适配器。

______________________________________________________________________

4） 安装

1. 创建干净的虚拟环境：

python3 -m venv venv
source venv/bin/activate

1. 安装依赖项：

pip install -r requirements.txt

1. 验证Python版本：

• 需要Python 3.10+（请参阅 pyproject.toml).推荐使用Python 3.11。

______________________________________________________________________

5） 快速入门

1. 吸收知识库（创建 data/chroma 带有嵌入）：

python scripts/kb_ingest.py

1. 使用默认示例运行CLI：

python -m src.langie run --config config/stages.yaml

1. 启动FastAPI服务器：

uvicorn app:app --reload --port 8000

打开http://localhost:8000/static/index.html

______________________________________________________________________

6） 运行CLI

命令：

python -m src.langie run --config config/stages.yaml [--input path/to/input.json] [--debug]

• --config：管道YAML的路径。
• --input：可选JSON有效载荷文件。如果省略，则使用内置示例。
• --debug：启用详细日志记录。

示例输入JSON：

{
  "customer_name": "Alice",
  "email": "alice@example.com",
  "query": "My order #123 hasn’t arrived",
  "priority": "High",
  "ticket_id": "TKT-5678"
}

发生了什么：

• 负载 stages.yaml
• 使用验证有效载荷 InputPayload
• 每个阶段的运行能力
• 将最终状态JSON打印到stdout

______________________________________________________________________

7） 运行FastAPI服务器

开始：

uvicorn app:app --reload --port 8000

终点：

• 获取 / → 供应 static/index.html
• 发布 /chat → 接受姓名/电子邮件/查询，运行KB搜索，返回带响应的工单

示例 curl:

curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{
        "customer_name": "Bob",
        "email": "bob@example.com",
        "query": "What is your refund policy?"
      }'

响应将包括：

• ticket_id
• response
• alternatives
• status （已解决/待定）
• timestamp

门票存放在 data/tickets.json.

______________________________________________________________________

8） 配置（stages.yaml）

config/stages.yaml 定义管道阶段和能力。模式：

• 确定性：按顺序运行能力
• 非确定性：执行评估然后分支（例如，升级）

摘录：

stages:
  - name: INTAKE
    mode: deterministic
    abilities:
      - { name: accept_payload, server: COMMON }

  - name: UNDERSTAND
    mode: deterministic
    abilities:
      - { name: parse_request_text, server: COMMON }
      - { name: extract_entities,   server: ATLAS }

  - name: RETRIEVE
    mode: deterministic
    abilities:
      - { name: knowledge_base_search, server: ATLAS }
      - { name: store_data,            server: COMMON }

  - name: DECIDE
    mode: non-deterministic
    abilities:
      - { name: solution_evaluation, server: COMMON }
      - { name: escalation_decision, server: ATLAS }
      - { name: update_payload,      server: COMMON }

  - name: CREATE
    mode: deterministic
    abilities:
      - { name: response_generation, server: COMMON }

• 能力被路由到 COMMON 或 ATLAS 通过 mcp_client.py.
• knowledge_base_search 通过中的检索器快捷方式处理 pipeline.py.

______________________________________________________________________

9） 知识库与摄入

• 来源常见问题解答： data/kb_faq.json
• 摄入： scripts/kb_ingest.py 读取JSON并使用嵌入的SentenceTransformers将其索引到ChromaDB中。

跑步摄入：

python scripts/kb_ingest.py

添加常见问题解答并重新构建（示例脚本）：

python test_insertDB.py

检索器实现（简化）：

# Python example showing how the retriever works
from src.langie.retriever import Retriever

# Create retriever (persisted DB)
retriever = Retriever(db_path="data/chroma", collection_name="faq")

# Search top-3 results for a query
hits = retriever.search("How do I get a refund?", top_k=3)

# Each hit contains question, answer, doc, and a distance-based score
for h in hits:
    print(h["answer"])  # comment: prints the retrieved answer text

笔记：

• “全MiniLM-L6-v2”的初始下载发生一次并被缓存。
• 确保 data/chroma/ 可写且存在（通过摄入创建）。

______________________________________________________________________

10） 管道内部

核心类： src/langie/pipeline.py → LangGraphAgent

• 加载YAML配置
• 通过以下方式验证有效载荷 InputPayload
• 执行阶段和能力
• 将结果合并到共享 state
• 提供结构化日志和摘要

关键能力亮点(src/langie/abilities.py):

• accept_payload：确保存在所需的密钥和默认结构。
• parse_request_text：对查询进行标记，提取简单模式（例如订单ID）。
• extract_entities：启发式意图/问题/产品提取。
• normalize_fields：规范电子邮件、优先级和清理订单ID格式。
• enrich_records：添加SLA/历史元数据。
• clarify_question / extract_answer / store_answer：针对缺失细节的模拟问答循环。
• knowledge_base_search：没有操作能力（管道本身已经注入KB结果）。
• store_data：衍生品 kb_hits 和 kb_top_answer 根据检索结果。
• solution_evaluation：根据KB点击量分配0-100分。
• escalation_decision：如果分数\=阈值，则将票状态标记为已解决。

1. 把票坚持到 data/tickets.json. - 答复：

    {
      "ticket_id": "TKT-00X",
      "customer_name": "Alice",
      "email": "alice@example.com",
      "query": "Where is my order?",
      "response": "KB answer...",
      "alternatives": [{"answer": "...", "score": 0.12}],
      "status": "resolved",
      "timestamp": "2025-..."
    }

注：网络流使用简化的能力类 pipeline/abilities/knowledge_base_search.py 而不是全部 LangGraphAgent 管道。CLI使用基于YAML的管道。

______________________________________________________________________

12） 测试

• test_pipeline.py:

- 烟雾测试 LangGraphAgent 阅读 config/stages.yaml 并端到端执行。 - 打印最终输出并断言基础知识。

• test_retriever.py 和 test_out_of_scope.py:

- 搜索知识库的演示。 - 这些调用 retriever.retrieve(...)，但目前 Retriever 暴露 search(...). - 您可以： - 替换 retrieve(...) 随着 search(...) 在测试中，或 - 添加一个小型适配器方法 Retriever:

      # Python: Optional adapter for backward compatibility
      class Retriever(...):
          def retrieve(self, query: str, top_k: int = 3):
              # comment: call the new search method for legacy tests
              return self.search(query, top_k=top_k)

运行测试：

pytest

______________________________________________________________________

13） 故障排除

• SentenceTransformers模型下载缓慢：

- 首次下载“全MiniLM-L6-v2”可能需要一些时间。它随后被缓存。

• ChromaDB目录缺失：

- 跑 python scripts/kb_ingest.py 创建/填充 data/chroma/.

• KB搜索没有结果：

- 验证 data/kb_faq.json 有效，摄入成功。

• 由于以下原因测试失败 retrieve 兽医 search:

- 更新要使用的测试 search，或添加上面显示的适配器方法。

• 日志记录不可见：

- 使用 --debug 在CLI或tail中 logs/pipeline.log.

______________________________________________________________________

14） 开发技巧

• 在哪里添加新能力：

- 实施中 src/langie/abilities.py 作为一个函数：

    # Python: new ability example with comments
    def my_new_ability(state: dict) -> dict:
        # comment: read from state
        text = state.get("query", "")

        # comment: do something simple
        if "coupon" in text.lower():
            state.setdefault("entities", {})["has_coupon"] = True

        # comment: return full state dict
        return state

- 注册 src/langie/mcp_client.py 在...之下 COMMON_ABILITY_MAP 或 ATLAS_ABILITY_MAP. - 参考文献 config/stages.yaml:

    - name: CUSTOM
      mode: deterministic
      abilities:
        - { name: my_new_ability, server: COMMON }

• 如何改变决策逻辑：

- 修改 solution_evaluation 在……里面 abilities.py 以反映你的得分策略。

• 如何自定义响应：

- 编辑 response_generation 设置回复格式或使用模板。

______________________________________________________________________

15） 许可证

该项目根据MIT许可证获得许可。有关详细信息，请参阅LICENSE文件。