Elixir版LangChain实战：函数式AI Agent开发与生产部署指南-平芜编程栈

1. 项目概述：Elixir生态中的LangChain

如果你是一名Elixir开发者，正琢磨着如何把ChatGPT、Claude这些大语言模型（LLM）的能力，像搭积木一样轻松地集成到你的应用里，那么brainlid/langchain这个库，就是你一直在找的那块“积木”。简单来说，它是一个专为Elixir语言设计的LangChain实现。LangChain这个概念，最初在Python和JavaScript社区火起来，核心思想是“链”（Chain），即把LLM当作一个核心组件，通过“链”的方式，把它和你应用里的其他功能（比如数据库查询、API调用、自定义业务逻辑）连接起来，从而构建出能理解、推理并执行复杂任务的智能应用。

brainlid/langchain的价值在于，它没有生硬地照搬Python/JS那套面向对象的设计，而是充分拥抱了Elixir的函数式编程哲学。这意味着你在使用它时，感受到的是Elixir特有的管道操作符（|>）带来的流畅感，以及进程（Process）和消息传递（Message）构建的清晰数据流。它不是一个简单的API客户端包装，而是一个完整的框架，提供了从模型调用、提示词管理、工具函数定义到执行链编排的全套工具。对于Elixir开发者而言，这大大降低了构建AI驱动应用的门槛，让你能更专注于业务逻辑，而不是在HTTP请求和JSON解析的细节里打转。

2. 核心设计理念与架构解析

2.1 为什么是Elixir？函数式与并发的天然优势

在深入代码之前，理解这个库的设计哲学至关重要。原版的LangChain（Python/JS）诞生于LLM以“补全”（Completion）模式为主的时代，设计上需要大量手动管理对话历史。而brainlid/langchain诞生于“聊天”（Chat）模型成为主流的时代，它直接拥抱了Message（系统消息、用户消息、助手消息）作为一等公民的设计。更重要的是，Elixir的函数式、不可变数据和Actor模型（通过OTP），与AI Agent（智能体）的概念有着惊人的契合度。

一个AI Agent本质上就是一个可以自主调用工具、与环境交互的进程。在Elixir中，你可以很自然地将一个LLMChain封装在一个GenServer里，让它拥有自己的状态（如对话历史、工具列表），并通过消息来驱动其推理循环。这种设计让构建稳定、可容错、可水平扩展的AI Agent系统变得非常直观。库本身不强制你采用某种架构，但它提供的模块化组件（ChatModel,Message,Function,Chain）能完美融入你现有的OTP应用体系。

2.2 核心模块拆解：从模型到执行链

库的核心模块结构清晰，围绕“链”的概念展开：

LangChain.ChatModels：这是与各种AI服务对话的抽象层。它定义了统一的ChatModel行为（Behaviour），然后为每个支持的提供商（如OpenAI、Anthropic、Ollama）提供了具体的实现模块（如ChatOpenAI，ChatAnthropic）。这种设计让你在开发时，可以面向统一的接口编程，运行时再通过配置决定使用哪个模型，极大地提升了代码的可移植性。
LangChain.Message：表示对话中的一条消息。它不仅是简单的文本容器，还支持多模态内容（ContentParts），可以包含文本、图像、文件甚至“思考”块。消息有不同的角色（:user,:assistant,:system,:tool），这是构建有效对话上下文的基础。
LangChain.Function：这是连接LLM和你的Elixir应用世界的桥梁。你可以将一个普通的Elixir函数包装成Function，定义好它的名称、描述和参数JSON Schema。LLM在推理过程中，如果认为需要调用这个函数来获取信息或执行操作，就会生成一个符合Schema的调用请求，库则会安全地执行你的Elixir函数并返回结果。这是实现“AI Agent”能动性的关键。
LangChain.Chains.LLMChain：整个库的“发动机”。一个链（Chain）将上述所有组件组合在一起：它持有一个ChatModel实例、一个消息列表（对话历史）、一个可用的Function工具列表，以及一些运行配置（如verbose模式）。当你运行一个链时，它会管理整个与LLM交互的循环：发送消息和上下文，接收LLM的响应，解析其中可能包含的工具调用请求，执行工具，将工具结果作为新消息追加，然后根据策略（如mode: :while_needs_response）决定是否继续询问LLM，直到获得最终答案。

这种架构的优势在于高度的解耦和可组合性。你可以轻松地替换模型提供商、动态增减工具、或者将多个链串联起来形成更复杂的工作流。

3. 环境配置与模型接入实战

3.1 基础安装与依赖管理

首先，在你的Elixir项目的mix.exs文件中添加依赖。库底层使用Req这个优秀的HTTP客户端来处理网络请求。

def deps do [ {:langchain, "~> 0.8.0"}, # Req是必须的，但LangChain已经将其声明为依赖，你通常无需显式添加 # {:req, "~> 0.4.0"} ] end

运行mix deps.get获取依赖后，接下来就是配置API密钥。绝对不要将密钥硬编码在代码中。推荐的方式是使用环境变量，并通过Elixir的配置系统来读取。

3.2 多模型服务商配置详解

库支持众多模型，每种模型的配置方式类似但略有不同。我们以最常用的OpenAI和Anthropic为例，展示如何在config/runtime.exs（或config/config.exs）中安全配置。

# config/runtime.exs import Config # 方式一：直接从系统环境变量读取（推荐用于生产环境） config :langchain, openai_key: System.fetch_env!("OPENAI_API_KEY"), openai_org_id: System.get_env("OPENAI_ORG_ID") # 组织ID可选 config :langchain, :anthropic_key, System.fetch_env!("ANTHROPIC_API_KEY") config :langchain, :xai_api_key, System.fetch_env!("XAI_API_KEY") # 方式二：使用函数或模块函数动态获取（更灵活） config :langchain, openai_key: {MyApp.Secrets, :fetch_openai_key, []}, openai_org_id: fn -> System.get_env("OPENAI_ORG_ID") end # 在MyApp.Secrets模块中 defmodule MyApp.Secrets do def fetch_openai_key do # 可以从Vault、KMS或其他秘密管理服务获取 System.fetch_env!("OPENAI_API_KEY") end end

注意：密钥安全是重中之重。在部署平台（如Fly.io, Render, Heroku）上，务必使用其秘密管理功能。例如在Fly.io上：fly secrets set OPENAI_API_KEY=sk-xxx。本地开发可以使用.env文件配合dotenvy库，但确保.env在.gitignore中。

3.3 本地与开源模型集成：Ollama与Bumblebee

除了云端API，库还强大地支持本地运行的模型，这对成本控制、数据隐私和离线开发至关重要。

Ollama集成：Ollama是运行本地LLM（如Llama 3, Mistral, Gemma）的绝佳工具。配置非常简单，只需指向本地Ollama服务的端点。

alias LangChain.ChatModels.ChatOllama {:ok, llm} = ChatOllama.new(%{ model: "llama3.2:3b", # Ollama中的模型名称 endpoint: "http://localhost:11434/api/chat", # Ollama默认的聊天API端点 temperature: 0.7 })

Bumblebee集成：这是Elixir机器学习生态Nx/Bumblebee的深度整合。你可以直接加载Hugging Face上的模型，并在本地用GPU/CPU进行推理。这提供了最高的灵活性和控制权。

# 首先，你需要用Bumblebee设置一个Nx.Serving defmodule MyApp.LlamaServing do @model_id "meta-llama/Llama-3.2-3B-Instruct" def start_link(_opts) do {:ok, model_info} = Bumblebee.load_model({:hf, @model_id}) {:ok, tokenizer} = Bumblebee.load_tokenizer({:hf, @model_id}) serving = Bumblebee.Text.conversation(model_info, tokenizer, max_new_tokens: 512, defn_options: [compiler: EXLA] # 使用EXLA加速（如有GPU） ) |> Nx.Serving.new(streaming: true) Nx.Serving.start_link(name: __MODULE__, serving: serving) end end # 在你的应用监督树中启动上述服务 children = [ {MyApp.LlamaServing, []} ] # 然后在LangChain中使用它 alias LangChain.ChatModels.ChatBumblebee {:ok, llm} = ChatBumblebee.new(%{ serving: MyApp.LlamaServing, # 指向启动的Serving进程名 stream: true # 支持流式响应 })

使用Bumblebee需要更多的机器学习栈知识，但它让在Elixir生态内实现端到端的AI功能成为可能，无需依赖外部HTTP服务。

4. 核心功能实现与代码实战

4.1 构建第一个对话链：从Hello World到复杂交互

让我们从一个最简单的例子开始，逐步增加复杂度。

基础对话：

alias LangChain.ChatModels.ChatOpenAI alias LangChain.Chains.LLMChain alias LangChain.Message # 1. 创建模型实例 {:ok, chat} = ChatOpenAI.new(%{model: "gpt-4o-mini", temperature: 0.7}) # 2. 创建链，并关联模型 {:ok, chain} = LLMChain.new(%{llm: chat}) # 3. 添加用户消息 chain = LLMChain.add_message(chain, Message.new_user!("用Elixir写一个Hello World程序。")) # 4. 运行链，获取AI回复 {:ok, updated_chain} = LLMChain.run(chain) # 5. 提取并打印回复 last_message = List.last(updated_chain.messages) IO.puts("AI回复：#{last_message.content}") # 输出大致为：AI回复：```elixir\nIO.puts("Hello, World!")\n```

多轮对话与上下文管理：链会自动维护消息历史。

# 接上例，updated_chain已经包含了第一轮对话 chain_2 = LLMChain.add_message(updated_chain, Message.new_user!("很好！现在修改它，让程序从环境变量`GREETING`读取问候语，如果不存在则使用默认值'Hello'。")) {:ok, final_chain} = LLMChain.run(chain_2) # 此时final_chain.messages包含四轮消息：用户1，助手1，用户2，助手2。 # AI能理解“它”指的是之前的代码，并给出修改后的版本。

4.2 赋予AI“手脚”：自定义工具函数（Function Calling）

这是LangChain最强大的特性之一。我们可以让AI调用我们写的Elixir函数。

假设我们有一个简单的天气服务：

defmodule MyApp.Weather do @fake_db %{"北京" => "晴朗，25°C", "上海" => "多云，23°C", "巴黎" => "小雨，15°C"} def get(city) do case Map.fetch(@fake_db, city) do {:ok, forecast} -> {:ok, forecast} :error -> {:error, "未找到城市 #{city} 的天气信息"} end end end

现在，我们将这个函数暴露给AI：

alias LangChain.Function # 定义工具函数 weather_function = Function.new!(%{ name: "get_weather", description: "根据城市名称获取当前的天气情况。", parameters_schema: %{ type: "object", properties: %{ city: %{ type: "string", description: "城市的名称，例如：北京、巴黎、纽约。" } }, required: ["city"] }, function: fn %{"city" => city}, _context -> # 第二个参数`_context`是创建链时传入的custom_context，这里暂未使用 MyApp.Weather.get(city) end }) # 创建链并添加工具 {:ok, chain} = LLMChain.new!(%{ llm: ChatOpenAI.new!(), verbose: true # 开启详细日志，方便观察AI如何思考 }) |> LLMChain.add_tools(weather_function) |> LLMChain.add_message(Message.new_user!("今天巴黎和上海的天气怎么样？")) # 以“需要响应则继续”的模式运行。AI会先思考，发现需要调用工具，生成调用请求。 # 库会执行工具，并将结果以`tool`消息的形式追加，然后再次询问AI。 # AI收到工具结果后，综合信息给出最终答案。 {:ok, final_chain} = LLMChain.run(chain, mode: :while_needs_response) IO.puts(LangChain.Utils.ChainResult.to_string!(final_chain)) # 输出可能为：“巴黎今天有小雨，气温15°C；上海则是多云，气温23°C。”

当verbose: true时，你会在日志中看到类似如下的过程，这有助于调试AI的思考链（Reasoning Chain）：

[LLMChain] User: “今天巴黎和上海的天气怎么样？” [LLMChain] Assistant (思考): “用户问了两个城市的天气。我有`get_weather`工具。我需要分别查询巴黎和上海。” [LLMChain] Assistant (工具调用): `get_weather` with arguments `{"city": "巴黎"}` [LLMChain] Tool Result: “小雨，15°C” [LLMChain] Assistant (工具调用): `get_weather` with arguments `{"city": "上海"}` [LLMChain] Tool Result: “多云，23°C” [LLMChain] Assistant (最终回答): “巴黎今天有小雨，气温15°C；上海则是多云，气温23°C。”

4.3 高级特性：上下文（Context）与流式响应（Streaming）

上下文（Context）的安全传递：在工具函数中，我们看到了context参数。这用于安全地将用户会话、权限等信息传递给工具。

# 假设我们有一个需要用户权限的“下单”工具 defmodule MyApp.Orders do def create(order_params, user_id) do # 这里根据user_id检查权限、创建订单 {:ok, order_id} = save_order(order_params, user_id) {:ok, "订单创建成功，ID: #{order_id}"} end end order_function = Function.new!(%{ name: "create_order", description: "为用户创建一个新订单。", parameters_schema: %{ type: "object", properties: %{ product_id: %{type: "string"}, quantity: %{type: "integer", minimum: 1} }, required: ["product_id", "quantity"] }, function: fn arguments, context -> # context 是在创建链时传入的 custom_context，这里我们存了user_id user_id = context["user_id"] MyApp.Orders.create(arguments, user_id) end }) # 在创建链时注入上下文 custom_context = %{"user_id" => 123, "session_id" => "abc"} chain = LLMChain.new!(%{llm: llm, custom_context: custom_context}) |> LLMChain.add_tools(order_function)

流式响应（Streaming）：对于需要实时显示AI思考过程的场景（如聊天界面），流式响应至关重要。大部分ChatModel实现都支持stream: true选项。

alias LangChain.ChatModels.ChatOpenAI {:ok, chat} = ChatOpenAI.new(%{model: "gpt-4o", stream: true}) {:ok, chain} = LLMChain.new(%{llm: chat}) chain = LLMChain.add_message(chain, Message.new_user!("讲述一个关于Elixir的简短故事。")) # run/2 在流式模式下会返回一个元组 {:ok, stream_fun, updated_chain} {:ok, stream_fn, updated_chain} = LLMChain.run(chain, mode: :while_needs_response) # stream_fun 是一个函数，每次调用返回 {:ok, event} 或 :done Stream.resource( fn -> :ok end, fn _ -> case stream_fn.() do {:ok, %{data: data}} -> # data 可能是 :start, {:content, delta_text}, :done 等 case data do {:content, delta} -> IO.write(delta) # 逐词输出 _ -> :ok end {[data], :ok} :done -> {:halt, :ok} end end, fn _ -> :ok end ) |> Stream.run()

这样，用户就能看到AI一个字一个字“思考”出故事的过程，体验更佳。

5. 生产环境实践：测试、监控与性能优化

5.1 基于轨迹（Trajectory）的Agent行为测试

在AI应用中，测试不能只关注最终输出是否正确，因为同样的答案可能由低效甚至危险的推理路径得出。LangChain.Trajectory模块用于捕获和分析Agent执行过程中的工具调用序列，是实现可靠测试的关键。

单元测试中的断言：

defmodule MyApp.WeatherAgentTest do use ExUnit.Case use LangChain.Trajectory.Assertions # 引入断言宏 test "询问天气会触发正确的工具调用" do # 假设 setup 中已经创建了包含天气工具的链 `chain` {:ok, final_chain} = LLMChain.run(chain, mode: :while_needs_response) # 断言轨迹完全匹配预期的工具调用序列（严格顺序和参数） assert_trajectory final_chain, [ %{name: "get_weather", arguments: %{"city" => "巴黎"}}, %{name: "get_weather", arguments: %{"city" => "上海"}} ] # 也可以使用更灵活的匹配模式 # 只关心调用了某个工具，不关心参数 assert_trajectory final_chain, [ %{name: "get_weather", arguments: nil} ], mode: :superset # 实际调用序列包含至少这个调用 # 确保没有调用危险工具 refute_trajectory final_chain, [ %{name: "delete_database", arguments: nil} ], mode: :superset end end

黄金文件（Golden File）回归测试：对于复杂的Agent工作流，可以保存一次“正确”的运行轨迹作为黄金标准，后续测试与之对比，防止回归。

test "天气查询Agent行为稳定" do # 首次生成并保存黄金文件（手动执行一次） # golden_trajectory = chain |> LLMChain.run!() |> Trajectory.from_chain() # File.write!("test/fixtures/golden_weather.json", Jason.encode!(Trajectory.to_map(golden_trajectory))) # 在常规测试中加载并比较 expected = "test/fixtures/golden_weather.json" |> File.read!() |> Jason.decode!() |> Trajectory.from_map() actual_chain = LLMChain.run!(chain) actual_trajectory = Trajectory.from_chain(actual_chain) assert Trajectory.matches?(actual_trajectory, expected, mode: :unordered, args: :subset) # mode: :unordered 允许工具调用顺序不同 # args: :subset 允许实际调用的参数是黄金文件中参数的子集（即更具体） end

5.2 性能优化与成本控制

提示词缓存（Prompt Caching）：对于长上下文且前缀重复的对话（如系统指令很长），提示词缓存能显著降低Token使用量和延迟。库对支持此功能的模型（如ChatGPT、Claude、DeepSeek）进行了集成。

# 对于Claude，需要显式在消息中标记缓存控制点 alias LangChain.ChatModels.ChatAnthropic system_msg = Message.new_system!("你是一个专业的Elixir代码助手。...很长的系统指令...") # 在Claude中，`cache_control: :ephemeral` 表示此条消息及之前的消息可以被缓存 user_msg = Message.new_user!("写一个Phoenix控制器。", cache_control: :ephemeral) {:ok, chain} = LLMChain.new!(%{llm: ChatAnthropic.new!()}) |> LLMChain.add_messages([system_msg, user_msg]) |> LLMChain.run()

在后续对话中，如果系统指令和第一条用户消息相同，Claude API会识别出缓存的部分，无需重复发送和计费。

异步与并发处理：Elixir的并发能力可以很好地用于同时处理多个AI请求或并行调用多个工具。

# 假设我们需要向三个不同的模型询问同一个问题以获取综合意见 tasks = [ Task.async(fn -> ask_model(ChatOpenAI.new!(%{model: "gpt-4o"}), question) end), Task.async(fn -> ask_model(ChatAnthropic.new!(%{model: "claude-3-5-sonnet"}), question) end), Task.async(fn -> ask_model(ChatGrok.new!(%{model: "grok-4"}), question) end) ] results = Task.await_many(tasks, 30_000) # 30秒超时 # 然后分析或整合三个结果

Token使用监控：每次链运行后，可以从LLMChain或Trajectory中获取本次交互的Token使用量，这对于成本监控和预算控制非常有用。

{:ok, chain} = LLMChain.run(some_chain) token_usage = chain.token_usage IO.inspect(token_usage, label: "本次消耗Token") # 输出: 本次消耗Token: %LangChain.TokenUsage{input: 1250, output: 320, total: 1570}

5.3 错误处理与韧性设计

AI服务调用可能因网络、速率限制、模型过载等原因失败。在生产环境中必须有健壮的错误处理。

defmodule MyApp.RobustAI do def ask_with_retry(chain, max_retries \\ 3) do case LLMChain.run(chain) do {:ok, result} -> {:ok, result} {:error, reason} when max_retries > 0 -> # 可能是速率限制（429）、服务器错误（5xx）等 Process.sleep(1000 * (4 - max_retries)) # 指数退避简化版 ask_with_retry(chain, max_retries - 1) {:error, reason} -> # 记录详细错误，并返回用户友好的提示 Logger.error("AI服务调用最终失败: #{inspect(reason)}") {:error, :service_unavailable} end end end

对于关键业务，可以考虑实现熔断器模式（使用:fuse等库），当AI服务连续失败时，暂时切断请求，直接返回降级内容（如预定义的回复），保护后端服务并快速响应用户。

6. 常见问题排查与调试技巧

在实际集成brainlid/langchain时，你可能会遇到一些典型问题。以下是一些快速排查的思路和解决方法。

问题一：API调用返回认证错误（401/403）

检查点：
1. 环境变量：确保OPENAI_API_KEY等环境变量已正确设置且在运行环境中可访问。在IEx中尝试System.fetch_env!("OPENAI_API_KEY")看是否报错。
2. 配置加载：确认你的配置（config/runtime.exs或config/config.exs）在应用启动时被正确加载。在Phoenix应用中，确保相关配置在正确的环境文件中。
3. 密钥格式：某些服务商的密钥可能有特定前缀（如sk-，claude-），确保完整复制，没有多余空格或换行。
4. 多配置冲突：如果你同时配置了openai_key和通过函数{Module, :fun, []}的方式，确保没有冲突。库会按特定顺序解析。

问题二：工具函数（Function）未被AI调用

检查点：
1. 函数描述：Function的description字段至关重要。AI根据描述决定是否以及何时调用它。确保描述清晰、准确地说明了函数的用途和适用场景。
2. 参数Schema：parameters_schema必须是一个有效的OpenAPI JSON Schema对象。确保type，properties，required字段定义正确。可以使用在线JSON Schema验证器检查。
3. 模型能力：确认你使用的模型支持“函数调用”（Function Calling）或“工具使用”（Tool Use）功能。gpt-3.5-turbo早期版本可能支持不佳，gpt-4，claude-3系列支持良好。Ollama的某些小模型可能不支持。
4. 提示词引导：有时需要在系统消息或用户消息中明确引导AI使用工具。例如：“请使用你拥有的工具来获取必要信息后再回答。”

问题三：流式响应（Streaming）不工作或数据格式异常

检查点：
1. 模型支持：并非所有ChatModel实现都支持流式。查阅对应模块的文档（如ChatOpenAI支持stream: true）。
2. 端点兼容性：如果你使用的是第三方兼容API（如本地部署的text-generation-webui），确保其聊天端点支持Server-Sent Events（SSE）流式响应。
3. 数据处理：流式事件的处理逻辑取决于库的实现。确保你按照stream_fn返回的事件结构（:start，{:content, delta}，:done）正确解析。打开verbose: true查看原始事件有助于调试。

问题四：Bumblebee本地模型推理速度慢或内存溢出

检查点：
1. 模型尺寸：首先尝试较小的模型（如Llama-3.2-3B-Instruct）。70B参数的模型需要极大的GPU内存。
2. 编译器：在Nx.Serving配置中设置defn_options: [compiler: EXLA]可以显著加速（如果安装了exla库且CUDA可用）。CPU推理则使用:default编译器。
3. 量化：使用Bumblebee提供的量化功能加载4-bit或8-bit的模型，可以大幅减少内存占用，对速度影响较小。查看Bumblebee文档的量化示例。
4. 批处理：如果同时处理多个请求，确保Nx.Serving配置了合适的批处理策略。

问题五：轨迹（Trajectory）断言测试失败

检查点：
1. 匹配模式：assert_trajectory默认是严格匹配（顺序和参数完全一致）。如果你的Agent行为有非确定性（例如，并行工具调用顺序不定），使用mode: :unordered。
2. 参数匹配：工具调用的参数可能包含动态生成的内容（如时间戳、随机ID）。使用arguments: nil来只匹配工具名，或者使用:subset模式进行部分匹配。
3. Token计数差异：即使是相同的提示词和模型，不同次运行的Token计数也可能有细微差异（取决于API端的计算方式）。如果测试依赖于精确的token_usage，可能需要放宽断言条件（如使用assert_in_delta）。

调试心法：开启verbose: true是调试链行为的最有效手段。它会打印出LLM的原始请求、响应、工具调用和中间结果，让你清晰地看到AI的“思考”过程。对于复杂的问题，将verbose日志与Trajectory记录的工具调用序列结合起来分析，能快速定位是提示词设计问题、工具定义问题，还是模型本身的理解问题。