使用 Rails + LangChain 构建 AI Agent 系统

• ActiveJob + Sidekiq：Agent 的多步骤 ReAct 循环天然适合异步 Job，避免 HTTP 超时，支持重试与优先级队列。
• 
  
• ActionCable 实时推送：Agent 执行中间步骤可通过 WebSocket 实时流回前端，用户体验接近原生 streaming。
• 
  
• ActiveRecord 作为记忆层：对话历史、工具调用记录、Agent 状态机全部持久化到 PostgreSQL，天然可审计。
• 
  
• LangChain 工具生态：通过 langchainrb gem 或 Python sidecar 直接复用数百个 LangChain 工具集成。
• 
  

1. 
   
2. 1. Browser / API Client 发起请求
3. 2. Rails Controller 接收，通过 ActionCable Channel 建立 WebSocket 连接
4. 3. 创建 AgentJob 交由 Sidekiq 异步执行
5. 4. AgentExecutor 主循环运行，读写 Memory Store（PostgreSQL）
6. 5. 调用 LangChain / Claude API 进行推理
7. 6. 根据 LLM 决策执行 Tool（Web Search、Code Runner、DB Query 等）
8. 
   

# Gemfile

# AI & LangChain
gem 'langchainrb', '~> 0.19'
gem 'anthropic'          # Claude SDK
gem 'ruby-openai'        # OpenAI 备选

# 向量存储
gem 'pgvector'           # PostgreSQL 向量扩展
gem 'neighbor'           # Rails 向量相似搜索

# 后台任务
gem 'sidekiq', '~> 7.0'

# 实时通信
gem 'redis'
gem 'actioncable'

# db/migrate/create_agent_sessions.rb

class CreateAgentSessions < ActiveRecord::Migration[7.1]
  def change
    create_table :agent_sessions do |t|
      t.references :user, null: false, foreign_key: true
      t.string     :status, default: 'pending'
      t.string     :goal, null: false
      t.jsonb      :context, default: {}
      t.timestamps
    end

    create_table :agent_steps do |t|
      t.references :agent_session, null: false, foreign_key: true
      t.integer    :step_index
      t.string     :thought      # LLM 推理过程
      t.string     :action       # 调用的工具名
      t.jsonb      :action_input
      t.text       :observation  # 工具返回结果
      t.timestamps
    end

    add_column :agent_sessions, :messages, :jsonb, default: []
    enable_extension 'vector'
    add_column :agent_sessions, :embedding, :vector, limit: 1536
  end
end

# app/services/agent_executor.rb

class AgentExecutor
  MAX_ITERATIONS = 10

  def initialize(session:, tools:, llm: nil)
    @session    = session
    @tools      = tools.index_by(&:name)
    @llm        = llm || default_llm
    @memory     = ConversationMemory.new(session)
  end

  def run(goal)
    @session.update!(status: 'running', goal: goal)
    messages = @memory.load

    MAX_ITERATIONS.times do |i|
      response = @llm.chat(
        messages: messages + [{ role: 'user', content: goal }],
        tools: tool_definitions
      )

      case response.stop_reason
      when 'end_turn'
        return finalize(response.content)

      when 'tool_use'
        tool_use = response.content.find { |b| b.type == 'tool_use' }
        result   = execute_tool(tool_use, step_index: i)

        messages << { role: 'assistant', content: response.content }
        messages << { role: 'user', content: [
          { type: 'tool_result', tool_use_id: tool_use.id, content: result }
        ]}

        broadcast_step(step_index: i, tool: tool_use.name, result: result)
      end
    end

    finalize("Max iterations reached")
  end

  private

  def execute_tool(tool_use, step_index:)
    tool   = @tools[tool_use.name]
    result = tool.call(tool_use.input)

    @session.agent_steps.create!(
      step_index:   step_index,
      action:       tool_use.name,
      action_input: tool_use.input,
      observation:  result
    )
    result
  end

  def broadcast_step(**payload)
    AgentChannel.broadcast_to(@session, payload)
  end

  def finalize(answer)
    @session.update!(status: 'completed')
    @memory.save
    answer
  end

  def default_llm
    Langchain::LLM::Anthropic.new(
      api_key: ENV['ANTHROPIC_API_KEY'],
      default_options: { model: 'claude-opus-4-5' }
    )
  end
end

# app/jobs/agent_job.rb

class AgentJob < ApplicationJob
  queue_as :agents
  sidekiq_options retry: 2, dead: false

  def perform(session_id)
    session  = AgentSession.find(session_id)
    tools    = [WebSearchTool.new, CodeRunnerTool.new, DatabaseTool.new]
    executor = AgentExecutor.new(session: session, tools: tools)
    executor.run(session.goal)
  rescue => e
    session&.update!(status: 'failed', context: { error: e.message })
    raise
  end
end

# app/tools/base_tool.rb

module Tools
  class BaseTool
    attr_reader :name, :description

    def call(input)
      raise NotImplementedError
    end

    def to_claude_definition
      {
        name:         name,
        description:  description,
        input_schema: input_schema
      }
    end
  end
end# app/tools/web_search_tool.rb

class WebSearchTool < Tools::BaseTool
  def name        = 'web_search'
  def description = '使用 Tavily 搜索实时网页信息。输入搜索关键词，返回相关摘要。'

  def input_schema
    {
      type: 'object',
      properties: {
        query: { type: 'string', description: '搜索查询词' }
      },
      required: ['query']
    }
  end

  def call(input)
    client  = Tavily::Client.new(api_key: ENV['TAVILY_API_KEY'])
    results = client.search(query: input['query'], max_results: 5)
    results.map { |r| "**#{r[:title]}**: #{r[:content]}" }.join("\n\n")
  rescue => e
    "Search failed: #{e.message}"
  end
end

1. 短期记忆（in-flight messages）：当前 ReAct 循环内的对话历史，存在内存中，随 Job 结束消失。
2. 会话记忆（AgentSession.messages）：序列化到 PostgreSQL JSONB 字段，跨 Job 恢复，支持断点续跑。
3. 语义记忆（pgvector）：将历史对话 embedding 存入向量列，允许 Agent 检索"我之前做过类似的任务"。
4. 
   

# app/services/conversation_memory.rb

class ConversationMemory
  def initialize(session)
    @session = session
  end

  def load
    messages = @session.messages || []
    return messages if within_token_limit?(messages)
    summarize_and_trim(messages)
  end

  def save
    @session.update!(messages: @session.reload.messages)
    embed_and_store  # 异步生成 embedding
  end

  private

  def within_token_limit?(messages)
    messages.sum { |m| m[:content].to_s.length } < 80_000
  end

  def summarize_and_trim(messages)
    recent   = messages.last(10)
    old_text = messages[0..-11].map { |m| m[:content] }.join("\n")
    summary  = LLM.summarize(old_text)
    [{ role: 'system', content: "历史摘要: #{summary}" }] + recent
  end
end

# app/channels/agent_channel.rb

class AgentChannel < ApplicationCable::Channel
  def subscribed
    session = AgentSession.find_by(id: params[:session_id])
    if session&.user == current_user
      stream_for session
    else
      reject
    end
  end
end

# 在 AgentExecutor 中广播：
# AgentChannel.broadcast_to(session, {
#   step: 3, tool: 'web_search', status: 'running', result: '...'
# })

# config/sidekiq.yml

:concurrency: 10
:queues:
  - [agents, 3]    # Agent 任务，优先级高
  - [default, 2]
  - [mailers, 1]

:limits:
  agents: 5        # 最多 5 个并发 Agent，防止 API 限速

• Rate Limiting：在 Sidekiq 层限制 Agent Job 并发数，配合 Sidekiq::Limiter 防止 Claude/OpenAI API 429 错误。
• 
  
• 幂等性：AgentJob 必须幂等，通过 session.status 检查防止重复执行，Sidekiq 重试时安全。
• 
  
• 可观测性：接入 Datadog / Sentry，对每个 Agent 步骤打 span，可追踪 LLM 延迟和工具耗时。
• 
  
• 成本控制：在 AgentSession 上记录 tokens_used，超出预算时主动终止 Job，防止无限循环。