An integration package connecting xAI and LangChain
npx @tessl/cli install tessl/pypi-langchain-xai@0.2.00
# LangChain xAI
1
2
An integration package connecting xAI and LangChain, providing access to xAI's chat completion models through LangChain's standardized interface. This package enables developers to integrate xAI's Grok models into LangChain-based applications for chat completions, tool calling, structured output, and conversational AI workflows.
3
4
## Package Information
5
6
- **Package Name**: langchain-xai
7
- **Language**: Python
8
- **Installation**: `pip install langchain-xai`
9
- **Requirements**: Python ≥3.9
10
- **Dependencies**: langchain-openai ≥0.3.28, langchain-core ≥0.3.70, requests ≥2, aiohttp ≥3.9.1
11
- **License**: MIT
12
13
## Core Imports
14
15
```python
16
from langchain_xai import ChatXAI
17
```
18
19
## Basic Usage
20
21
```python
22
from langchain_xai import ChatXAI
23
24
# Initialize the chat model
25
llm = ChatXAI(
26
model="grok-4",
27
temperature=0,
28
api_key="your-xai-api-key" # or set XAI_API_KEY environment variable
29
)
30
31
# Simple chat completion
32
messages = [
33
("system", "You are a helpful assistant."),
34
("human", "What is the capital of France?")
35
]
36
response = llm.invoke(messages)
37
print(response.content)
38
39
# Streaming chat completion
40
for chunk in llm.stream(messages):
41
print(chunk.content, end="")
42
```
43
44
## Architecture
45
46
The langchain-xai package provides a single primary class `ChatXAI` that inherits from LangChain's `BaseChatOpenAI`. This design leverages the OpenAI-compatible API structure while providing xAI-specific features:
47
48
- **Standard LangChain Interface**: Supports all LangChain chat model methods (invoke, stream, batch, async operations)
49
- **xAI-Specific Features**: Tool calling, structured output, live search, reasoning content, citations
50
- **Authentication**: Environment variable or constructor-based API key management
51
- **Customization**: Configurable base URL, timeout, retry logic, and HTTP client settings
52
53
## Capabilities
54
55
### Chat Completions
56
57
Core chat completion functionality supporting both synchronous and asynchronous operations with streaming capabilities.
58
59
```python { .api }
60
class ChatXAI:
61
def __init__(
62
self,
63
model: str = "grok-4",
64
temperature: float = 1.0,
65
max_tokens: Optional[int] = None,
66
timeout: Optional[Union[float, Tuple[float, float]]] = None,
67
max_retries: int = 2,
68
api_key: Optional[str] = None,
69
xai_api_key: Optional[str] = None,
70
xai_api_base: str = "https://api.x.ai/v1/",
71
search_parameters: Optional[Dict[str, Any]] = None,
72
logprobs: Optional[bool] = None,
73
**kwargs
74
):
75
"""
76
Initialize ChatXAI model.
77
78
Parameters:
79
- model: Name of xAI model to use (default: "grok-4")
80
- temperature: Sampling temperature 0-2 (default: 1.0)
81
- max_tokens: Maximum tokens to generate
82
- timeout: Request timeout in seconds
83
- max_retries: Maximum retry attempts
84
- api_key: xAI API key (alias for xai_api_key)
85
- xai_api_key: xAI API key, reads from XAI_API_KEY env var if not provided
86
- xai_api_base: Base URL for xAI API
87
- search_parameters: Parameters for live search functionality
88
- logprobs: Whether to return log probabilities
89
"""
90
91
def invoke(self, input: LanguageModelInput, **kwargs) -> BaseMessage:
92
"""Generate chat completion for input messages."""
93
94
def stream(self, input: LanguageModelInput, **kwargs) -> Iterator[BaseMessageChunk]:
95
"""Stream chat completion chunks for input messages."""
96
97
async def ainvoke(self, input: LanguageModelInput, **kwargs) -> BaseMessage:
98
"""Async generate chat completion for input messages."""
99
100
async def astream(self, input: LanguageModelInput, **kwargs) -> AsyncIterator[BaseMessageChunk]:
101
"""Async stream chat completion chunks for input messages."""
102
103
def batch(self, inputs: List[LanguageModelInput], **kwargs) -> List[BaseMessage]:
104
"""Generate chat completions for multiple inputs."""
105
106
async def abatch(self, inputs: List[LanguageModelInput], **kwargs) -> List[BaseMessage]:
107
"""Async generate chat completions for multiple inputs."""
108
```
109
110
**Usage Example:**
111
112
```python
113
from langchain_xai import ChatXAI
114
115
# Basic setup
116
llm = ChatXAI(
117
model="grok-4",
118
temperature=0.7,
119
max_tokens=1000
120
)
121
122
# Invoke with message list
123
messages = [("human", "Explain quantum computing")]
124
response = llm.invoke(messages)
125
126
# Streaming
127
for chunk in llm.stream(messages):
128
print(chunk.content, end="")
129
130
# Async operations
131
import asyncio
132
133
async def chat_async():
134
response = await llm.ainvoke(messages)
135
return response
136
137
response = asyncio.run(chat_async())
138
```
139
140
### Tool Calling
141
142
Support for function/tool calling with parallel execution, enabling the model to call external functions and tools.
143
144
```python { .api }
145
def bind_tools(
146
self,
147
tools: Sequence[Union[Dict[str, Any], Type[BaseModel], Callable, BaseTool]],
148
**kwargs: Any
149
) -> Runnable[LanguageModelInput, BaseMessage]:
150
"""
151
Bind tools/functions to the model for tool calling.
152
153
Parameters:
154
- tools: List of tools (Pydantic models, functions, or tool definitions)
155
156
Returns:
157
Runnable that can invoke tools
158
"""
159
```
160
161
**Usage Example:**
162
163
```python
164
from pydantic import BaseModel, Field
165
from langchain_xai import ChatXAI
166
167
class GetWeather(BaseModel):
168
"""Get current weather for a location."""
169
location: str = Field(description="City and state, e.g. San Francisco, CA")
170
171
class GetPopulation(BaseModel):
172
"""Get population for a location."""
173
location: str = Field(description="City and state, e.g. San Francisco, CA")
174
175
llm = ChatXAI(model="grok-4")
176
llm_with_tools = llm.bind_tools([GetWeather, GetPopulation])
177
178
# Model will decide which tools to call
179
response = llm_with_tools.invoke("Compare weather and population of LA vs NY")
180
print(response.tool_calls)
181
182
# Control tool choice via extra_body
183
llm_no_tools = ChatXAI(
184
model="grok-4",
185
extra_body={"tool_choice": "none"}
186
)
187
188
llm_required_tools = ChatXAI(
189
model="grok-4",
190
extra_body={"tool_choice": "required"}
191
)
192
193
llm_specific_tool = ChatXAI(
194
model="grok-4",
195
extra_body={
196
"tool_choice": {
197
"type": "function",
198
"function": {"name": "GetWeather"}
199
}
200
}
201
)
202
```
203
204
### Structured Output
205
206
Generate responses conforming to specified schemas using JSON schema, JSON mode, or function calling approaches.
207
208
```python { .api }
209
def with_structured_output(
210
self,
211
schema: Optional[Union[Dict[str, Any], Type[BaseModel], Type]] = None,
212
*,
213
method: Literal["function_calling", "json_mode", "json_schema"] = "function_calling",
214
include_raw: bool = False,
215
strict: Optional[bool] = None,
216
**kwargs: Any
217
) -> Runnable[LanguageModelInput, Union[Dict, BaseModel]]:
218
"""
219
Configure model to return structured output matching schema.
220
221
Parameters:
222
- schema: Output schema (Pydantic class, TypedDict, or OpenAI tool schema)
223
- method: Steering method ("function_calling", "json_schema", "json_mode")
224
- include_raw: If True, return both raw and parsed responses
225
- strict: Whether to enforce strict schema validation
226
227
Returns:
228
Runnable that outputs structured data
229
"""
230
```
231
232
**Usage Example:**
233
234
```python
235
from typing import Optional
236
from pydantic import BaseModel, Field
237
from langchain_xai import ChatXAI
238
239
class Joke(BaseModel):
240
"""A joke with setup and punchline."""
241
setup: str = Field(description="The setup of the joke")
242
punchline: str = Field(description="The punchline of the joke")
243
rating: Optional[int] = Field(description="Funniness rating 1-10")
244
245
llm = ChatXAI(model="grok-4")
246
structured_llm = llm.with_structured_output(Joke)
247
248
joke = structured_llm.invoke("Tell me a joke about cats")
249
print(f"Setup: {joke.setup}")
250
print(f"Punchline: {joke.punchline}")
251
print(f"Rating: {joke.rating}")
252
253
# Using JSON schema method
254
json_llm = llm.with_structured_output(
255
Joke,
256
method="json_schema",
257
strict=True
258
)
259
260
# Including raw response
261
raw_llm = llm.with_structured_output(
262
Joke,
263
include_raw=True
264
)
265
result = raw_llm.invoke("Tell me a joke")
266
# result = {"raw": BaseMessage, "parsed": Joke, "parsing_error": None}
267
```
268
269
### Live Search
270
271
Enable Grok models to ground responses using web search results with configurable search parameters.
272
273
```python { .api }
274
# Configure via search_parameters in constructor
275
search_parameters: Optional[Dict[str, Any]] = {
276
"mode": str, # Search mode, e.g. "auto"
277
"max_search_results": int, # Maximum search results to use
278
"from_date": str, # Start date for search (YYYY-MM-DD)
279
"to_date": str, # End date for search (YYYY-MM-DD)
280
}
281
```
282
283
**Usage Example:**
284
285
```python
286
from langchain_xai import ChatXAI
287
288
# Configure live search
289
llm = ChatXAI(
290
model="grok-4",
291
search_parameters={
292
"mode": "auto",
293
"max_search_results": 5,
294
"from_date": "2025-01-01",
295
"to_date": "2025-01-02"
296
}
297
)
298
299
# Model will use web search to ground response
300
response = llm.invoke("What are the latest developments in AI research?")
301
print(response.content)
302
303
# Citations available in Grok 3 models
304
if hasattr(response, 'additional_kwargs') and 'citations' in response.additional_kwargs:
305
citations = response.additional_kwargs['citations']
306
print("Sources:", citations)
307
```
308
309
### Reasoning Content
310
311
Access reasoning content from supported models (Grok 3) that provide transparent reasoning processes.
312
313
**Usage Example:**
314
315
```python
316
from langchain_xai import ChatXAI
317
318
# Configure reasoning effort for Grok 3 models
319
llm = ChatXAI(
320
model="grok-3-mini",
321
extra_body={"reasoning_effort": "high"}
322
)
323
324
response = llm.invoke("Solve this logic puzzle: If all cats are animals...")
325
print("Response:", response.content)
326
327
# Access reasoning content
328
if hasattr(response, 'additional_kwargs') and 'reasoning_content' in response.additional_kwargs:
329
reasoning = response.additional_kwargs['reasoning_content']
330
print("Reasoning:", reasoning)
331
332
# Reasoning also available in streaming
333
for chunk in llm.stream("Complex math problem"):
334
if hasattr(chunk, 'additional_kwargs') and 'reasoning_content' in chunk.additional_kwargs:
335
print("Reasoning chunk:", chunk.additional_kwargs['reasoning_content'])
336
print(chunk.content, end="")
337
```
338
339
### Log Probabilities
340
341
Access token-level probability information for generated responses.
342
343
```python { .api }
344
# Enable via logprobs parameter
345
logprobs: Optional[bool] = True
346
```
347
348
**Usage Example:**
349
350
```python
351
from langchain_xai import ChatXAI
352
353
# Enable logprobs
354
llm = ChatXAI(model="grok-4", logprobs=True)
355
356
# Or bind logprobs to existing model
357
logprobs_llm = llm.bind(logprobs=True)
358
359
response = logprobs_llm.invoke([("human", "Say Hello World!")])
360
361
# Access logprobs from response metadata
362
logprobs_data = response.response_metadata.get("logprobs")
363
if logprobs_data:
364
print("Tokens:", logprobs_data["tokens"])
365
print("Token IDs:", logprobs_data["token_ids"])
366
print("Log probabilities:", logprobs_data["token_logprobs"])
367
```
368
369
## Response Metadata
370
371
All responses include comprehensive metadata about the generation process.
372
373
```python { .api }
374
response_metadata: Dict[str, Any] = {
375
"token_usage": {
376
"completion_tokens": int,
377
"prompt_tokens": int,
378
"total_tokens": int
379
},
380
"model_name": str,
381
"system_fingerprint": Optional[str],
382
"finish_reason": str, # "stop", "length", "tool_calls", etc.
383
"logprobs": Optional[Dict]
384
}
385
386
usage_metadata: Dict[str, int] = {
387
"input_tokens": int,
388
"output_tokens": int,
389
"total_tokens": int
390
}
391
```
392
393
**Usage Example:**
394
395
```python
396
from langchain_xai import ChatXAI
397
398
llm = ChatXAI(model="grok-4")
399
response = llm.invoke([("human", "Hello!")])
400
401
# Access token usage
402
print("Usage:", response.usage_metadata)
403
# {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}
404
405
# Access detailed metadata
406
print("Model:", response.response_metadata["model_name"])
407
print("Finish reason:", response.response_metadata["finish_reason"])
408
print("Token usage:", response.response_metadata["token_usage"])
409
```
410
411
## Authentication and Configuration
412
413
### Environment Variables
414
415
```python
416
# Required
417
XAI_API_KEY = "your-xai-api-key"
418
```
419
420
### Constructor Configuration
421
422
```python { .api }
423
# Primary configuration options
424
model: str = "grok-4" # Model name
425
temperature: float = 1.0 # Sampling temperature (0-2)
426
max_tokens: Optional[int] = None # Maximum tokens
427
timeout: Optional[Union[float, Tuple[float, float]]] = None # Request timeout
428
max_retries: int = 2 # Retry attempts
429
xai_api_key: Optional[str] = None # API key
430
xai_api_base: str = "https://api.x.ai/v1/" # API base URL
431
432
# Advanced options
433
search_parameters: Optional[Dict[str, Any]] = None # Live search config
434
logprobs: Optional[bool] = None # Enable log probabilities
435
default_headers: Optional[Dict] = None # Default request headers
436
default_query: Optional[Dict] = None # Default query parameters
437
http_client: Optional[Any] = None # Custom HTTP client
438
http_async_client: Optional[Any] = None # Custom async HTTP client
439
```
440
441
## Supported Models
442
443
- **grok-4**: Latest high-performance model (default)
444
- **grok-3-mini**: Smaller model with reasoning capabilities
445
- **grok-3**: Full Grok 3 model with reasoning and citations
446
- Other xAI models as available through the API
447
448
## Error Handling
449
450
```python
451
from langchain_xai import ChatXAI
452
453
try:
454
llm = ChatXAI(model="grok-4")
455
response = llm.invoke(messages)
456
except ValueError as e:
457
# API key not set or invalid parameters
458
print(f"Configuration error: {e}")
459
except Exception as e:
460
# Network, API, or other errors
461
print(f"Runtime error: {e}")
462
```
463
464
Common error scenarios:
465
- **Missing API Key**: Raises `ValueError` if `XAI_API_KEY` not set and no `api_key` provided
466
- **Invalid Parameters**: Raises `ValueError` for invalid parameter combinations (e.g., `n > 1` with streaming)
467
- **API Errors**: Network timeouts, rate limits, and API-specific errors from xAI service
468
- **Model Errors**: Invalid model names or unsupported model features
469
470
## Types
471
472
```python { .api }
473
from typing import Any, Dict, List, Optional, Union, Tuple, Iterator, AsyncIterator, Literal
474
from pydantic import BaseModel, SecretStr
475
from langchain_core.language_models.chat_models import LanguageModelInput
476
from langchain_core.messages import BaseMessage, BaseMessageChunk
477
from langchain_core.runnables import Runnable
478
479
# Message types for input
480
LanguageModelInput = Union[
481
str,
482
List[Union[str, Dict[str, Any]]],
483
List[BaseMessage]
484
]
485
486
# Response types
487
BaseMessage # Complete response message
488
BaseMessageChunk # Streaming response chunk
489
490
# Tool definition types
491
ToolDefinition = Union[
492
Dict[str, Any], # OpenAI tool schema
493
Type[BaseModel], # Pydantic model
494
Callable, # Function
495
BaseTool # LangChain tool
496
]
497
```