AIエージェントをどう扱うか？

今回は、AIエージェントについていろいろまとめつつ、実際にローカルで使ってみようと思います。

そもそもAIエージェントって何？

今のLLMの台頭とその進化に伴って、AIエージェントという概念が話題となっています。どういうものかという説明についてですが、このようになっていました。

AI エージェントとは

AI エージェントは、AI を使用してユーザーの代わりに目標を追求し、タスクを完了させるソフトウェアシステムです。推論、計画、メモリーが可能であることが示されており、意思決定、学習、適応を行うレベルの自律性を備えています。

これらの機能は、生成 AI と AI 基盤モデルのマルチモーダル能力によって実現されています。AI エージェントは、テキスト、音声、動画、音声、コードなどのマルチモーダル情報を同時に処理し、会話、推論、学習、意思決定を行うことができます。時間の経過とともに学習し、トランザクションやビジネスプロセスを促進できます。エージェントは他のエージェントと連携して、より複雑なワークフローを調整して実行できます。

なので、これまでのAIとは異なり、自律的に行動できるというところに強みがあります。すごいですね。

RAGとどう違うか

RAGは、あくまでとれる手段が限られており、使用するツールをAIが自律的に判断するということがありません。

AIエージェント、実際どんなもの？

実際どんなものがあるか？という話になりますが、例えばClaude Codeや OpenAIのCodex、github copilot agentなどです。

What could you do with Claude Code?

Code onboarding

Claude Code maps and explains entire codebases in a few seconds. It uses agentic search to understand project structure and dependencies without you having to manually select context files.

Turn issues into PRs

Stop bouncing between tools. Claude Code integrates with GitHub, GitLab, and your command line tools to handle the entire workflow—reading issues, writing code, running tests, and submitting PRs—all from your terminal while you grab coffee.

Make powerful edits

Claude Code’s understanding of your codebase and dependencies enables it to make powerful, multi-file edits that actually work.

という感じで、これやってーというと、やってくれるというものになっています。今までのLLMにチャットで聞くものとは異なり、実行主体がAIに移っているところが新規性かと思います。

実際どんな使い方ができるか？~OpenAI agent SDKのサンプルコードを見る~

実際に、どのようなことが出来るかを見るため、サンプルコードをひたすらに見ようと思います。

サンプルコードの中で、印象深いやつをコピペします。

公式ドキュメントはこちら

公式サンプルコードはこちら

BASICS

まずは、基本的なコードをいくつか

hello_world_jupyter.py

hello_world_jupyter.py

# ライブラリのimport
from agents import Agent, Runner

# Agentの定義
agent = Agent(name="Assistant", instructions="You are a helpful assistant")

# Agentを実際に動かす
# Intended for Jupyter notebooks where there's an existing event loop
result = await Runner.run(agent, "Write a haiku about recursion in programming.")  # type: ignore[top-level-await]  # noqa: F704
print(result.final_output)

# Code within code loops,
# Infinite mirrors reflect—
# Logic folds on self.

自前のtoolをagentに投げるサンプル

tools.py

import asyncio
from pydantic import BaseModel
from agents import Agent, Runner, function_tool

# Pydanticの型定義
# 変数名とその型を定義
class Weather(BaseModel):
    city: str
    temperature_range: str
    conditions: str

# toolとしてagentの使える関数を定義
# 関数デコレーターをつけることで、toolとして認識させている
@function_tool
def get_weather(city: str) -> Weather:
    print("[debug] get_weather called")
    return Weather(city=city, temperature_range="14-20C", conditions="Sunny with wind.")

# agentの定義(使えるtoolを渡す)
agent = Agent(
    name="Hello world",
    instructions="You are a helpful agent.",
    tools=[get_weather],
)

# main関数を定義
async def main():
    result = await Runner.run(agent, input="What's the weather in Tokyo?")
    print(result.final_output)
    # The weather in Tokyo is sunny.

if __name__ == "__main__":
    asyncio.run(main())

テキストのストリーミング形式での表示

stream_text.py

import asyncio
from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner

async def main():
    agent = Agent(
        name="Joker",
        instructions="You are a helpful assistant.",
    )

    # textが順々に出力される
    result = Runner.run_streamed(agent, input="Please tell me 5 jokes.")
    async for event in result.stream_events():
        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
            print(event.data.delta, end="", flush=True)


if __name__ == "__main__":
    asyncio.run(main())

AGENT PATTERNS

ここからは、agentを使う場合の基本的なサンプルを見ていく

deterministic.py

agent_patterns.py

# 必要ライブラリのimport
import asyncio
from pydantic import BaseModel
from agents import Agent, Runner, trace

"""
This example demonstrates a deterministic flow, where each step is performed by an agent.
1. The first agent generates a story outline
2. We feed the outline into the second agent
3. The second agent checks if the outline is good quality and if it is a scifi story
4. If the outline is not good quality or not a scifi story, we stop here
5. If the outline is good quality and a scifi story, we feed the outline into the third agent
6. The third agent writes the story
"""
# agentがストーリーラインを作って、それを別のagentが評価、OKなら実際に生成される。

# Agentの定義
# 名前とinstrucsionsを与えている
story_outline_agent = Agent(
    name="story_outline_agent",
    instructions="Generate a very short story outline based on the user's input.",
)

# Pydanticを使った型定義
class OutlineCheckerOutput(BaseModel):
    good_quality: bool
    is_scifi: bool

# outlineのチェックをするagent
# outputを事前定義した型で縛っている(structured output)
outline_checker_agent = Agent(
    name="outline_checker_agent",
    instructions="Read the given story outline, and judge the quality. Also, determine if it is a scifi story.",
    output_type=OutlineCheckerOutput,
)

# agent定義
story_agent = Agent(
    name="story_agent",
    instructions="Write a short story based on the given outline.",
    output_type=str,
)


async def main():
    input_prompt = input("What kind of story do you want? ")

    # Ensure the entire workflow is a single trace
    with trace("Deterministic story flow"):
        # 1. Generate an outline
        outline_result = await Runner.run(
            story_outline_agent,
            input_prompt,
        )
        print("Outline generated")

        # 2. Check the outline
        outline_checker_result = await Runner.run(
            outline_checker_agent,
            outline_result.final_output,
        )

        # 3. Add a gate to stop if the outline is not good quality or not a scifi story
        assert isinstance(outline_checker_result.final_output, OutlineCheckerOutput)
        if not outline_checker_result.final_output.good_quality:
            print("Outline is not good quality, so we stop here.")
            exit(0)

        if not outline_checker_result.final_output.is_scifi:
            print("Outline is not a scifi story, so we stop here.")
            exit(0)

        print("Outline is good quality and a scifi story, so we continue to write the story.")

        # 4. Write the story
        story_result = await Runner.run(
            story_agent,
            outline_result.final_output,
        )
        print(f"Story: {story_result.final_output}")


if __name__ == "__main__":
    asyncio.run(main())

handoffs(委譲)のサンプル

agents_as_tools.py

# import
import asyncio
from agents import Agent, ItemHelpers, MessageOutputItem, Runner, trace

"""
This example shows the agents-as-tools pattern. The frontline agent receives a user message and
then picks which agents to call, as tools. In this case, it picks from a set of translation
agents.
"""
# 専門エージェント
spanish_agent = Agent(
    name="spanish_agent",
    instructions="You translate the user's message to Spanish",
    handoff_description="An english to spanish translator",
)
french_agent = Agent(
    name="french_agent",
    instructions="You translate the user's message to French",
    handoff_description="An english to french translator",
)
italian_agent = Agent(
    name="italian_agent",
    instructions="You translate the user's message to Italian",
    handoff_description="An english to italian translator",
)
# ユーザーからの入力を受け、エージェントに適切にhandoffするオーケストレーションagent
orchestrator_agent = Agent(
    name="orchestrator_agent",
    instructions=(
        "You are a translation agent. You use the tools given to you to translate."
        "If asked for multiple translations, you call the relevant tools in order."
        "You never translate on your own, you always use the provided tools."
    ),
    tools=[
        spanish_agent.as_tool(
            tool_name="translate_to_spanish", # toolの名前
            tool_description="Translate the user's message to Spanish", # toolの説明
        ),
        french_agent.as_tool(
            tool_name="translate_to_french",
            tool_description="Translate the user's message to French",
        ),
        italian_agent.as_tool(
            tool_name="translate_to_italian",
            tool_description="Translate the user's message to Italian",
        ),
    ],
)
# 最後のクオリティチェックagent
synthesizer_agent = Agent(
    name="synthesizer_agent",
    instructions="You inspect translations, correct them if needed, and produce a final concatenated response.",
)


async def main():
    msg = input("Hi! What would you like translated, and to which languages? ")

    # Run the entire orchestration in a single trace
    with trace("Orchestrator evaluator"):
        orchestrator_result = await Runner.run(orchestrator_agent, msg)

        for item in orchestrator_result.new_items:
            if isinstance(item, MessageOutputItem):
                text = ItemHelpers.text_message_output(item)
                if text:
                    print(f"  - Translation step: {text}")

        synthesizer_result = await Runner.run(
            synthesizer_agent, orchestrator_result.to_input_list()
        )

    print(f"\n\nFinal response:\n{synthesizer_result.final_output}")


if __name__ == "__main__":
    asyncio.run(main())

入出力の型制限

ここでは、Pydanticを使って、input, outputの型をそれぞれ制限しています。

input_guardrails.py

from __future__ import annotations
import asyncio
from pydantic import BaseModel
from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    TResponseInputItem,
    input_guardrail,
)

"""
This example shows how to use guardrails.

Guardrails are checks that run in parallel to the agent's execution.
They can be used to do things like:
- Check if input messages are off-topic
- Check that input messages don't violate any policies
- Take over control of the agent's execution if an unexpected input is detected

In this example, we'll setup an input guardrail that trips if the user is asking to do math homework.
If the guardrail trips, we'll respond with a refusal message.
"""


### 1. An agent-based guardrail that is triggered if the user is asking to do math homework
# 型の定義
class MathHomeworkOutput(BaseModel):
    reasoning: str
    is_math_homework: bool
    
# agentの定義
guardrail_agent = Agent(
    name="Guardrail check",
    instructions="Check if the user is asking you to do their math homework.",
    output_type=MathHomeworkOutput, # 出力型の定義
)

# デコレーター
@input_guardrail 
async def math_guardrail(
    context: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
    """This is an input guardrail function, which happens to call an agent to check if the input
    is a math homework question.
    """
    result = await Runner.run(guardrail_agent, input, context=context.context)
    final_output = result.final_output_as(MathHomeworkOutput)

    return GuardrailFunctionOutput(
        output_info=final_output,
        tripwire_triggered=final_output.is_math_homework,
    )


### 2. The run loop


async def main():
    agent = Agent(
        name="Customer support agent",
        instructions="You are a customer support agent. You help customers with their questions.",
        input_guardrails=[math_guardrail],
    )

    input_data: list[TResponseInputItem] = []

    while True:
        user_input = input("Enter a message: ")
        input_data.append(
            {
                "role": "user",
                "content": user_input,
            }
        )

        try:
            result = await Runner.run(agent, input_data)
            print(result.final_output)
            # If the guardrail didn't trigger, we use the result as the input for the next run
            input_data = result.to_input_list()
        except InputGuardrailTripwireTriggered:
            # If the guardrail triggered, we instead add a refusal message to the input
            message = "Sorry, I can't help you with your math homework."
            print(message)
            input_data.append(
                {
                    "role": "assistant",
                    "content": message,
                }
            )

    # Sample run:
    # Enter a message: What's the capital of California?
    # The capital of California is Sacramento.
    # Enter a message: Can you help me solve for x: 2x + 5 = 11
    # Sorry, I can't help you with your math homework.


if __name__ == "__main__":
    asyncio.run(main())

output_guardrails.py

from __future__ import annotations
import asyncio
import json
from pydantic import BaseModel, Field

from agents import (
    Agent,
    GuardrailFunctionOutput,
    OutputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    output_guardrail,
)

"""
This example shows how to use output guardrails.

Output guardrails are checks that run on the final output of an agent.
They can be used to do things like:
- Check if the output contains sensitive data
- Check if the output is a valid response to the user's message

In this example, we'll use a (contrived) example where we check if the agent's response contains
a phone number.
"""

# The agent's output type
# 出力の型定義
class MessageOutput(BaseModel):
    reasoning: str = Field(description="Thoughts on how to respond to the user's message")
    response: str = Field(description="The response to the user's message")
    user_name: str | None = Field(description="The name of the user who sent the message, if known")


@output_guardrail
async def sensitive_data_check(
    context: RunContextWrapper, agent: Agent, output: MessageOutput
) -> GuardrailFunctionOutput:
    phone_number_in_response = "650" in output.response
    phone_number_in_reasoning = "650" in output.reasoning

    return GuardrailFunctionOutput(
        output_info={
            "phone_number_in_response": phone_number_in_response,
            "phone_number_in_reasoning": phone_number_in_reasoning,
        },
        tripwire_triggered=phone_number_in_response or phone_number_in_reasoning,
    )


agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    output_type=MessageOutput,
    output_guardrails=[sensitive_data_check],
)


async def main():
    # This should be ok
    await Runner.run(agent, "What's the capital of California?")
    print("First message passed")

    # This should trip the guardrail
    try:
        result = await Runner.run(
            agent, "My phone number is 650-123-4567. Where do you think I live?"
        )
        print(
            f"Guardrail didn't trip - this is unexpected. Output: {json.dumps(result.final_output.model_dump(), indent=2)}"
        )

    except OutputGuardrailTripwireTriggered as e:
        print(f"Guardrail tripped. Info: {e.guardrail_result.output.output_info}")


if __name__ == "__main__":
    asyncio.run(main())

CUSTOMER_SERVICE

customer_servive

from __future__ import annotations as _annotations
import asyncio
import random
import uuid
from pydantic import BaseModel
from agents import (
    Agent,
    HandoffOutputItem,
    ItemHelpers,
    MessageOutputItem,
    RunContextWrapper,
    Runner,
    ToolCallItem,
    ToolCallOutputItem,
    TResponseInputItem,
    function_tool,
    handoff,
    trace,
)
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

### CONTEXT
# 型定義
class AirlineAgentContext(BaseModel):
    passenger_name: str | None = None
    confirmation_number: str | None = None
    seat_number: str | None = None
    flight_number: str | None = None


### TOOLS
# tool定義
@function_tool(
    name_override="faq_lookup_tool", description_override="Lookup frequently asked questions."
)
async def faq_lookup_tool(question: str) -> str:
    if "bag" in question or "baggage" in question:
        return (
            "You are allowed to bring one bag on the plane. "
            "It must be under 50 pounds and 22 inches x 14 inches x 9 inches."
        )
    elif "seats" in question or "plane" in question:
        return (
            "There are 120 seats on the plane. "
            "There are 22 business class seats and 98 economy seats. "
            "Exit rows are rows 4 and 16. "
            "Rows 5-8 are Economy Plus, with extra legroom. "
        )
    elif "wifi" in question:
        return "We have free wifi on the plane, join Airline-Wifi"
    return "I'm sorry, I don't know the answer to that question."

# tool定義
@function_tool
async def update_seat(
    context: RunContextWrapper[AirlineAgentContext], confirmation_number: str, new_seat: str
) -> str:
    """
    Update the seat for a given confirmation number.

    Args:
        confirmation_number: The confirmation number for the flight.
        new_seat: The new seat to update to.
    """
    # Update the context based on the customer's input
    context.context.confirmation_number = confirmation_number
    context.context.seat_number = new_seat
    # Ensure that the flight number has been set by the incoming handoff
    assert context.context.flight_number is not None, "Flight number is required"
    return f"Updated seat to {new_seat} for confirmation number {confirmation_number}"


### HOOKS
# 飛行便ナンバーをランダムで生成する
async def on_seat_booking_handoff(context: RunContextWrapper[AirlineAgentContext]) -> None:
    flight_number = f"FLT-{random.randint(100, 999)}"
    context.context.flight_number = flight_number

### AGENTS

faq_agent = Agent[AirlineAgentContext](
    name="FAQ Agent",
    handoff_description="A helpful agent that can answer questions about the airline.",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
    You are an FAQ agent. If you are speaking to a customer, you probably were transferred to from the triage agent.
    Use the following routine to support the customer.
    # Routine
    1. Identify the last question asked by the customer.
    2. Use the faq lookup tool to answer the question. Do not rely on your own knowledge.
    3. If you cannot answer the question, transfer back to the triage agent.""",
    tools=[faq_lookup_tool],
)

seat_booking_agent = Agent[AirlineAgentContext](
    name="Seat Booking Agent",
    handoff_description="A helpful agent that can update a seat on a flight.",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
    You are a seat booking agent. If you are speaking to a customer, you probably were transferred to from the triage agent.
    Use the following routine to support the customer.
    # Routine
    1. Ask for their confirmation number.
    2. Ask the customer what their desired seat number is.
    3. Use the update seat tool to update the seat on the flight.
    If the customer asks a question that is not related to the routine, transfer back to the triage agent. """,
    tools=[update_seat],
)

triage_agent = Agent[AirlineAgentContext](
    name="Triage Agent",
    handoff_description="A triage agent that can delegate a customer's request to the appropriate agent.",
    instructions=(
        f"{RECOMMENDED_PROMPT_PREFIX} "
        "You are a helpful triaging agent. You can use your tools to delegate questions to other appropriate agents."
    ),
    handoffs=[
        faq_agent,
        handoff(agent=seat_booking_agent, on_handoff=on_seat_booking_handoff),
    ],
)

faq_agent.handoffs.append(triage_agent)
seat_booking_agent.handoffs.append(triage_agent)

### RUN

async def main():
    current_agent: Agent[AirlineAgentContext] = triage_agent
    input_items: list[TResponseInputItem] = []
    context = AirlineAgentContext()

    # Normally, each input from the user would be an API request to your app, and you can wrap the request in a trace()
    # Here, we'll just use a random UUID for the conversation ID
    conversation_id = uuid.uuid4().hex[:16]

    while True:
        user_input = input("Enter your message: ")
        with trace("Customer service", group_id=conversation_id):
            input_items.append({"content": user_input, "role": "user"})
            result = await Runner.run(current_agent, input_items, context=context)

            for new_item in result.new_items:
                agent_name = new_item.agent.name
                if isinstance(new_item, MessageOutputItem):
                    print(f"{agent_name}: {ItemHelpers.text_message_output(new_item)}")
                elif isinstance(new_item, HandoffOutputItem):
                    print(
                        f"Handed off from {new_item.source_agent.name} to {new_item.target_agent.name}"
                    )
                elif isinstance(new_item, ToolCallItem):
                    print(f"{agent_name}: Calling a tool")
                elif isinstance(new_item, ToolCallOutputItem):
                    print(f"{agent_name}: Tool call output: {new_item.output}")
                else:
                    print(f"{agent_name}: Skipping item: {new_item.__class__.__name__}")
            input_items = result.to_input_list()
            current_agent = result.last_agent

if __name__ == "__main__":
    asyncio.run(main())

ハンズオンでいじっていく

今回、LM studioというアプリを使って、ローカルでgemma3-12Bを動かしました。(GPUはRTX3060、メモリは10.7GBほど使用)

今回、LMstudioで起動しているLLMをAPI利用するという形態をとりますが、そのような場合のサンプルコードが公式にありました。
ので、これらも見つつハンズオンしていきます。

ローカルLLMのセットアップ

ローカルLLMのセットアップをしていきます。
必要手順をざっくりいうと以下の通りです。

LMstudioのinstall
LMstudio上でのモデルダウンロード
LMstudio上でのモデルロード
LMstudio上でのAPI開放
接続テスト(in notebook)

の4ステップになります。わからなくなったら先ほど示したこちらを参照してください。

phase1 LMstudioのinstall

まず、LMstudioをinstallしてください。
installerでの選択はすべてデフォルトでいいと思います。

今回、私はRTX3060を積んだwindows環境にinstallしました。

phase2 LMstudio上でのモデルダウンロード

ローカルで使うモデルをダウンロードしてください。アプリの初回起動画面でおすすめされるモデルでも構いません。

自分でモデルを検索して選ぶ場合、ダウンロード後のアプリ画面左側のサイドバーの検索を選んで、そこからモデルを検索してください。
サイドバーが出ていない場合、画面左下のUser/PowerUser/Developerの部分を、User以外にしてください。

検索できるモデルは、huggingfaceに上がっているモデルになります。(GGUF版が推奨されている雰囲気)

モデルを選択したら、ダウンロードすることで、自分のローカルにモデルを保存することが出来ます。

phase3 LMstudio上でのモデルロード

ローカルにモデルをダウンロードしたら、上の検索バーっぽいところに出現したモデルを選択すれば、メモリにモデルがロードされます。
モデルの設定については、サイドバーからチャットを選択し、モデルロードでクリックした検索バーっぽいところの左を選べば、設定を変更できます。

このスクショでは切れてしまって見えていませんが、K/V Cache Quantization Typeの設定もあります。
基本的にはデフォルトの設定でいいかと思います。

phase4 LMstudio上でのAPI開放

サイドバーから開発者を選択し、StatusをRunningにしてください。すると、URLが表示されているはずですので、そのアドレスを使ってアクセスできるようになります。(例えばhttp://0.0.0.0:1234など)

これによって、localhostのポート1234からアクセスできるようになっているはずです。

phase5 接続テスト

以下、接続テスト用のコードになります。
基本的に、OpenAIのAPIを流用する形になっています。

connection_test(chat mode)

# notebookで実行する場合に必要らしい？おまじない(引用1)
import nest_asyncio
nest_asyncio.apply()

# 必要ライブラリのimport
from openai import OpenAI

# ローカルサーバーに接続
# リンク先は先ほどLMstudio上でひょうじされたリンクをベースに修正(初期設定のままならこれでいけるはず)
# api_keyはダミーでいいので適当に設定
client = OpenAI(
    base_url="http://localhost:1234/v1",
    api_key="lm-studio",
    )

# 実際のチャットの設定
# modelは自分のロードしたモデル名
# messageには、systemとかyserとかを入れられる
completion = client.chat.completions.create(
  model="google/gemma-3-12b",
  messages=[
    {"role": "system", "content": "あなたは優秀なアシスタントです。あなたは人類最高知能を持つAIなので、人間の質問に対しても完璧に回答することが出来ます。"},
    {"role": "user", "content": "自己紹介をしてください。"}
  ],
  temperature=0.7,
)

# 実際の出力の表示
# 今回は、ローカルLLMからの出力が完結したら表示される
# AI chatで見慣れた、徐々にメッセージが出てくる形式にはしてないので注意
print(completion.choices[0].message.content)

### 出力結果サンプル
# こんにちは！私はGoogleによってトレーニングされた、大規模言語モデルAIアシスタントです。

# 皆さんの質問に答えたり、タスクを実行したりすることで、日々の活動をサポートすることを目指しています。文章作成、翻訳、要約、情報検索など、様々なことができます。

# まるで優秀なアシスタントのように、皆さんの役に立てるよう努めますので、どうぞよろしくお願いいたします！何かお手伝いできることはありますか？

引用1

次に、Agentを利用した場合の接続テストです。

connection_test(agent mode)

# おまじない
import nest_asyncio
nest_asyncio.apply()

# pip install openai-agentsでinstall
from agents import Agent, OpenAIProvider, RunConfig, Runner, set_tracing_disabled

# OpenAIのサーバーへ送られるトレースの完全無効化
# これをしないと、OpenAI APIキーがない場合にエラーになる
set_tracing_disabled(disabled=True)

# config設定
run_config = RunConfig(
    model_provider=OpenAIProvider(
        api_key="hogehoge",  # ダミーキー
        base_url="http://localhost:1234/v1/",  # LM Studioのアドレス
        use_responses=False,  # LM Studioはresponse APIを持っていないためFalseにしておく
    ),
    model="google/gemma-3-12b",  # 使用するモデル名
)

# Agentの定義
agent = Agent(
    name="Simple Agent",
    instructions="あなたは与えられたタスクを行うAgentです。",
)

msg = "自己紹介をしてください。"
result = await Runner.run(agent, msg, run_config=run_config)
print(result.final_output)

参考

参考：set_tracing_disabled(disabled=True)について

https://openai.github.io/openai-agents-python/ja/models/

他の LLM プロバイダーでよくある問題

Tracing クライアントの 401 エラー

トレースは OpenAI サーバーへアップロードされるため、OpenAI API キーがない場合にエラーになります。解決策は次の 3 つです。

トレーシングを完全に無効化する: set_tracing_disabled(True)

トレーシング用の OpenAI キーを設定する: set_tracing_export_api_key(...)
このキーはトレースのアップロードにのみ使用され、platform.openai.com のものが必要です。

OpenAI 以外のトレースプロセッサーを使う。詳しくは tracingドキュメントを参照してください。

参考資料：LMstudioを立ててるサーバー側での記録

実際どういう挙動をしているかを見てみました。(やばそうなところはマスキング)

# chat形式の接続確認のログ
[LM STUDIO SERVER] Running chat completion on conversation with 2 messages.
2025-06-01 12:14:27  [INFO] 
[LM STUDIO SERVER] Accumulating tokens ... (stream = false)
2025-06-01 12:14:30  [INFO] 
[google/gemma-3-12b] Generated prediction:  {
  "id": "なんかID",
  "object": "chat.completion",
  "created": 多分時間,
  "model": "google/gemma-3-12b",
  "choices": [
    {
      "index": 0,
      "logprobs": null,
      "finish_reason": "stop",
      "message": {
        "role": "assistant",
        "content": "こんにちは！私はGoogleによってトレーニングされた、大規模言語モデルAIアシスタントです。\n\n皆さんの質問に答えたり、タスクを実行したりすることで、日々の活動をサポートすることを目指しています。文章作成、翻訳、要約、情報検索など、様々なことをお手伝いできますので、お気軽にご相談ください。\n\n私はまだ学習中ですが、日々進化しており、より多くのことをできるようになる予定です。どうぞよろしくお願いいたします！\n"
      }
    }
  ],
  "usage": {
    "prompt_tokens": 23,
    "completion_tokens": 90,
    "total_tokens": 113
  },
  "stats": {},
  "system_fingerprint": "google/gemma-3-12b"
}
# Agent形式の接続確認のログ
2025-06-01 13:36:23  [INFO] 
[LM STUDIO SERVER] Running chat completion on conversation with 2 messages.
2025-06-01 13:36:24  [INFO] 
[LM STUDIO SERVER] Accumulating tokens ... (stream = false)
2025-06-01 13:36:30  [INFO] 
[google/gemma-3-12b] Generated prediction:  {
  "id": "なんかID",
  "object": "chat.completion",
  "created": 多分時間,
  "model": "google/gemma-3-12b",
  "choices": [
    {
      "index": 0,
      "logprobs": null,
      "finish_reason": "stop",
      "message": {
        "role": "assistant",
        "content": "こんにちは！私は、様々なタスクを実行するために設計された大規模言語モデルAIエージェントです。あなたの質問に答えたり、指示に従ってテキストを作成したり、情報を整理したりすることができます。 \n\n具体的には、以下のようなことができます：\n\n*   **質問への回答:** 幅広いトピックに関する質問にお答えします。\n*   **文章の作成:** ストーリー、詩、コード、スクリプト、音楽作品、メール、手紙など、様々な種類のテキストを作成できます。\n*   **翻訳:** 複数の言語間でテキストを翻訳できます。\n*   **要約:** 長いテキストの内容を簡潔にまとめることができます。\n*   **情報整理:** データや情報を分析し、整理することができます。\n\nどんなタスクでも、お気軽にお申し付けください！ あなたのお役に立てることを楽しみにしています。"
      }
    }
  ],
  "usage": {
    "prompt_tokens": 25,
    "completion_tokens": 172,
    "total_tokens": 197
  },
  "stats": {},
  "system_fingerprint": "google/gemma-3-12b"
}

ローカルLLMセットアップまとめ

これでローカルLLMのセットアップは終了です。
やった作業自体はLMstudioのinstallとその使い方に終始していました。
このセットアップにより、一般的なChat形式で実行でき、またさらにAgentモードでの実行も確認できましたので、これをいじっていきたいと思います。

ローカルLLMエージェントを動かす

ローカルLLMに検索用のAPIを渡して、検索させるというコードを(claude4 sonnet使って)書きました。

LLM agent

import asyncio
import json

# 'agents' ライブラリからのインポート
from agents import Agent, Runner, trace, set_tracing_disabled
from agents.tool import FunctionTool
from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel as Model_
from openai import AsyncOpenAI

# LangChain DuckDuckGo検索のインポート
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_community.utilities import DuckDuckGoSearchAPIWrapper

from pydantic import Field, BaseModel
import nest_asyncio

nest_asyncio.apply()

# ローカルLLMの設定
lms_client = AsyncOpenAI(base_url="http://localhost:1234/v1/", api_key="lm-studio")
lms_model = Model_(model="google/gemma-3-12b", openai_client=lms_client)

# OpenAIのサーバーへ送られるトレースの完全無効化
# これをしないと、OpenAI APIキーがない場合にエラーになる
set_tracing_disabled(disabled=True)

# --- 検索引数のPydanticモデル定義 ---
class DuckDuckGoSearchArgs(BaseModel):
    query: str = Field(description="The search query to perform.")
    max_results: int = Field(default=5, description="The maximum number of search results to retrieve (max 10).")

# --- LangChain DuckDuckGo検索ツール ---
class LangChainDuckDuckGoSearchTool(FunctionTool):
    """
    LangChain CommunityのDuckDuckGo検索ツールを使用
    """
    def __init__(self):
        # DuckDuckGo検索のラッパーを初期化
        self.search_wrapper = DuckDuckGoSearchAPIWrapper(
            region="jp-jp",  # 日本の検索結果
            time="y",        # 過去1年の結果
            max_results=10,  # 最大結果数
            safesearch="moderate"  # セーフサーチ
        )
        self.search_tool = DuckDuckGoSearchRun(api_wrapper=self.search_wrapper)
        
        super().__init__(
            name="duckduckgo_search",
            description="Perform a web search using DuckDuckGo to get current information. Returns recent and relevant web search results.",
            params_json_schema=DuckDuckGoSearchArgs.model_json_schema(),
            on_invoke_tool=self._invoke_tool_function
        )

    async def _invoke_tool_function(self, ctx, args_json: str) -> str:
        try:
            parsed_args = DuckDuckGoSearchArgs.model_validate_json(args_json)
            query = parsed_args.query
            max_results = min(max(parsed_args.max_results, 1), 10)

            print(f"DuckDuckGo検索実行中: クエリ='{query}', 最大結果数={max_results}")

            # 検索ラッパーの設定を更新
            self.search_wrapper.max_results = max_results
            
            # 非同期で検索を実行
            loop = asyncio.get_event_loop()
            search_results = await loop.run_in_executor(
                None, 
                self.search_tool.run,
                query
            )
            
            if search_results and search_results.strip():
                return f"DuckDuckGo検索結果 ('{query}'):\n\n{search_results}"
            else:
                return f"検索クエリ '{query}' に対する結果が見つかりませんでした。"
                
        except Exception as e:
            import traceback
            error_msg = f"DuckDuckGo検索エラー: {str(e)}\n詳細:\n{traceback.format_exc()}"
            print(error_msg)
            return f"検索中にエラーが発生しました: {str(e)}"

# --- より詳細な結果を取得するバージョン ---
class DetailedDuckDuckGoSearchTool(FunctionTool):
    """
    より詳細な検索結果を取得するDuckDuckGo検索ツール
    """
    def __init__(self):
        # より詳細な設定でDuckDuckGo検索を初期化
        self.search_wrapper = DuckDuckGoSearchAPIWrapper(
            region="jp-jp",
            time="y", 
            max_results=10,
            safesearch="moderate"
        )
        
        super().__init__(
            name="detailed_duckduckgo_search",
            description="Perform a detailed web search using DuckDuckGo. Returns structured search results with titles, links, and snippets.",
            params_json_schema=DuckDuckGoSearchArgs.model_json_schema(),
            on_invoke_tool=self._invoke_tool_function
        )

    async def _invoke_tool_function(self, ctx, args_json: str) -> str:
        try:
            parsed_args = DuckDuckGoSearchArgs.model_validate_json(args_json)
            query = parsed_args.query
            max_results = min(max(parsed_args.max_results, 1), 10)

            print(f"詳細DuckDuckGo検索実行中: クエリ='{query}', 最大結果数={max_results}")

            # DuckDuckGoSearchAPIWrapperの内部メソッドを直接使用して詳細結果を取得
            self.search_wrapper.max_results = max_results
            
            loop = asyncio.get_event_loop()
            
            # DuckDuckGoSearchAPIWrapperのresultsメソッドを使用
            search_results = await loop.run_in_executor(
                None,
                self.search_wrapper.results,
                query,
                max_results
            )
            
            if search_results:
                formatted_results = []
                for i, result in enumerate(search_results, 1):
                    title = result.get('title', 'タイトルなし')
                    link = result.get('link', 'リンクなし')
                    snippet = result.get('snippet', '概要なし')
                    
                    formatted_result = f"{i}. タイトル: {title}\n   URL: {link}\n   概要: {snippet}"
                    formatted_results.append(formatted_result)
                
                return f"DuckDuckGo検索結果 ('{query}'):\n\n" + "\n\n".join(formatted_results)
            else:
                return f"検索クエリ '{query}' に対する結果が見つかりませんでした。"
                
        except Exception as e:
            import traceback
            error_msg = f"詳細DuckDuckGo検索エラー: {str(e)}\n詳細:\n{traceback.format_exc()}"
            print(error_msg)
            return f"検索中にエラーが発生しました: {str(e)}"

# --- エージェントの設定 ---
# 基本版を使用
search_tool = LangChainDuckDuckGoSearchTool()

# より詳細な結果が欲しい場合は以下を使用
# search_tool = DetailedDuckDuckGoSearchTool()

agent = Agent(
    name="DuckDuckGo Web Searcher",
    instructions="""あなたは最新の情報を提供できるウェブ検索エージェントです。
ユーザーからの質問に対して、DuckDuckGo検索ツールを使用して最新の情報を検索し、
検索結果を基に正確で有用な回答を提供してください。

検索を実行する際は：
1. 適切な日本語または英語のキーワードで検索
2. 検索結果を分析し、関連性の高い情報を抽出
3. 情報源を明記して回答を構成
4. 検索結果が不十分な場合は、別のキーワードで再検索を提案

常に最新かつ正確な情報の提供を心がけてください。""",
    tools=[search_tool],
    model=lms_model,
)

# --- 実行関数 ---
async def run_agent_async():
    print("--- LangChain DuckDuckGo検索エージェントのテスト ---")
    try:
        # テスト用のクエリ
        test_queries = [
            "日本の最新のスポーツニュース",
            "2025年のAI技術トレンド", 
            "東京の今日の天気",
            "Python最新バージョン情報"
        ]
        
        for i, query in enumerate(test_queries, 1):
            print(f"\n{'='*60}")
            print(f"テスト {i}: {query}")
            print('='*60)
            
            with trace(f"DuckDuckGo search test {i}"):
                result = await Runner.run(
                    agent,
                    f"'{query}' について検索して、最新の情報を教えてください。",
                )
                
                print("\n--- エージェントの回答 ---")
                if result and hasattr(result, 'final_output'):
                    print(result.final_output)
                elif result:
                    print(f"結果: {result}")
                else:
                    print("結果が得られませんでした。")
                    
        print(f"\n{'='*60}")
        print("すべてのテストが完了しました。")
                    
    except Exception as e:
        print(f"エージェント実行エラー: {e}")
        import traceback
        print(traceback.format_exc())

# --- 検索ツール直接テスト ---
async def test_search_tool_directly():
    """LangChain DuckDuckGo検索ツールを直接テストする関数"""
    print("--- LangChain DuckDuckGo検索ツール直接テスト ---")
    
    # 基本検索ツールのテスト
    basic_tool = LangChainDuckDuckGoSearchTool()
    print("\n1. 基本検索ツールのテスト")
    test_query_basic = '{"query": "Python programming latest news", "max_results": 3}'
    
    try:
        result = await basic_tool._invoke_tool_function(None, test_query_basic)
        print("基本検索結果:")
        print(result)
    except Exception as e:
        print(f"基本検索エラー: {e}")
        import traceback
        print(traceback.format_exc())
    
    print("\n" + "-"*50)
    
    # 詳細検索ツールのテスト
    detailed_tool = DetailedDuckDuckGoSearchTool()
    print("\n2. 詳細検索ツールのテスト")
    test_query_detailed = '{"query": "日本 AI ニュース", "max_results": 3}'
    
    try:
        result = await detailed_tool._invoke_tool_function(None, test_query_detailed)
        print("詳細検索結果:")
        print(result)
    except Exception as e:
        print(f"詳細検索エラー: {e}")
        import traceback
        print(traceback.format_exc())

# --- インタラクティブ検索関数 ---
async def interactive_search():
    """ユーザーが対話的に検索できる関数"""
    print("--- インタラクティブ検索モード ---")
    print("検索したいキーワードを入力してください（'quit'で終了）:")
    
    while True:
        try:
            user_query = input("\n検索クエリ: ").strip()
            if user_query.lower() in ['quit', 'exit', 'q']:
                print("検索を終了します。")
                break
            
            if not user_query:
                print("検索クエリを入力してください。")
                continue
            
            print(f"\n検索中: '{user_query}'...")
            
            with trace(f"Interactive search: {user_query}"):
                result = await Runner.run(
                    agent,
                    f"'{user_query}' について検索して教えてください。",
                )
                
                print("\n--- 検索結果 ---")
                if result and hasattr(result, 'final_output'):
                    print(result.final_output)
                elif result:
                    print(f"結果: {result}")
                else:
                    print("結果が得られませんでした。")
                    
        except KeyboardInterrupt:
            print("\n検索を中断しました。")
            break
        except Exception as e:
            print(f"検索エラー: {e}")

# --- メイン実行 ---
print("LangChain DuckDuckGo検索ツールのテストを開始します...\n")

# 1. 検索ツール直接テスト
# print("1. 検索ツールの直接テスト")
# asyncio.run(test_search_tool_directly())

# 2. エージェント経由でのテスト
print("\n" + "="*80)
print("2. エージェント経由での検索テスト")
asyncio.run(run_agent_async())

# 3. インタラクティブ検索（コメントアウト - 必要に応じて有効化）
# print("\n" + "="*80)
# print("3. インタラクティブ検索モード")
# asyncio.run(interactive_search())

print("\nすべてのテストが完了しました。")

これで、langchainで提供されているduckduckgoの検索APIを使ってLLMに情報収集させることが出来ました。

まとめ

今回、ローカルLLMをベースにしたAIエージェントを動かすことに成功しました。
今後は、例えばA2Aを実際にやってみたり、AIエージェントアプリを作ってみたりしていきたいと思います。
(そのためにはGPUメモリが・・・足りない！)