用 LLM 处理非结构化数据，结果不可重现怎么办？Fenic 把 AI 分析变成可复用的 DataFrame 管道

你有没有遇到过这种情况——用 ChatGPT 或 Claude 分析一批客户反馈，得到了几个分类结果，但下周换了个提示词、换了个模型版本，结果完全不同？或者 AI 编程代理跑了一次日志分析，找到了一些有用的模式，但你想让它再跑一次的时候，发现那个分析脚本早就不见了？

这是目前非结构化数据处理的普遍困境：AI 的一次性分析很容易，但把分析变成可重复、可审查的可靠流程很难。

为什么这是个真问题

大多数团队处理非结构化数据（日志、工单、转录、评估跟踪）的方式是「写一个提示词 → 跑一次 → 看看结果 → 下次重来」。这种方式有几个硬伤：

• 不可重现：下次跑结果可能不一样——是模型变了还是数据变了？不知道。
• 不可审查：AI 编程代理跑了一堆数据，用它自己的方式得出了一些结论，但你没法逐行审查它是怎么做的。
• 不可演进：上周发现的模式没法直接复用到本周的数据上，每次都要从头来一遍。

传统方案是写正则表达式或者 Prefect 脚本，但正则处理不了语义理解，而 Prefect 脚本和提示词是两套体系——你得来回切换上下文。

Fenic 的做法不一样：把 AI 操作做成 DataFrame 的一等公民。提取(extract)、分类(classify)、摘要(summarize)、嵌入(embed)——这些以前需要单独调 LLM 的事情，现在成了 DataFrame 查询中的普通算子。结果是一个类型安全、可审查、可重跑的管道（pipeline），而不是一段死掉的聊天记录。

Fenic 是什么

Fenic（GitHub 463⭐，Apache-2.0，Python）是一个语义 DataFrame 引擎。你写 PySpark 或 SQL 风格的操作——select, filter, join, group_by, agg——同时在同一个查询里调用 LLM 算子。模型在 Session 上配置一次，管道延迟构建，Fenic 编译后在专为推理优化的引擎上运行：自动批处理、限速、重试、token/成本记账和响应缓存。

核心区别就两点：

1. 推理活在查询模型内部——提取、分类、摘要、嵌入是有 Schema 和类型的算子，不是你自己手写的零散 API 调用
2. 管道本身就是产物——因为分析工作是用类型化的算子表达的，它天然就是可审查的（行级血缘、explain、每次查询的指标）、可重跑的（惰性计划 + 缓存）、并且可以提升为一个 MCP 工具供 AI 编程代理调用

快速上手：一条命令安装

pip install fenic

Fenic 支持多种 LLM 提供商：

export OPENAI_API_KEY=sk-...        # OpenAI
export ANTHROPIC_API_KEY=sk-ant-...  # Anthropic Claude
export GOOGLE_API_KEY=...            # Google Gemini
export COHERE_API_KEY=...            # Cohere Embeddings
export OPENROUTER_API_KEY=...        # OpenRouter

如果你用 AI 编程代理（Claude Code、Cursor、Codex）写 Fenic 代码，先运行：

fenic skill install

这会把 Fenic 的 API 规则安装到代理的技能目录中，让它写代码时不会猜错 API 签名。写完后再用 fenic check pipeline.py 静态检查一下——不执行脚本，只检查 namespace 和 import 错误。

实战案例 1：从工单文本提取结构化信息

假设你有上百条客户反馈，想从中提取产品、情感和问题类别。传统做法是写一个 ChatGPT 提示词，逐条粘贴，然后手动整理成表格。Fenic 的做法是定义一个 Pydantic 模型，然后一行算子搞定：

import fenic as fc
from pydantic import BaseModel, Field

class Ticket(BaseModel):
    product: str = Field(description="用户提到的产品")
    sentiment: str = Field(description="positive、neutral 或 negative")
    issue: str = Field(description="用户问题的单行摘要")

session = fc.Session.get_or_create(
    fc.SessionConfig(
        app_name="ticket_analysis",
        semantic=fc.SemanticConfig(
            language_models={
                "mini": fc.OpenAILanguageModel(model_name="gpt-4o-mini", rpm=500, tpm=200_000)
            },
        ),
    )
)

df = session.create_dataframe([
    {"id": 1, "text": "自从上次更新后，报表导出一直超时。"},
    {"id": 2, "text": "新面板很赞，但移动端 SSO 登录还是坏的。"},
])

tickets = (
    df.select("id", fc.semantic.extract("text", Ticket).alias("t"))
      .unnest("t")
)
tickets.show()

关键点：这里 extract 不是一个黑盒提示词——它有 Pydantic Schema 约束，每个字段有明确的类型和描述。你可以 tickets.explain() 查看执行计划，也可以 tickets.lineage() 追溯任一行来自哪条原始数据。

实战案例 2：语义 Join 匹配简历和职位

有时候你需要跨表做「语义匹配」而不是精确匹配——比如把候选人的简历和岗位描述按「契合度」做关联。Fenic 的 semantic.join 用自然语言谓词（predicate）代替了 ON key = key：

matches = candidates.semantic.join(
    roles,
    predicate=(
        "Candidate background: {{ left_on }}\n"
        "Role requirements: {{ right_on }}\n"
        "The candidate is a strong fit for the role."
    ),
    left_on=fc.col("resume"),
    right_on=fc.col("job_description"),
)

ranked = (
    matches.group_by("role_id")
    .agg(fc.count("*").alias("n_candidates"))
    .order_by(fc.desc("n_candidates"))
)

这个 join 不是拼关键字匹配——它让 LLM 判断两个文本是否「语义匹配」，这在简历筛选、支持工单路由、日志关联等场景中极其有用。

实战案例 3：把分析管道变成 MCP 工具给代理调用

这是 Fenic 最独特的能力——你写好的分析管道可以一键变成 MCP 工具，供同一个代理或其他代理调用。比如把前面工单分析的结果注册为一个可查询的表：

from fenic import SystemToolConfig

session.catalog.set_table_description(
    "tickets", "客户反馈工单的结构化分析结果（产品、情感、问题类别）"
)

server = fc.create_mcp_server(
    session,
    "Ticket Intelligence",
    system_tools=SystemToolConfig(
        table_names=["tickets"],
        tool_namespace="support",
        max_result_rows=100,
    ),
)

fc.run_mcp_server_sync(server, transport="http", port=8000)

或者定义一个带参数的工具：

negative_tickets = session.table("tickets").filter(
    fc.col("sentiment") == fc.tool_param("sentiment", fc.StringType)
)

session.catalog.create_tool(
    "tickets_by_sentiment",
    "按情感分类查询客户反馈",
    negative_tickets,
    tool_params=[fc.ToolParam(name="sentiment", description="筛选的情感类别")],
)

然后通过 CLI 直接暴露：

fenic-serve --app-name ticket_analysis --port 8000

这样 AI 编程代理就可以通过 MCP 协议直接查询这个分析结果了——不需要重复跑一次分析逻辑。

常用语义算子一览

Fenic 内置了 9 个列级语义算子和 3 个 DataFrame 级算子：

所有算子都可以和标准 DataFrame 操作（select, filter, with_columns, join, group_by/agg, order_by, limit）自由组合，也支持完整的 SQL 查询。

最佳实践

1. 先用 fenic skill install 再让代理写代码：Fenic 的 API 相对新，AI 编程代理靠训练数据猜不准。装技能文件能减少 80% 的命名空间错误。
2. 善用缓存：Fenic 自带 LLM 响应缓存。在探索阶段，同样的文本不会被反复调用 API 消耗 token。
3. explain() 和 lineage() 是你的朋友：写复杂管道时，先用 explain() 看执行计划确认逻辑，再用 lineage() 追溯最终结果的每一行来源。
4. 从简单的 Pydantic Schema 开始：刚开始不追求完美字段定义——先二三个关键字段跑通，再逐步加。Schema 变了管道不变，这是类型化设计的优势。
5. 成本控制：在 SessionConfig 中设好 rpm（每分钟请求数）和 tpm（每分钟 token 数），Fenic 会自动限速不会超支。

总结

Fenic 解决的不是「怎么调用 LLM」的问题——那是 API 文档的事。它解决的是「怎么让 LLM 处理数据的结果变成工程化资产」的问题。当你需要让 AI 编程代理重复执行某个数据分析任务、或者需要让团队其他成员审查和复用一个分析流程时，Fenic 的 DataFrame 抽象比提示词粘贴板和零散脚本要可靠得多。

• 项目地址：github.com/typedef-ai/fenic
• 许可证：Apache-2.0
• 安装：pip install fenic
• 适用场景：日志分析、支持工单分类、评估跟踪、文档处理、简历匹配

相关链接

• Fenic GitHub
• Fenic 示例仓库
• PyPI: fenic